redis实现消息队列和消息订阅之Jedis的Publish/Subscribe功能

今天药忘吃喽~ 2022-05-14 20:47 232阅读 0赞

### **这个消息订阅大概是原理** ###

> 1.  redis客户端1中使用命令 ***`SUBSCRIBE talk`* **可以订阅通道 talk上的消息 
> 2.  redis客户端2中也同样运行这个命令一起订阅通道 talk 
> 3.  redis客户端3使用命令 ***`PUBLISH talk 'test'`*** 可以发现客户端1和2同时受到消息，
> 4.  redis客户端3相当于服务器，给订阅者发送消息，
> 5.  这个功能的话 相当于redis自己内置的

### 代码逻辑 ###

> 1.  通过spring链接到redis（推荐连接池）
> 2.  创建RedisSubPubListener类继承于JedisPubSub  并且实现他必要的方法
> 3.  在相应的方法里面做逻辑处理
> 4.  处理过后，再调用jedis的subscribe方法，给多个频道发订阅消息。

###  代码实现 ###

>     redis.queue.hostname=xxxxxxx
>     redis.queue.port=16379
>     redis.queue.password=test123
>     redis.queue.timeout=300000
>     redis.queue.database=0
>     redis.queue.name=report_queue

>     
>       <bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
>            <property name="maxTotal" value="${redis.pool.maxTotal}"/>
>            <property name="maxIdle" value="${redis.pool.maxIdle}"/>
>            <property name="maxWaitMillis" value="${redis.pool.maxWait}"/>
>            <property name="testOnBorrow" value="${redis.pool.testOnBorrow}"/>
>      </bean>
>     
>     <bean id="queueJedisPool" class="redis.clients.jedis.JedisPool">
>         <constructor-arg name="poolConfig" ref="jedisPoolConfig" /> 
>         <constructor-arg name="host" value="${redis.queue.hostname}" />
>         <constructor-arg name="port" value="${redis.queue.port}"/> 
>         <constructor-arg name="database" value="${redis.queue.database}"/> 
>         <constructor-arg name="timeout" value="${redis.queue.timeout}" />
>         <constructor-arg name="password" value="${redis.queue.password}"/>
>         <constructor-arg name="clientName" value="${redis.queue.name}"/>
>     </bean>
> 
> **这是我们项目的源码，里面有很多，你们使用时 可以去掉**
> 
>     public class RedisSubPubListener extends JedisPubSub {
>         private JedisUtil jedisUtil;
>         private VenderService venderService;
>         private VenderAisleService venderAisleService;
>         private VenderFaultService venderFaultService;
>         private SnService snService;
>         RedisSubPubListener(JedisUtil jedisUtil,VenderService venderService,VenderAisleService venderAisleService,VenderFaultService venderFaultService,SnService snService){
>             this.jedisUtil = jedisUtil;
>             this.venderService = venderService;
>             this.venderAisleService = venderAisleService;
>             this.venderFaultService = venderFaultService;
>             this.snService = snService;
>         }
>         // 取得订阅的消息后的处理(业务需求，在这里做逻辑处理)
>         public void onMessage(String channel, String message) {
>             //TODO:接收订阅频道消息后，业务处理逻辑
>             System.out.println("通道："+channel + " 数据：" + message);
>             JSONObject jsonData = JSON.parseObject(message);
>             String type = jsonData.getString("type");
>             List<Integer> serialNumber = new ArrayList();
>           /*  String sn = null;*/
>             deviceType(jsonData, type, serialNumber);
>         }
>     
>         private void deviceType(JSONObject jsonData, String type, List<Integer> serialNumber) {
>             String sn = null;
>             switch (type){
>                 case "status":
>                     JSONObject _data = jsonData.getJSONObject("data");
>                     //数据转为实体
>                     MessQueue messQueue =  _data.toJavaObject(MessQueue.class);
>                     messQueue.setIsOnLine(true);
>                     //获取到当前售货机的编号
>                     sn = messQueue.getSn();
>                     //redis里面的json字符串
>                     String messQueueStr = jedisUtil.HASH.hget("messQueue",sn);
>                     //判断redis是否已存在，存在的话  作更新操作，不存在作添加操作
>                     if(messQueueStr==null){
>                         findMessQueueRedis(serialNumber, sn, _data, messQueue);
>                     }else{
>                         editMessQueueStr(serialNumber, sn, _data, messQueue, messQueueStr);
>                     }
>                     jedisUtil.HASH.hset("messQueue",sn,jsonData.getJSONObject("data").toJSONString());
>                     break;
>                 case "offline":
>                     sn = jsonData.getString("sn");
>                     //去数据库把对应售货机标为离线
>                     Vender _vender = new Vender();
>                     _vender.setSn(sn);
>                     _vender.setIsAbnormal(0);
>                     venderService.editEntity(_vender);
>                     //把redis中的对应售货机数据标为离线便于下次收到状态改回到在线
>                     JSONObject _dataJson = jsonData.getJSONObject("data");
>                     //数据转为实体
>                     MessQueue _messQueue =  _dataJson.toJavaObject(MessQueue.class);
>                     _messQueue.setIsOnLine(true);
>                     VenderFault venderFault = new VenderFault();
>                     venderFault.setVenderId(_vender.getId());
>                     venderFault.setType(1);
>                     try {
>                         venderFault.setSn(snService.generate(Type.faultNumber));
>                     }catch (Exception e){
>                         e.printStackTrace();
>                     }
>                     venderFault.setDescription("网络离线");
>                     venderFault.setOrders(1L);
>                     venderFaultService.addEntity(venderFault);
>                     jedisUtil.HASH.hset("messQueue",sn,JSON.toJSONString(_messQueue));
>                     break;
>                 case "gps":
>                     sn = jsonData.getString("sn");
>                     //获取实时经纬度
>                     String longitude = jsonData.getString("longitude");//经度
>                     String latitude = jsonData.getString("latitude"); //维度
>                     Vender _vender_ = new Vender();
>                     _vender_.setSn(sn);
>                     _vender_.setLatitude(latitude);
>                     _vender_.setLongitude(longitude);
>                     //更新数据库
>                     venderService.editEntity(_vender_);
>                         break;
>                 case "collisionWarning":
>                     sn = jsonData.getString("sn");
>                     Vender vender_ = new Vender();
>                     vender_.setSn(sn);
>                     venderService.editEntity(vender_);
>                     VenderFault _venderFault = new VenderFault();
>                     _venderFault.setVenderId(vender_.getId());
>                     _venderFault.setType(3);
>                     try {
>                         _venderFault.setSn(snService.generate(Type.faultNumber));
>                     }catch (Exception e){
>                         e.printStackTrace();
>                     }
>                     _venderFault.setDescription("碰撞报警");
>                     _venderFault.setOrders(1L);
>                     venderFaultService.addEntity(_venderFault);
>                     break;
>             }
>         }
>     
>         private void findMessQueueRedis(List<Integer> serialNumber, String sn, JSONObject _data, MessQueue messQueue) {
>             //添加操作
>             jedisUtil.HASH.hset("messQueue",sn,_data.toJSONString());
>             //更新到数据库对应sn的机器
>             Vender vender = new Vender();
>             vender.setSn(sn);
>             //时间戳   分别是年月日时分秒
>             Integer[] timestamp = messQueue.getTimestamp();
>             //时间字符串
>             StringBuffer timestampStr = new StringBuffer();
>             timestampStr.append(timestamp[0]+"-");
>             timestampStr.append(timestamp[1]+"-");
>             timestampStr.append(timestamp[2]+" ");
>             timestampStr.append(timestamp[3]+":");
>             timestampStr.append(timestamp[4]+":");
>             timestampStr.append(timestamp[5]);
>             String _timestampStr = timestampStr.toString();
>             SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
>             try{
>                 vender.setModifyDate(sdf.parse(_timestampStr));
>             }catch (Exception e){
>                 e.printStackTrace();
>             }
>             vender.setNumber( messQueue.getCargoRoadCounts().longValue());
>             vender.setNetworkStrength(messQueue.getNetworkSignalStrength());
>             vender.setType(String.valueOf(messQueue.getType()));
>             //修改信息
>             venderService.editEntity(vender);
>             //获取货道状态集合
>             Integer[] status = messQueue.getStatus();
>             Vender _vender = venderService.findBySn(sn);
>             if(_vender!=null){
>                 for (int i=0;i<status.length;i++){
>                     //当前为异常货道
>                     if(status[i]==1){
>                         serialNumber.add(i);
>                     }
>                 }
>                 venderAisleService.editError(_vender.getId(),serialNumber);
>             }
>         }
>     
>         private void editMessQueueStr(List<Integer> serialNumber, String sn, JSONObject _data, MessQueue messQueue, String messQueueStr) {
>             //当数据不一样的时候去作更新的操作
>             if(!_data.equals(messQueueStr)){
>                 MessQueue _messQueue  =  JSON.parseObject(messQueueStr,MessQueue.class);
>                 try{
>                     //返回
>                     Map<String, Object> result = messQueue.compare(_messQueue);
>                     //存放异常货道的序列号
>                     Boolean isFaultFlag = true ;
>                     Vender vender = new Vender();
>                     VenderAisle venderAisle = new VenderAisle();
>                     Vender _vender = venderService.findBySn(sn);
>                     venderAisle.setVenderId(_vender.getId());
>                     vender.setSn(sn);
>                     vender.setId(_vender.getId());
>                     vender.setIsAbnormal(0);
>                     Set<String> _keySet = result.keySet();
>                     for(String key : _keySet){
>                         Object value =  result.get(key);
>                         switch (key){
>                             case "cargoRoadCounts":
>                                 vender.setNumber((Long) result.get("cargoRoadCounts"));
>                                 break;
>                             case "networkStrength":
>                                 vender.setNetworkStrength((Integer) result.get("networkStrength"));
>                                 break;
>                             case "modifyDate":
>                                 vender.setModifyDate((Date) result.get("modifyDate"));
>                                 break;
>                             case "type":
>                                 vender.setType(result.get("type").toString());
>                                 break;
>                             case "temperature1":
>                                 vender.setNowTemperature(result.get("temperature1").toString());
>                                 break;
>                             case "temperature2":
>                                 vender.setSetTemperature(result.get("temperature2").toString());
>                                 break;
>                             default:
>                                 //货道是否异常
>                                 if(key.startsWith("status_")){
>                                    String[] _key =  key.split("_");
>                                    //_key[1]为货道的序列号   isFault为是否故障   0为正常 1为故障
>                                    Integer isFault = (Integer) result.get(key);
>                                    if(isFault == 1){
>                                       isFaultFlag = false;
>                                       serialNumber.add(Integer.parseInt(_key[1]));
>                                     }
>                                 }
>                                 break;
>                         }
>                     }
>                     //将售货机标记为故障
>                     if(isFaultFlag==false){
>                         vender.setIsFault(1);
>                         vender.setNums(_vender.getNums()+1);
>                     }
>                     //去修改信息 除了货道异常。
>                     venderService.editEntity(vender);
>                     //修改货道的是否正常
>                     venderAisleService.editError(_vender.getId(),serialNumber);
>                     for(Integer index: serialNumber){
>                         //添加故障记录
>                         VenderFault venderFault = new VenderFault();
>                         venderFault.setVenderId(_vender.getId());
>                         venderFault.setType(2);
>                         venderFault.setSn(snService.generate(Type.faultNumber));
>                         venderFault.setDescription(_vender.getName()+"的第"+index+"货道出现故障");
>                         venderFault.setOrders(1L);
>                         venderFaultService.addEntity(venderFault);
>                     }
>                 }catch (Exception e){
>                 }
>             }
>         }
>         
>         // 初始化订阅时候的处理
>         public void onSubscribe(String channel, int subscribedChannels) {
>             // System.out.println(channel + "=" + subscribedChannels);
>         }
>     
>         // 取消订阅时候的处理
>         public void onUnsubscribe(String channel, int subscribedChannels) {
>             // System.out.println(channel + "=" + subscribedChannels);
>         }
>     
>         // 初始化按表达式的方式订阅时候的处理
>         public void onPSubscribe(String pattern, int subscribedChannels) {
>             // System.out.println(pattern + "=" + subscribedChannels);
>         }
>     
>         // 取消按表达式的方式订阅时候的处理
>         public void onPUnsubscribe(String pattern, int subscribedChannels) {
>             // System.out.println(pattern + "=" + subscribedChannels);
>         }
>     
>         // 取得按表达式的方式订阅的消息后的处理
>         public void onPMessage(String pattern, String channel, String message) {
>             System.out.println(pattern + "=" + channel + "=" + message);
>         }
>     }

