排版设计的网站,广州市天河区建设局网站,长沙3合1网站建设,网站建设方案 filetype doc#x1f525;博客主页#xff1a; 【小扳_-CSDN博客】 ❤感谢大家点赞#x1f44d;收藏⭐评论✍ 文章目录 1.0 消息队列的认识 2.0 基于 List 实现消息队列 2.1 基于 List 实现消息队列的优缺点 3.0 基于 PubSub 实现消息队列 3.1 基于 PubSub 的消息队列优缺点 4.0 基于 St… 博客主页 【小扳_-CSDN博客】 ❤感谢大家点赞收藏⭐评论✍ 文章目录 1.0 消息队列的认识 2.0 基于 List 实现消息队列 2.1 基于 List 实现消息队列的优缺点 3.0 基于 PubSub 实现消息队列 3.1 基于 PubSub 的消息队列优缺点 4.0 基于 Stream 实现消息队列 4.1 Stream 的单消费模式 4.2 Stream 的消费组模式 1.0 消息队列的认识 消息队列Message Queue字面意思就是存放消息的队列。最简单的消息队列模型包含 3 个角色 1消息队列存储和管理消息也被称为消息代理Message Broker 2生产者发送消息到消息队列。 3消费者从消息队列获取消息并处理消息。 2.0 基于 List 实现消息队列 Redis 的 list 数据结构是一个双向链表很容易模拟出队列效果。
实现思路 队列时入口和出口不在一边因此我们可以利用LPUSH 结合 RPOP 或者 RPUSH 结合 LPOP 来实现。不过需要注意的是当队列中没有消息时 RPOP 或 LPOP 操作会直接返回 null 并不像 JVM 的阻塞队列那样会阻塞并等待消息因此这里应该使用 BRPOP 或者 BLPOP 来实现阻塞效果。
代码演示 当数据要进入队列时那么可以使用 LPUSH KEY VALUE 命令KEY 为队列名称VALUE 为数据值将数据写入 Redis 中。当要获取数据的时使用 BRPOP KEY TIMEOUT命令KEY 为队列名称TIMEOUT 为最大阻塞时间在最大阻塞时间内仍旧没有获取数据则返回 null 。该命令主要做了两步将数据移除队列中并将该数据返回。 2.1 基于 List 实现消息队列的优缺点 优点 1利用 Redis 存储不受限于 JVM 内存上限。 2基于 Redis 的持久化机制数据安全性有保证。 3可以满足消息有序性。 缺点 1无法避免消息丢失。 当在 BRPOP 获取数据的时候出现异常返回数据失败从而导致数据丢失。因为数据已经从队列中移除出来了所以队列中已经不存在之前的数据了。 2只支持单消费者。 当一个消费者来消费之后其他再来的消费者就不能再获取到第一个消费者的数据所以说数据只能给一个消费者。 3.0 基于 PubSub 实现消息队列 PubSub 是 Redis2.0 版本引入的消息传递模型。顾名思义消费者可以订阅一个或多个 channel 生产者向对应 channel 发送消息后所有订阅者都能收到相关消息。
常用的命令 1SUBSCRIBE channel [channel]订阅一个或者多个频道。 2PUBLISH channel msg向一个频道发送消息。 3PSUBSCRIBE pattern [pattern]订阅与 pattern 格式匹配的所有频道。* 代表通配符订阅所有频道。 这就实现了支持多个消费者获取到相同的消息。当消息被发布了那么已经订阅该频道的消费者就可以及时获取到消息了。
代码演示 先订阅频道 发送消息 当生产者发送完消息消费者就会收到通知从通道中获取到消息。 3.1 基于 PubSub 的消息队列优缺点 优点 1采用发布订阅模型支持多生产、多消费。 解决了基于 List 实现的消息队列的缺点单消费。 缺点 1不支持数据持久化。 将消息发布出去之后不会进行数据保存。不管有无消费者订阅都会将消息直接发布出去。 2无法避免消息丢失。 因为不支持持久化当消息丢失之后无法再找到原本的数据。 3消息堆积有上限超出时数据丢失。 在消费者中接收到的数据会暂时存放起来一旦超过存放的大小就会导致数据丢失。 4.0 基于 Stream 实现消息队列 Stream 是 Redis 5.0 引入的一种新数据类型可以实现一个功能非常完善的消息队列。
发送消息的命令 XADD key ID field string [field string ...] key 为队列名称*|ID 为消息的唯一 id* 代表由 Redis 自动生成。格式是“时间戳-递增数字”例如 1644804662707-0。field value 代表发送到队列中的消息称为 Entry 。格式就是多个 key-value 键值对。
代码演示 4.1 Stream 的单消费模式
单消费者获取数据的命名 XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...] COUNT count 为每次读取消息的最大数量BLOCK milliseconds 代表当没有消息时是否阻塞阻塞时长STREAMS key 代表要从哪个队列读取消息key 就是队列名ID 代表起始 id 只返回大于该 ID 的消息0 为从第一个消息开始而 $ 为从最新的消息开始。
代码演示 当 ID 使用 $ 时不会从原本 s 中直接获取原本的数据而是在 2 秒内有无最新的数据添加进来如果有则返回该数据如果没有则返回 null。 当 ID 使用 0 时则从原本 s 中直接获取原本的数据。 在业务开发中我们可以循环的调用 XREAD 阻塞方式来查询最新消息从而实现持续监听队列的效果伪代码如下 需要注意的地方 当使用 Stream 单消费者模式的时候我们指定起始 ID 为 $ 时代表读取最新的消息如果我们处理一条消息的过程中又有超过 1 条以上的消息到达队列则下次获取时也只能获取到最新的一条会出现漏读消息的问题。 XREAD 命令特点 1消息可回溯。 2一个消息可以被多个消费者读取。 3可以阻塞读取。 4有消息漏读的风险。 4.2 Stream 的消费组模式 将多个消费者划分到一个组中监听同一个队列。
特点 1消息分流 队列中的消息会分流给组内的不同消费者而不是重复消费从而加快消息处理的速度。 2消息标示 消费者组会维护一个标示记录最后一个被处理的消息哪怕消费者宕机重启还会从标示之后读取消息。确保每一个消息都会被消费。 3消息确认 消费者获取消息后消息处于 pending 状态并存入一个 pending-list 。当处理完成后需要通过 XACK 来确认消息标记消息为已处理才会从 pending-list 移除。 创建消费者组 XGROUP CREATE key groupname id|$ [MKSTREAM] key 代表队列名称groupName 代表消费者组名称ID 起始 ID 标示$ 代表队列中最后一个消息0 则代表队列中第一个消息。MKSTREAMS 代表不存在时自动创建队列。 如果之前列表的数据要继续获取则 ID 选为 0如果之前的列表中的数据不需要了则 ID 选为 $ 。 从消费者组读取消息 XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...] group 代表组名consumer 代表组内消费者名称count 代表每次读取的最大数量milliseconds 代表当没有消息时最长的等待时间NOACK 代表无需手动 ACK获取消息后自动确认。key 代表指定队列名称 ID 代表获取消息的起始 ID 当 ID 为 从下一个未消费的消息开始。 当 ID 为其他根据指定 id 从 pending-list 中获取已消费但未确认的消息例如 0是从 pending-list 中的第一个消息开始。 确认消息 XACK key groupName ID key 为队列名groupName 为组名ID 为消息唯一 id 。 查看未确认的消息 XPENDING key group [start end count] [consumer] key 为队列名group 为组名start 起始地址count 个数consumer 组内消费者名称。 消费者监听消息思路 Java 代码实现从消息队列中获取消息 import cn.hutool.core.bean.BeanUtil;
import com.project.volunteermanagementproject.pojo.StreamObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import java.time.Duration;
import java.util.List;
import java.util.Map;
Component
public class StreamUtil {AutowiredStringRedisTemplate stringRedisTemplate;//实现从消息队列中获取消息public void getStream(){while (true){try {ListMapRecordString, Object, Object read stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),StreamOffset.create(s1, ReadOffset.lastConsumed()));if (read null || read.isEmpty()){//如果获取失败说明没有消息继续下一次循环continue;}//解析消息中的消息MapRecordString, Object, Object entries read.get(0);MapObject, Object value entries.getValue();StreamObject streamObject BeanUtil.fillBeanWithMap(value, new StreamObject(), true);//这就拿到了消息队列中的数据了,就可以去使用该对象了System.out.println(streamObject);//这就需要确认消息队列stringRedisTemplate.opsForStream().acknowledge(s1, g1, entries.getId());} catch (Exception e) {//如果在获取消息过程中出现异常则需要再次执行该消息任务while (true){try {ListMapRecordString, Object, Object read stringRedisTemplate.opsForStream().read(Consumer.from(g1, c1),StreamReadOptions.empty().count(1),StreamOffset.create(s1, ReadOffset.from(0)));if (read null || read.isEmpty()){break;}MapRecordString, Object, Object entries read.get(0);MapObject, Object value entries.getValue();StreamObject streamObject BeanUtil.fillBeanWithMap(value, new StreamObject(), true);//重新拿到未确认的数据System.out.println(streamObject);//再次进行消息确认Long acknowledge stringRedisTemplate.opsForStream().acknowledge(s1, g1, entries.getId());} catch (Exception ex) {throw new RuntimeException(ex);}}}}}
} XREADGROUP 命令特点 1消息可回溯 2可以多消费者争抢消息加快消费速度 3可以阻塞读取 4没有消息漏读的风险 5有消息确认机制保证消息至少被消费一次
