前言
关于 redis 我们前面已经讨论过了缓存、分布式锁、分布式唯一标识、LBS服务的用法，这里我们来谈谈利用 redis 来实现一个消息服务。
典型的消息服务是一个生产者和消费者模式的服务。一般是有生产者产生消息，将消息发送到队列中。而消息的消费者则监听消息，对消息进行处理。
有很多非常优秀的消息队列服务的产品。例如 RabbitMQ、RocketMQ、Kafka 等。这些产品都具备非常高级的功能。可靠性、扩展性都非常的好。
但是 redis 自身也能够很简单的实现消息队列的生产者和消费者模式。本文简单介绍一下在 Java 下是如何实现的。
相关命令介绍
参考 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 {@Testpublic void consumerTest() { RedisConnection redisConnection = RedisConnectionUtil.create(); new MessageConsumerRedisImpl(redisConnection, new String[]{"channel1", "channel2"}) { public void handleMessage(String message) { System.out.println(message); }
注：本文内容来自互联网，旨在为开发者提供分享、交流的平台。如有涉及文章版权等事宜，请你联系站长进行处理。