【由浅至深】redis 实现发布订阅的几种方式

本文涉及的产品
云数据库 Redis 版,社区版 2GB
推荐场景:
搭建游戏排行榜
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
云数据库 RDS MySQL Serverless,价值2615元额度,1个月
简介: 原文:【由浅至深】redis 实现发布订阅的几种方式非常感谢依乐祝发表文章《.NET Core开发者的福音之玩转Redis的又一傻瓜式神器推荐》,对csredis作了一次完整的诠释。 前言 提到消息队列,最熟悉无疑是 rabbitmq,它基本是业界标准的解决方案。
原文: 【由浅至深】redis 实现发布订阅的几种方式

非常感谢依乐祝发表文章《.NET Core开发者的福音之玩转Redis的又一傻瓜式神器推荐》,对csredis作了一次完整的诠释。

前言

提到消息队列,最熟悉无疑是 rabbitmq,它基本是业界标准的解决方案。本文详细介绍 redis 多种实现轻订阅方法,作者认为非常有趣并加以总结,希望对有需要的朋友学习 redis 功能有一定的带入作用。

方法一:SUBSCRIBE + PUBLISH

img_be7f7ed72ff7aac23bfe6dc850e9c298.png

//程序1:使用代码实现订阅端
var sub = RedisHelper.Subscribe(("chan1", msg => Console.WriteLine(msg.Body)));
//sub.Disponse(); //停止订阅

//程序2:使用代码实现发布端
RedisHelper.Publish("chan1", "111");

优势:支持多端订阅、简单、性能高;
缺点:数据会丢失;

参考资料:http://doc.redisfans.com/pub_sub/subscribe.html

方法二:BLPOP + LPUSH(争抢)

img_8fff01d58068d75378aae3906cd057a4.png

//程序1:使用代码实现订阅端
while (running) {
    try {
        var msg = RedisHelper.BLPop(5, "list1");
        if (string.IsNullOrEmpty(msg) == false) {
            Console.WriteLine(msg);
        }
    } catch (Exception ex) {
        Console.WriteLine(ex.Message);
    }
}

//程序2:使用代码实现发布端
RedisHelper.LPush("list1", "111");

优势:数据不会丢失、简单、性能高;
缺点:不支持多端(存在资源争抢);

总结:为了解决方法一的痛点,我们实现了本方法,并且很漂亮的制造了一个新问题(不支持多端订阅)。

学习使用 BLPOP

BLPOP key [key ...] timeout

BLPOP 是列表的阻塞式(blocking)弹出原语。

它是 LPOP 命令的阻塞版本,当给定列表内没有任何元素可供弹出的时候,连接将被 BLPOP 命令阻塞,直到等待超时或发现可弹出元素为止。

当给定多个 key 参数时,按参数 key 的先后顺序依次检查各个列表,弹出第一个非空列表的头元素。

非阻塞行为

当 BLPOP 被调用时,如果给定 key 内至少有一个非空列表,那么弹出遇到的第一个非空列表的头元素,并和被弹出元素所属的列表的名字一起,组成结果返回给调用者。

当存在多个给定 key 时, BLPOP 按给定 key 参数排列的先后顺序,依次检查各个列表。

假设现在有 job 、 command 和 request 三个列表,其中 job 不存在, command 和 request 都持有非空列表。考虑以下命令:

BLPOP job command request 0

BLPOP 保证返回的元素来自 command ,因为它是按”查找 job -> 查找 command -> 查找 request “这样的顺序,第一个找到的非空列表。

redis> DEL job command request           # 确保key都被删除
(integer) 0

redis> LPUSH command "update system..."  # 为command列表增加一个值
(integer) 1

redis> LPUSH request "visit page"        # 为request列表增加一个值
(integer) 1

redis> BLPOP job command request 0       # job 列表为空,被跳过,紧接着 command 列表的第一个元素被弹出。
1) "command"                             # 弹出元素所属的列表
2) "update system..."                    # 弹出元素所属的值

阻塞行为

如果所有给定 key 都不存在或包含空列表,那么 BLPOP 命令将阻塞连接,直到等待超时,或有另一个客户端对给定 key 的任意一个执行 LPUSH 或 RPUSH 命令为止。

超时参数 timeout 接受一个以秒为单位的数字作为值。超时参数设为 0 表示阻塞时间可以无限期延长(block indefinitely) 。

redis> EXISTS job                # 确保两个 key 都不存在
(integer) 0
redis> EXISTS command
(integer) 0

redis> BLPOP job command 300     # 因为key一开始不存在,所以操作会被阻塞,直到另一客户端对 job 或者 command 列表进行 PUSH 操作。
1) "job"                         # 这里被 push 的是 job
2) "do my home work"             # 被弹出的值
(26.26s)                         # 等待的秒数

redis> BLPOP job command 5       # 等待超时的情况
(nil)
(5.66s)                          # 等待的秒数

更多学习资料:http://doc.redisfans.com/list/blpop.html

方法三:BLPOP + LPUSH(非争抢)

本方法根据方法二演变而来,设计图如下:

img_5ef76eac340531a1629d870704dc1921.png

如何实现三端订阅,都可收到消息,三端分别为 sub3, sub4, sub5:

