使用线程监听Redis多个频道及多个数据类型

1、说明：
使用手动创建线程的方式，对应的一个线程监听一个频道或者数据类型，而对应new 出来的线程对象放在了map中，根据key为频道名称或数据类型名称，value为线程对象。
此方式为了随时可以手动启动关闭监听对应的频道及数据类型，当然也可使用线程池来管理频道的监听，redis频道具备订阅与取消订阅的功能，所以可以通过取消订阅来做到不再监听指定频道。而数据类型则无法做到，所以需要通过手动终止线程的方式来停止对指定数据类型的监听。
2、编码
2.1、首先创建一个类，继承Thread类并且实现她的run方法

import com.ruoyi.utils.RedisConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;/**** 监听redis频道的线程类*/
public class RedisTopicThread extends Thread{private Logger logger= LoggerFactory.getLogger(RedisTopicThread.class);//频道监听器private Subscriber subscriber;//频道名称private  String channel;//redis连接配置private RedisConfig redisConfig;public RedisTopicThread(Subscriber subscriber, String channelName, RedisConfig redisConfig) {super();this.subscriber = subscriber;this.channel = channelName;this.redisConfig=redisConfig;}/**** 启动线程*/public synchronized void run() {Jedis jedis = null;try {if(this.interrupted()){    //如果当前线程处于停止状态，抛出异常throw new InterruptedException();}else{//jedis = RedisConnect.redis(ip,port,pass);   //取出一个连接jedis=redisConfig.redis();jedis.subscribe(subscriber, channel);    //通过subscribe 的api去订阅，入参是订阅者和频道名}} catch (Exception e) {//捕捉抛出的异常终止线程logger.info("终止线程！"+e.toString());} finally {if (jedis != null) {jedis.close();}}}}

2.2、创建一个类型继承JedisPubSub类，重写其onMessage(),onSubscribe(),onUnsubscribe()方法，如下

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.domain.RedisTopic;
import com.ruoyi.service.RedisTopicService;
import com.ruoyi.web.utils.MongodbClient;
import com.ruoyi.utils.RedisConfig;
import com.ruoyi.web.utils.MongodbConfig;
import com.ruoyi.web.utils.WebSocketServer;
import redis.clients.jedis.JedisPubSub;import java.util.Date;public class Subscriber extends JedisPubSub {private String subName;private RedisConfig redisConfig;private RedisTopic redisTopic;private RedisTopicService redisTopicService;private MongodbConfig mongodbConfig;public Subscriber(RedisConfig redisConfig,String subName,RedisTopic redisTopic,RedisTopicService redisTopicService,MongodbConfig mongodbConfig){this.subName = subName;this.redisConfig=redisConfig;this.redisTopic=redisTopic;this.redisTopicService=redisTopicService;this.mongodbConfig=mongodbConfig;}/*** 订阅后收到消息触发* @param channel* @param message*/public void onMessage(String channel, String message) {JSONObject jsonObject=new JSONObject();JSONObject jsonObject1=new JSONObject();try {redisTopic.setRecord(message);redisTopicService.updateTopicRecordByName(redisTopic);jsonObject.put("time",new Date().getTime());jsonObject.put("message",message);jsonObject.put("topic",channel);new MongodbClient(mongodbConfig).insertInfo(channel,jsonObject);redisConfig.redis().lpush(channel,jsonObject.toString());jsonObject1.put("type","message");jsonObject1.put("value",jsonObject);jsonObject1.put("topic",channel);WebSocketServer.sendInfo(jsonObject1.toString(),"topicData");WebSocketServer.sendInfo(jsonObject1.toString(),"main");WebSocketServer.sendInfo("topicOnMessage","redisPage");}catch (Exception e){e.printStackTrace();}}/**** 订阅频道* @param channel* @param subscribedChannels*/public void onSubscribe(String channel, int subscribedChannels) {System.out.println("已成功订阅"+channel+"频道,subscribedChannels："+subscribedChannels);}/*** 取消订阅* @param channel* @param subscribedChannels*/public void onUnsubscribe(String channel, int subscribedChannels) {System.out.println("已取消订阅"+channel+"频道,subscribedChannels："+subscribedChannels);}
}

2.3、定义两个map用于存放线程对象及频道监听器对象

    //用于存放频道监听线程对象Map<String, RedisTopicThread> map=new HashMap<>();//用于存放频道监听器对象Map<String,Subscriber>map1=new HashMap<>();

2.4、开始监听指定频道及关闭监听指定频道
2.4.1、开始监听

