消息队列，消息的发布订阅模式貌似是只有kafka,MQ类中间件才能实现的事情，但是别出新裁的是redis也同样具有这些功能。但是这些消息并没有持久化机制，属于即发即弃模式，也就是说它们不能像MQ中的消息那样保证持久化消息订阅者不会错过任何消息，无论这些消息订阅者是否随时在线。由于本来就是即发即弃的消息模式，所以Redis也不需要专门制定消息的备份和恢复机制。Redis也没有为发布者和订阅者准备保证消息性能的任何方案，例如在大量消息同时到达Redis服务是，如果消息订阅者来不及完成消费，就可能导致消息堆积。而MQ中有专门针对这种情况的慢消息机制。
如果能容忍这些缺点的场景下，则redis的消息队列则为不错的选择方案。这里就提供了这样的应用场景，具体应用方面不详细介绍，主要记录学习的心得和技术的使用。
一.代码设计
1.1 redis topic的一些定义
/*** redis发布定义* @author lhy* @date .12.10*/public interface RedisPubSub {/*** 订阅处理方法*/String INVOKE_METHOD = "receiveMessage";/*** 订阅频道:处理打卡记录*/String STUDENT_RECORD = "ztxc:pub";}
1.2 redis消息发布(模拟终端发布消息)
/*** redis消息发布* @author lhy* @date .12.10*/public class RdeisPublisher {private final JedisPool jedisPool;public RdeisPublisher(JedisPool jedisPool) {this.jedisPool = jedisPool;}public void sendMsg(){JSONObject json = new JSONObject();json.put("time","-12-10");json.put("vid","k123456");json.put("address","山东省济南市");json.put("carNum","123456");Jedis jedis = jedisPool.getResource();//从 mychannel 的频道上推送消息jedis.publish("ztxc:pub", json.toString());}}
1.3 消息订阅业务处理类
/*** 消息订阅业务处理类* @author lhy* @date .12.10*/@Servicepublic class BusinessSubService {private static Logger LOGGER = LoggerFactory.getLogger(BusinessSubService.class);public void receiveMessage(String message, String channel) {LOGGER.debug("################SUBSCRIBE######### message:{}, channel:{} #############", message, channel);switch (channel) {case RedisPubSub.STUDENT_RECORD:msgSubService(message);break;default:break;}}/*** 解析消息体进行业务操作* @param message*/public void msgSubService(String message){JSONObject jsonObject = JSONObject.parseObject(message);String time = (String) jsonObject.get("time");String vid =(String) jsonObject.get("vid");String address =(String) jsonObject.get("address");String cardNum =(String) jsonObject.get("carNum");//以下是业务操作LOGGER.debug("业务处理： " + jsonObject);}}
1.4 redis配置类 配置消息订阅业务处理bean
@Configurationpublic class RedisConfig {@Autowiredprivate BusinessSubService businessSubService;@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);List<PatternTopic> topicList = new ArrayList<>();topicList.add(new PatternTopic(RedisPubSub.STUDENT_RECORD));container.addMessageListener(listenerAdapter, topicList);return container;}@Beanpublic MessageListenerAdapter listenerAdapter() {return new MessageListenerAdapter(businessSubService, RedisPubSub.INVOKE_METHOD);}}
1.5 SpringBoot Task发布消息队列测试
/*** 定时任务发布消息队列*/@Configuration@EnableSchedulingpublic class RedisPubTask {@Scheduled(fixedRate=5000)private void configureTasks() {JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "127.0.0.1", 6379,0,"123456");RdeisPublisher rdeisPublisher = new RdeisPublisher(jedisPool);rdeisPublisher.sendMsg();}}
二.实现结果
-12-16 20:32:21,306 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:26,305 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:26,305 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:31,305 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:31,305 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:36,311 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:36,312 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:41,301 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:41,301 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:46,303 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:46,304 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:51,305 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:51,305 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:32:56,302 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:32:56,302 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:01,310 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:01,311 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:06,310 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:06,310 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:11,307 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:11,307 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:16,318 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:16,318 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:21,312 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:21,312 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:26,315 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:26,316 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:31,318 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:31,319 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:36,313 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:36,313 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:41,300 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:41,301 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:46,312 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:46,312 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:51,314 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:51,315 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:33:56,336 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:33:56,336 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:34:01,320 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:34:01,321 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}-12-16 20:34:06,321 DEBUG (BusinessSubService.java:20)- ################SUBSCRIBE######### message:{"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}, channel:ztxc:pub #############-12-16 20:34:06,321 DEBUG (BusinessSubService.java:42)- 业务处理： {"vid":"k123456","address":"山东省济南市","carNum":"123456","time":"-12-10"}