### **调用subscribe方法** ###

>     public class QueueJedisUtil {
>        @Autowired
>        private JedisPool queueJedisPool;
>        @Value("${redis.queue.name}")
>        private String queueName;
>        @Resource(name = "jedisUtil")
>        private JedisUtil jedisUtil;
>        @Resource(name = "venderServiceImpl")
>        private VenderService venderService;
>        @Resource(name = "venderAisleServiceImpl")
>        private VenderAisleService venderAisleService;
>        @Resource(name = "venderFaultServiceImpl")
>        private VenderFaultService VenderFaultService;
>        @Resource(name = "snServiceImpl")
>        private SnService snService;
>     
>        public void queueSub() {
>           final RedisSubPubListener listener = new RedisSubPubListener(jedisUtil,venderService,venderAisleService,VenderFaultService,snService);
>           Jedis jedis = queueJedisPool.getResource();
>           jedis.subscribe(listener, queueName);
>        }
>     }

### **在项目启动的时候调用** ###

>     public class InitListener implements ServletContextAware, ApplicationListener<ContextRefreshedEvent> {
>     
>        @SuppressWarnings("unused")
>        private ServletContext servletContext;
>     
>        @Value("${system.name}")
>        private String systemName;
>        @Value("${system.version}")
>        private String systemVersion;
>        
>        @Autowired
>        private QueueJedisUtil queueJedisUtil;
>        
>        public void setServletContext(ServletContext servletContext) {
>           this.servletContext = servletContext;//消息订阅线程不要放在这里，有可能会占满cpu，导致初始化完成不了
>     
>        }
>     
>     
>        public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
>           new startQueueThread().start();
>        }
>     
>        /**
>         * 启动消息订阅线程
>         * @author lhy
>         *
>         */
>        public class startQueueThread extends Thread {
>           public void run() {
>              queueJedisUtil.queueSub();
>           }
>        }
>     }
> 
> //我们的业务需求时每一分钟就会有消息订阅，所以具体的你要根据自己的需求来搞