1、sub3, sub4, sub5 使用【方法二】订阅 listkey:list1_sub3,list1_sub4,list1_sub5;

2、总订阅端订阅 listkey:list1,总订阅端收到消息后,执行 lpush list1_sub1 msg, lpush list1_sub2 msg, lpush list1_sub3 msg;

总订阅端订阅原始消息,随后将消息分发给其他订阅端,从而解决【方法二】不支持多端同时订阅的缺点。

最终实现的逻辑为:多端先争抢 list1 消息,抢到者再向其他端转发消息。

测试代码

nuget Install-Package CSRedisCore

var rds = new CSRedis.CSRedisClient("127.0.0.1:6379,password=,poolsize=50,ssl=false,writeBuffer=10240");

//sub1, sub2 争抢订阅(只可一端收到消息)
var sub1 = rds.SubscribeList("list1", msg => Console.WriteLine($"sub1 -> list1 : {msg}"));
var sub2 = rds.SubscribeList("list1", msg => Console.WriteLine($"sub2 -> list1 : {msg}"));

//sub3, sub4, sub5 非争抢订阅(多端都可收到消息)
var sub3 = rds.SubscribeListBroadcast("list2", "sub3", msg => Console.WriteLine($"sub3 -> list2 : {msg}"));
var sub4 = rds.SubscribeListBroadcast("list2", "sub4", msg => Console.WriteLine($"sub4 -> list2 : {msg}"));
var sub5 = rds.SubscribeListBroadcast("list2", "sub5", msg => Console.WriteLine($"sub5 -> list2 : {msg}"));

//sub6 是redis自带的普通订阅
var sub6 = rds.Subscribe(("chan1", msg => Console.WriteLine(msg.Body)));

Console.ReadKey();
sub1.Dispose();
sub2.Dispose();
sub3.Dispose();
sub4.Dispose();
sub5.Dispose();
sub6.Dispose();

rds.Dispose();
return;

测试功能时,发布端可以使用 redis-cli 工具。

img_b6d19c3dd38fb025a8c0374f446e4c50.png

结语

redis 功能何其多且相当好玩有趣 ,大家应尽可能多带着兴趣爱好去学习它。

若文中有不好的地方,请提出批评与改正方法,谢谢观赏。

本文使用到 CSRedisCore 的开源地址:https://github.com/2881099/csredis

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore     ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库 ECS 实例和一台目标数据库 RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
相关文章
|
3月前
|
存储 缓存 NoSQL
聊聊 Redis 的高级特性之一: 发布订阅
Redis 发布订阅 (pub/sub) 是一种消息通信模式:发送者 (pub) 发送消息,订阅者 (sub) 接收消息。 图中,消费者1和消费者2 订阅了 Redis 服务的频道 channel ,当生产者通过 PUBLISH 命令发送给频道 channel 时, 这个消息就会被发送给订阅它的两个客户端。
聊聊 Redis 的高级特性之一: 发布订阅
|
4月前
|
消息中间件 移动开发 NoSQL
Redis 协议 事务 发布订阅 异步连接
Redis 协议 事务 发布订阅 异步连接
|
4月前
|
NoSQL Redis
Redis之发布订阅
Redis之发布订阅
|
5月前
|
存储 NoSQL Redis
Redis 简介 + Python 操作发布订阅
Redis 简介 + Python 操作发布订阅
|
28天前
|
消息中间件 设计模式 缓存
Redis发布订阅:美丽的陷阱与不宜之境
Redis发布订阅:美丽的陷阱与不宜之境
64 1
|
4月前
|
消息中间件 NoSQL Java
硬核 | Redis Pub/Sub 发布订阅与宅男有什么关系?
硬核 | Redis Pub/Sub 发布订阅与宅男有什么关系?
52 0
|
4月前
|
存储 NoSQL 关系型数据库
Redis协议与异步方式(redis网络层、pipeline、事务、lua脚本、ACID特性、发布订阅、hiredis实现同步连接与异步连接)
Redis协议与异步方式(redis网络层、pipeline、事务、lua脚本、ACID特性、发布订阅、hiredis实现同步连接与异步连接)
78 0
|
4月前
|
消息中间件 NoSQL Redis
Redis发布订阅
【1月更文挑战第7天】 消息队列 Redis 发布订阅(publsub)是一种消息通信模式︰发送者(pub)发送消息,订阅者(sub)接收消息。微信、微博、关注系统! Redis客户端可以订阅任意数量的频道。
173 1
|
9月前
|
消息中间件 缓存 NoSQL
探索Redis发布订阅与消息队列:构建实时消息通信系统
本篇深入探讨了Redis的发布订阅模式和消息队列功能,展示了如何使用这两个特性构建实时消息通信系统。我们首先介绍了Redis的发布订阅模式,演示了如何通过PUBLISH命令将消息发布到特定频道,并使用SUBSCRIBE和UNSUBSCRIBE命令进行订阅和取消订阅操作。
357 0
|
5月前
|
存储 缓存 NoSQL
Redis数据库 | 发布订阅、主从复制、哨兵模式、缓存雪崩
Redis数据库 | 发布订阅、主从复制、哨兵模式、缓存雪崩
55 0