在 Java 中使用 redis 的消息队列服务

梦里梦外; 2023-09-23 23:33 55阅读 0赞

### 前言 ###

关于 redis 我们前面已经讨论过了缓存、分布式锁、分布式唯一标识、LBS服务的用法，这里我们来谈谈利用 redis 来实现一个消息服务。

典型的消息服务是一个生产者和消费者模式的服务。一般是有生产者产生消息，将消息发送到队列中。而消息的消费者则监听消息，对消息进行处理。

有很多非常优秀的消息队列服务的产品。例如 RabbitMQ、RocketMQ、Kafka 等。这些产品都具备非常高级的功能。可靠性、扩展性都非常的好。

但是 redis 自身也能够很简单的实现消息队列的生产者和消费者模式。本文简单介绍一下在 Java 下是如何实现的。

### 相关命令介绍 ###

参考 [redis 官网][redis] redis 和 pubsub 模式相关的命令如下。

*  PUBLISH channel message ：发布一条消息到指定的频道
 *  SUBSCRIBE channel \[channel ...\] ：订阅指定频道的消息。频道可以是多个
 *  PSUBSCRIBE pattern \[pattern ...\] ：订阅符合给定模式频道的消息
 *  UNSUBSCRIBE \[channel \[channel ...\]\] ：停止订阅指定频道的消息
 *  PUNSUBSCRIBE \[pattern \[pattern ...\]\] ：停止订阅指定模式频道的消息
 *  PUBSUB subcommand \[argument \[argument ...\]\] ： 这个没搞清

这里面我们主要会用到前三个命令来实现一个基于 redis 的消息队列服务。

### Jedis 相关接口 ###

发送消息

public Long publish(final String channel, final String message)
    复制代码

在这里，我们会用这个接口来向消息队列指定的频道发送消息，在 Java里面的核心代码如下：

jedis = redisConnection.getJedis();
    jedis.publish(channel, message);
    复制代码

订阅消息

public void subscribe(final JedisPubSub jedisPubSub, final String... channels)
    复制代码

在这里，我们用这个接口来订阅指定频道的消息，并且通过实现的 JedisPubSub 对象的回调接口来处理收到的消息。改逻辑核心代码如下所示

Jedis jedis = redisConnection.getJedis();
    jedis.subscribe(new JedisPubSub() {
        @Override
        public void onMessage(String channel, String message) {
            //处理消息逻辑代码
         }
    复制代码

\}, channels);

### 消息发布完整实现 ###

定义接口

package com.x9710.common.redis;
    
    /**
     * 消息生产者服务接口
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public interface MessagePublisher {
        boolean sendMessage(String message);
    }
    复制代码

实现接口

package com.x9710.common.redis.impl;
    
    import com.x9710.common.redis.MessagePublisher;
    import com.x9710.common.redis.RedisConnection;
    import redis.clients.jedis.Jedis;
    
    /**
     * 消息生产者服务 redis 实现类
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public class MessagePublisherRedisImpl implements MessagePublisher {
    private RedisConnection redisConnection;
    
    /**
     * 可以同时向多个频道发送数据
     */
    private String[] channels;
    
    public void setRedisConnection(RedisConnection redisConnection) {
        this.redisConnection = redisConnection;
    }
    
    public void setChannels(String[] channels) {
        this.channels = channels;
    }
    
    public boolean sendMessage(String message) {
        Jedis jedis = null;
        try {
            if (channels != null && channels.length > 0) {
                jedis = redisConnection.getJedis();
                for (String channel : channels) {
                    jedis.publish(channel, message);
                }
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return false;
    }
    }
    复制代码

### 消息消费完整实现 ###

定义接口

package com.x9710.common.redis;
    
    /**
     * 消息消费者服务接口
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public interface MessageConsumer {
        void handleMessage(String message);
    }
    复制代码

实现接口

package com.x9710.common.redis.impl;
    
    import com.x9710.common.redis.MessageConsumer;
    import com.x9710.common.redis.RedisConnection;
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPubSub;
    
    /**
     * 消息消费者服务 redis 实现类
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public abstract class MessageConsumerRedisImpl implements MessageConsumer {
    
    /**
     *
     * @param redisConnection redis 连接类
     * @param channels 订阅的频道列表
     */
    public MessageConsumerRedisImpl(RedisConnection redisConnection, String[] channels) {
        Jedis jedis = null;
        try {
            if (channels != null && channels.length > 0) {
                jedis = redisConnection.getJedis();
                jedis.subscribe(new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        System.out.println("receive " + message + " from " + channel);
                        handleMessage(message);
                    }
                }, channels);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }
    }
    复制代码

这里定义了一个抽象类，目的是让用户继承该类来实现 handleMessage 方法，在该方法中实现收到消息后的业务逻辑。

### 测试 ###

实现消息订阅测试

package com.x9710.common.redis.test;
    
    import com.x9710.common.redis.RedisConnection;
    import com.x9710.common.redis.impl.MessageConsumerRedisImpl;
    import org.junit.Test;
    
    /**
     * 消息消费服务测试类
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public class RedisConsumerTest {
    
    @Test
    public void consumerTest() {
        RedisConnection redisConnection = RedisConnectionUtil.create();
        new MessageConsumerRedisImpl(redisConnection, new String[]{"channel1", "channel2"}) {
            public void handleMessage(String message) {
                System.out.println(message);
            }
        };
    }
    
    }
    复制代码

实现消息发送测试

package com.x9710.common.redis.test;
    
    import com.x9710.common.redis.RedisConnection;
    import com.x9710.common.redis.impl.MessagePublisherRedisImpl;
    import org.junit.Assert;
    import org.junit.Before;
    import org.junit.Test;
    
    /**
     * 消息发送服务测试类
     *
     * @author 杨高超
     * @since 2018-01-03
     */
    public class RedisPublisherTest {
        private MessagePublisherRedisImpl messagePublisherRedis;
    
    @Before
    public void before() {
        RedisConnection redisConnection = RedisConnectionUtil.create();
        messagePublisherRedis = new MessagePublisherRedisImpl();
        messagePublisherRedis.setRedisConnection(redisConnection);
        messagePublisherRedis.setChannels(new String[]{"channel1", "channel2"});
    }
    
    @Test
    public void publisherTest() {
        for (int i = 0; i < 3; i++) {
            Assert.assertTrue(messagePublisherRedis.sendMessage("message" + i));
        }
    }
    }
    复制代码

运行测试

首先运行消息消费测试用来订阅消息，然后运行消息发布测试用例来发布消息。测试结果如下

![3104ce1b476ff3e65b1cf82d62eecbf6.png][]

### 后记 ###

这个程序实在前面的文章[《在 Java 中使用 redis》][Java _ redis]的基础上添加新的实现类的方式完成的。代码同步发布在 [GitHub 仓库][GitHub]中

**原文发表在简书中，[原始链接][Link 1]**

作者：杨高超  
原文链接：https://juejin.cn/post/6844903544672223240

[redis]: https://link.juejin.cn?target=https%3A%2F%2Fredis.io%2Fcommands
[3104ce1b476ff3e65b1cf82d62eecbf6.png]: https://img-blog.csdnimg.cn/img_convert/3104ce1b476ff3e65b1cf82d62eecbf6.png
[Java _ redis]: https://link.juejin.cn?target=https%3A%2F%2Fwww.jianshu.com%2Fp%2Fe030632b19d3
[GitHub]: https://link.juejin.cn?target=https%3A%2F%2Fgithub.com%2Fgaochao2000%2Fredis_util
[Link 1]: https://link.juejin.cn?target=