前言
本期和大家一起探讨了如何基于 redis 实现消息队列，其中实现方案包括三类：
- redis list：最简单粗暴的实现，存在问题包括：不支持发布/订阅模式、消费端缺少 ack 机制
- redis pub/sub：支持发布/订阅模式，有较高的丢数据风险，消费端同样不支持 ack 机制
- redis stream：趋近于成熟的 mq 实现方式. 支持发布/订阅模式，消费端能支持 ack 机制. 但是受限于 redis 自身的特性，仍无法杜绝丢失数据的可能性（本文只聊这个）
Redis Stream 简介
Redis Stream 是 Redis 5.0 版本中引入的一种数据结构，用于存储和处理消息流。它类似于消息队列，但具有更高的性能和更丰富的特性。ACK 操作用于确认消息已经被消费，从而避免重复消费。
Redis作为消息队列的优缺点
优点
- 简单轻量：Redis是一个内存中的数据存储系统，具有轻量级和简单的特点。相比较专门的消息队列系统，使用Redis作为消息队列不需要引入额外的组件和依赖，可以减少系统的复杂性。
- 速度快：由于Redis存储在内存中，它具有非常高的读写性能。这对于需要低延迟的应用程序非常有优势。
- 多种数据结构支持：Redis提供了丰富的数据结构，如列表、发布/订阅、有序集合等。这使得Redis在处理不同类型的消息和任务时更加灵活。
- 数据持久化：Redis可以通过将数据持久化到磁盘来提供数据的持久性。这意味着即使Redis重启，之前的消息也不会丢失。
- 广泛的应用场景：Redis不仅可以用作消息队列，还可以用作缓存、数据库、分布式锁等多种用途。如果你的应用程序已经使用了Redis，那么使用Redis作为消息队列可以减少技术栈的复杂性。
缺点
- 缺少一些高级特性：相对于专门的消息队列系统，Redis在消息队列方面的功能可能相对简单。例如，它可能缺乏一些高级消息传递功能，如消息重试、消息路由、持久化消息等。
- 可靠性和一致性：Redis的主要设计目标是提供高性能和低延迟，而不是强一致性和高可靠性。在某些情况下，Redis可能会丢失消息，或者在出现故障时可能无法提供持久性保证。
推荐应用场景
适用于简单的中小型项目 如果功能简单，访问量并不大可以考虑 如果你的应用程序对可靠性和高级功能有严格要求，并且需要处理大量的消息和复杂的消息路由，那么使用专门的消息队列系统可能更合适。
实战
maven依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>配置StreamConfig(监听)
import com.lasse.mq.redis.listener.AutoAckStreamConsumeListener;
import com.lasse.mq.redis.listener.BasicAckStreamConsumeListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.util.ErrorHandler;import javax.annotation.PostConstruct;
import java.time.Duration;@Configuration
public class RedisStreamConfig {@Autowiredprivate RedisTemplate redisTemplate;@Autowiredprivate AutoAckStreamConsumeListener autoAckStreamConsumeListener;@Autowiredprivate BasicAckStreamConsumeListener basicAckStreamConsumeListener;@PostConstructpublic void initializeStream() {StreamOperations<String, Object, Object> streamOperations = redisTemplate.opsForStream();// 创建一个流try {streamOperations.createGroup("stream_queue_key", AutoAckStreamConsumeListener.GROUP);streamOperations.createGroup("stream_queue_key", BasicAckStreamConsumeListener.GROUP);} catch (RedisSystemException e) {}}@Beanpublic StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(RedisConnectionFactory redisConnectionFactory) {// 用于配置消息监听容器的选项。在这个方法中，通过设置不同的选项，如轮询超时时间和消息的目标类型，可以对消息监听容器进行个性化的配置。StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小.pollTimeout(Duration.ofSeconds(3))// 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理.errorHandler(new ErrorHandler() {@Overridepublic void handleError(Throwable t) {System.out.println("出现异常就来这里了" + t);}})// 指定了消息的目标类型为 String。这意味着容器会将接收到的消息转换为 String 类型，以便在后续的处理中使用。.build();// 创建一个可用于监听Redis流的消息监听容器。StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =StreamMessageListenerContainer.create(redisConnectionFactory, options);// 方法配置了容器来接收来自特定消费者组和消费者名称的消息。它还指定了要读取消息的起始偏移量，以确定从哪里开始读取消息。// 消费组A,自动ack// 从消费组中没有分配给消费者的消息开始消费streamMessageListenerContainer.receiveAutoAck(Consumer.from(AutoAckStreamConsumeListener.GROUP, "AutoAckConsumer"),StreamOffset.create("stream_queue_key", ReadOffset.lastConsumed()), autoAckStreamConsumeListener);// 消费组B,不自动ackstreamMessageListenerContainer.receive(Consumer.from(BasicAckStreamConsumeListener.GROUP, "BasicAckConsumer"),StreamOffset.create("stream_queue_key", ReadOffset.lastConsumed()), basicAckStreamConsumeListener);streamMessageListenerContainer.start();return streamMessageListenerContainer;}}
配置消费者
自动ack
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;import java.util.Map;@Slf4j
@Component
public class AutoAckStreamConsumeListener implements StreamListener<String, MapRecord<String, String, String>> {//分组名public static final String GROUP = "autoack_stream";@Autowiredprivate RedisTemplate<String, Object> redisTemplate;@Overridepublic void onMessage(MapRecord<String, String, String> message) {String stream = message.getStream();RecordId id = message.getId();Map<String, String> map = message.getValue();log.info("[自动ACK]接收到一个消息 stream:[{}],id:[{}],value:[{}]", stream, id, map);redisTemplate.opsForStream().delete(GROUP, id.getValue());}
}手动ack
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;import java.util.Map;@Slf4j
@Component
public class BasicAckStreamConsumeListener implements StreamListener<String, MapRecord<String, String, String>> {//分组名public static final String GROUP = "basicack_stream";@Autowiredprivate RedisTemplate<String, Object> redisTemplate;@Overridepublic void onMessage(MapRecord<String, String, String> message) {String stream = message.getStream();RecordId id = message.getId();Map<String, String> map = message.getValue();log.info("[手动ACK]接收到一个消息 stream:[{}],id:[{}],value:[{}]", stream, id, map);redisTemplate.opsForStream().acknowledge(stream, GROUP, id.getValue());//消费完毕删除该条消息redisTemplate.opsForStream().delete(GROUP, id.getValue());}
}配置生产者
@Autowiredprivate RedisTemplate redisTemplate;@RequestMapping(value = "/stream/produce", method = RequestMethod.GET)public void streamProduce() {Map<String, String> map = new HashMap<>();map.put("name", "大家好我是周杰伦");map.put("time", DateUtil.now());redisTemplate.opsForStream().add("stream_queue_key", map).getValue();}实现效果
INFO 17560 --- [cTaskExecutor-1] c.l.m.r.l.AutoAckStreamConsumeListener : [自动ACK]接收到一个消息 stream:[stream_queue_key],id:[1712733195117-0],value:[{name="大家好我是周杰伦", time="2024-04-10 15:13:15"}]
INFO 17560 --- [cTaskExecutor-2] c.l.m.r.l.BasicAckStreamConsumeListener : [手动ACK]接收到一个消息 stream:[stream_queue_key],id:[1712733195117-0],value:[{name="大家好我是周杰伦", time="2024-04-10 15:13:15"}]常见问题
Redis消息队列的注意事项
一、消息丢失
问题：
当使用Redis作为消息队列时，如果Redis服务器关闭或发生故障，所有未处理的消息都将被删除，这将导致消息丢失。
解决方法：
因此，应考虑使用Redis持久化功能和备份策略，以确保消息不会丢失。
Redis提供了两种持久化选项：RDB和AOF。RDB（Redis database backup）是将整个Redis数据集在指定时间间隔内写入磁盘的快照；AOF（Append-only file）则是记录Redis服务器执行的写命令，以便在服务器重新启动时重新执行这些命令。
可以通过设置更改持久化选项的自动与手动触发，以适应您的应用程序的需求。此外，在主Redis实例故障后，备份Redis实例可以被用作备份。
二、消息可靠性
问题：
当Redis消息队列中的消息被消费者处理后，我们无法保证消费者已经成功处理了消息。如果发生故障或异常，消息将被重新处理，这可能导致问题，尤其在需要保证消息处理顺序和应用程序的幂等性时。
解决方法：
我们可以使用ACK（应答）机制，当消费者成功处理消息后，应在Redis中发送ACK。当Redis收到ACK时，将从队列中移除处理过的消息，并继续处理下一个消息。
另一种方法是使用双重消费者模式。这种模式中，每条消息都有两个消费者处理。当第一个消费者将消息标记为已完成时，Redis将消息发送给第二个消费者，如果第二个消费者没有收到消息，则原始消费者将重新处理消息。虽然这种方法会增加系统的复杂性，但可以保证消息的可靠性。
三、单点故障
问题：
使用单个Redis实例作为消息队列存在单点故障的风险。如果Redis服务器宕机，所有消息都无法处理。
解决方法：
使用Redis Sentinel，由多个Redis Sentinel实例组成的集群，可以管理和监视Redis服务器集群。当主Redis实例宕机时，Sentinel会自动将从实例提升为主实例，并通知客户端。
四、性能问题
问题：
当消息队列中的消息数量非常大时，Redis的性能可能会受到影响。在Redis中实现异步调用的最佳方法是使用发布-订阅模式，因为该模式可以处理大量的并发请求。
解决方法：
在发布-订阅模式下，发布者发布消息并将其发送到Redis频道，订阅者订阅Redis频道并处理消息。这种模式最大的优点是扩展性好，在处理大量消息的情况下可以提供可靠的性能。
Redis Stream ACK 会删除消息吗？
不会！！！
关于为什么 Redis Streams 中的消息不自动删除，这主要是基于 Stream 的设计和使用场景。以下是几个原因：
- 持久化：Streams 是为持久化消息日志而设计的。这意味着消息被保存在磁盘上，以便在 Redis 服务器重启或故障后能够恢复。自动删除消息会破坏这种持久化特性。
- 消费者进度跟踪：在发布/订阅模型中，当消息被发布后，消费者通常需要确认它们已经处理了这些消息。Streams 通过消费者组（Consumer Groups）和消费者偏移量（Consumer Offsets）来跟踪消费者的进度。如果消息被自动删除，那么跟踪消费者的进度就会变得困难。
- 手动管理：Redis 提供了 XTRIM 命令，允许用户手动删除旧的或不再需要的消息。这种手动管理的方式给了用户更多的灵活性，可以根据实际需求来决定何时删除消息。
- 资源消耗：自动删除消息可能会消耗大量的系统资源，特别是在高吞吐量的场景中。由用户根据需要手动删除消息可以更好地控制资源的使用。
需要注意的是，虽然 Streams 中的消息不会自动删除，但 Redis 会对 Streams 进行一些内部优化，例如压缩旧的消息或合并小的文件，以节省磁盘空间。这些优化操作是透明的，对用户来说是无感知的。
相关性”违规?())
)
