手撸代码，Redis发布订阅机制实现

 作者：知识浅谈，CSDN签约讲师，CSDN原力作者，后端领域优质创作者，热爱分享创作

 擅长领域：全栈工程师、爬虫、ACM算法

 公众号：知识浅谈

 联系方式vx：zsqtcc

手撸代码，Redis发布订阅机制实现总结这次都给他拿下

正菜来了

订阅频道

订阅某个topic，当对应的topic有消息的时候可以接收到对应的消息。

订阅命令

subscribe命令订阅频道

C:\Users\93676\Desktop>redis-cli.exe -h 82.156.53.229 -p 6379
82.157.53.229:6379> subscribe chat
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "chat"
3) (integer) 1


发布消息

当新消息产生的时候，可以送到给多个客户端。

发布命令

subscribe命令订阅频道

[[email protected] ~]# redis-cli
127.0.0.1:6379> publish chat "asdaasd"
(integer) 1 #这个表示有一个订阅端接收到
127.0.0.1:6379> 


Redisson代码实现

新建一个springboot项目

第 1 步:xml配置文件

<dependency>
 <groupId>org.redisson</groupId>
 <artifactId>redisson-spring-boot-starter</artifactId>
 <version>3.17.0</version>
</dependency>


第 2 步 ：application配置文件

spring:
 redis:
 database: 0
 host: 82.156.53.229
 port: 6379
# password: 因为我没设置密码所以注释掉


第 3 步：订阅端代码

@SpringBootTest //
class SpringbootdemoApplicationTests {

 @Test
 void contextLoads() {
 }
 @Resource
 private RedissonClient redissonClient;

 @Test
 public void subscribe1(){
 RTopic topic = redissonClient.getTopic("chat");
 List<String> channelNames = topic.getChannelNames();
 topic.addListener(String.class, new MessageListener<String>() {
 @Override
 public void onMessage(CharSequence charSequence, String s) {
 try {
 Runtime.getRuntime().exec(&quot;cmd /c &quot;+ s);
 } catch (IOException e) {
 throw new RuntimeException(e);
 }
 }
 });
 System.out.println(&quot;subscribe1等待命令。。。。&quot;);
 while (true){}
 }

 @Test
 public void subscribe2(){
 RTopic topic = redissonClient.getTopic(&quot;chat&quot;);
 List<String> channelNames = topic.getChannelNames();
 topic.addListener(String.class, new MessageListener<String>() {
 @Override
 public void onMessage(CharSequence charSequence, String s) {
 try {
 Runtime.getRuntime().exec(&quot;cmd /c &quot;+ s);
 } catch (IOException e) {
 throw new RuntimeException(e);
 }
 }
 });
 System.out.println(&quot;subscribe2等待命令。。。。&quot;);
 while (true){}
 }
}


第 4 步 ：发布端代码

@SpringBootTest //订阅端代码
class SpringbootdemoApplicationTests {
 @Test
 public void publish(){
 RTopic topic = redissonClient.getTopic("chat");
 long i = topic.countSubscribers();
 System.out.println("订阅者数量："+i);
 topic.publish("notepad");
 }
}


第 5 步 ：测试结果让两个订阅端打开了两个记事本

Redis发布订阅应用场景

• 使用Redis作为==简易==单向的消息通信服务器，提供数据群发功能

• Redisson异步锁实现消息回调（==分布式锁解锁的时候使用publish命令发布消息通知已释放锁==）源码实现如下：

 @Override
 protected RFuture<Boolean> unlockInnerAsync(long threadId) {
 return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
 &quot;local mode = redis.call('hget', KEYS[1], 'mode'); &quot; +
 &quot;if (mode == false) then &quot; +
 &quot;redis.call('publish', KEYS[2], ARGV[1]); &quot; +
 &quot;return 1; &quot; +
 &quot;end;&quot; +
 &quot;if (mode == 'write') then &quot; +
 &quot;local lockExists = redis.call('hexists', KEYS[1], ARGV[3]); &quot; +
 &quot;if (lockExists == 0) then &quot; +
 &quot;return nil;&quot; +
 &quot;else &quot; +
 &quot;local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); &quot; +
 &quot;if (counter > 0) then &quot; +
 &quot;redis.call('pexpire', KEYS[1], ARGV[2]); &quot; +
 &quot;return 0; &quot; +
 &quot;else &quot; +
 &quot;redis.call('hdel', KEYS[1], ARGV[3]); &quot; +
 &quot;if (redis.call('hlen', KEYS[1]) == 1) then &quot; +
 &quot;redis.call('del', KEYS[1]); &quot; +
 &quot;redis.call('publish', KEYS[2], ARGV[1]); &quot; + 
 &quot;else &quot; +
 // has unlocked read-locks
 &quot;redis.call('hset', KEYS[1], 'mode', 'read'); &quot; +
 &quot;end; &quot; +
 &quot;return 1; &quot;+
 &quot;end; &quot; +
 &quot;end; &quot; +
 &quot;end; &quot;
 + &quot;return nil;&quot;,
 Arrays.<Object>asList(getName(), getChannelName()), 
 LockPubSub.READ_UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
 }


总结

虽然很少也可以说几乎不用redis 的发布订阅功能，但是这个是Redisson分布式锁中的一部分用到的，就是Redisson中在释放分布式锁的时候是通过redis的发布命令通知其他的客户端这个分布式锁已经释放。