  /*** 开始监听指定频道* @return*/@Log(title = "Redis监听配置,开启频道监听",businessType = BusinessType.UPDATE)@RequestMapping("/startListen")@ResponseBodypublic AjaxResult listen(@RequestParam(value = "topic",required = false)String topic){Integer result = null;try {RedisTopic redisTopic=redisTopicService.findTopicInfoByName(topic);//将未被监听的频道修改为监听状态redisTopicService.updateTopicStatusByTopicName(1,topic);//注册监听器Subscriber subscriber = new Subscriber(redisConfig,topic,redisTopic,redisTopicService,mongodbConfig);//状态为0，不在监听状态，重新创建一个线程来进行监听RedisTopicThread redisTopicThread =new RedisTopicThread(subscriber,topic,redisConfig);redisTopicThread.start();//将线程放到map中map.put(topic, redisTopicThread);//将注册的监听器放到map中map1.put(topic,subscriber);//threadPoolTest.startTopicThread(subscriber,topic,redisConfig);map1.put(topic,subscriber);//将监听频道列表变化通知给前端WebSocketServer.sendInfo("listenTopic","redisPage");result=1;}catch (Exception e){result=0;e.printStackTrace();}return toAjax(result);}

2.4.2、停止监听，根据传过来的频道名称到map中找到对应的监听器对象取消其订阅，找到对应的线程对象进行终止。并将map中key为指定频道名称的数据移除掉

    /**** 停止监听告警频道* @return*/@Log(title = "Redis监听配置,停止频道监听",businessType = BusinessType.UPDATE)@RequestMapping("/stopListen")@ResponseBodypublic AjaxResult stopListen(@RequestParam(value = "topic",required = false)String topic){Integer result;try {//修改表中的指定频道的监听状态redisTopicService.updateTopicStatusByTopicName(0,topic);//根据监听的频道得到对应的线程RedisTopicThread redisTopicThread =map.get(topic);//根据频道名称获取注册的监听器Subscriber subscriber=map1.get(topic);//根据频道名称移除map中的数据map.remove(topic);//移除指定的监听器map1.remove(topic);//取消订阅指定频道subscriber.unsubscribe(topic);//将当前线程标记为终止状态redisTopicThread.interrupt();//将监听频道列表变化通知给前端WebSocketServer.sendInfo("listenTopic","redisPage");result=1;}catch (Exception e){result=0;e.printStackTrace();}return toAjax(result);}

3、效果
启动监听四个频道

往频道发送信息后，触发onMessage()方法后通过，websocket来推送监听到的数据至前端

以上为频道监听的方式

数据类型的监听方式与此类似，下面就不作复述。

使用线程监听Redis多个频道及多个数据类型相关推荐

swoole redis mysql_教你使用swoole监听redis数据
swoole如何监听redis数据? Laravel使用swoole监听redis 开始之前,请先确保redis已经正确安装,并正常运行. Laravel代码在App\Events目录下新建Redi ...
redis 能不能监听特定的key失效_php监听redis key失效触发回调事件
订单超时.活动过期解决方案:php监听redis key失效触发回调事件 Redis 的 2.8.0 版本之后可用,键空间消息(Redis Keyspace Notifications),配合 2.0 ...
springboot listener_Springboot 监听redis key的过期事件
项目中常常会碰到这样的需求,用户下订单后,30分钟未完成自动取消订单的功能. 有人说这个简单呀,写个定时任务就搞定了.除了定时任务之外,难道就没有其他的方法来实现吗?有--Redis 的键空间通知事件 ...
php 监听redis,php监听redis key失效触发回调事件_后端开发
php中使用fsockopen实现异步请求(代码示例)_后端开发 php执行一段程序,有可能几毫秒就执行完毕,也有可能耗时较长.例如,用户下单这个事件,如果调用了些第三方服务进行发邮件.短信.推送等通 ...
php 监听redis,php监听redis key失效触发回调事件
订单超时.活动过期解决方案:php监听redis key失效触发回调事件 Redis 的 2.8.0 版本之后可用,键空间消息(Redis Keyspace Notifications),配合 2.0 ...
Spring Boot 监听 Redis Key 失效事件实现定时任务
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试文章作者:彭超 https://antoniopeng.com 业务场 ...
C++开启后台线程监听控制台输入实现按任意键退出
//通过开启后台线程监听控制台输入来实现按任意键退出 //也可以自行修改成其他键 vector<thread> threads; threads.push_back(thread([]() ...
SpringBoot监听redis订阅监听和发布订阅
前言我们可以在redis中发布一条订阅到通道中,所有监听了这个通道的都可以收到这个发布的内容! redis订阅监听配置类代码如下: RedisListenerConfig.java package ...
winform 线程监听两个目录下的文件_vb.net 利用.net自带的GZipStream压缩或者解压文件的代码，不需要任何第三方控件...
网上很少有用VB写的压缩文件的代码,但是,在网络传输,文件下载,打包发布等等方面的需求又比较多,所以,借鉴了一下C#代码的例子,改造成了VB用的类.另外加上了多层文件夹压缩解压.但是,因为时间有限,只 ...

使用线程监听Redis多个频道及多个数据类型

使用线程监听Redis多个频道及多个数据类型相关推荐

最新文章

热门文章