关于 redis 我们前面已经讨论过了缓存、分布式锁、分布式唯一标识、LBS 服务的用法,这里我们来谈谈利用 redis 来实现一个消息服务。
典型的消息服务是一个生产者和消费者模式的服务。一般是有生产者产生消息,将消息发送到队列中。而消息的消费者则监听消息,对消息进行处理。
有很多非常优秀的消息队列服务的产品。例如 RabbitMQ、RocketMQ、Kafka 等。这些产品都具备非常高级的功能。可靠性、扩展性都非常的好。
但是 redis 自身也能够很简单的实现消息队列的生产者和消费者模式。本文简单介绍一下在 Java 下是如何实现的。
参考 redis 官网 redis 和 pubsub 模式相关的命令如下。
这里面我们主要会用到前三个命令来实现一个基于 redis 的消息队列服务。
- 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));
- }
- }
- }
首先运行消息消费测试用来订阅消息,然后运行消息发布测试用例来发布消息。测试结果如下
这个程序实在前面的文章 《在 Java 中使用 redis》 的基础上添加新的实现类的方式完成的。代码同步发布在 GitHub 仓库 中
原文发表在简书中,原始链接来源: https://juejin.im/post/5a4da028f265da4311209fb7