详解 Redis 队列 实现

Redis 是一个高性能的键值存储系统，它的多种数据结构使其能够以不同方式实现队列，包括普通队列、延时队列和异步队列的介绍和示例。

介绍

Redis 的 List 数据结构可以用来实现普通的队列。
生产者使用 LPUSH 或 RPUSH 命令将消息添加到列表的头部或尾部，消费者使用 LPOP 或 RPOP 命令从列表的头部或尾部取出消息。

这种方式简单轻量，但缺少一些高级特性，如消息重试、持久化等 。

延时队列

延时队列可以通过 Redis 的 Sorted Set 数据结构来实现。
消息的到期时间作为分数（score），消息内容作为成员（member）。使用 ZADD 命令添加消息，并通过 ZRANGEBYSCORE 命令获取到期的消息进行消费。

这种方式可以保证消息的有序性，并且处理效率非常高 。

异步队列

Redis 的发布/订阅（pub/sub）模式可以实现异步队列。
生产者使用 PUBLISH 命令发送消息到一个频道，消费者使用 SUBSCRIBE 命令订阅频道来接收消息。

这种方式可以支持消息的广播，但消息无法持久化，且可能会出现消息丢失的情况。

Redis Stream 的使用

Redis 5.0 版本引入了 Stream 数据结构，它是一个持久化的、支持消费者组的消息队列。
使用 XADD 命令添加消息，XREAD 或 XREADGROUP 命令读取消息，并通过 XACK 命令确认消息已被处理。

这种方式支持消息的持久化、消费者组的概念以及消息确认机制 。

示例

RedisTemplate来实现队列操作

在Spring Boot中配置和使用RedisTemplate来实现队列操作主要涉及以下几个步骤：

添加依赖：

确保你的pom.xml文件中已经添加了Spring Boot的Redis依赖。

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

配置Redis：

在application.properties或application.yml中配置Redis服务器的连接信息。
application.yml

spring:redis:host: localhostport: 6379

配置RedisTemplate：

创建一个配置类，配置RedisTemplate的序列化器等。

@Configuration
public class RedisConfig {@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(factory);// 使用StringRedisSerializer序列化keytemplate.setKeySerializer(new StringRedisSerializer());// 使用Jackson2JsonRedisSerializer序列化valuetemplate.setValueSerializer(new Jackson2JsonRedisSerializer<>(Object.class));// 设置hash的key和value序列化方式template.setHashKeySerializer(new StringRedisSerializer());template.setHashValueSerializer(new Jackson2JsonRedisSerializer<>(Object.class));// 启用事务template.setEnableTransactionSupport(true);return template;}
}

使用RedisTemplate进行队列操作：

注入RedisTemplate并使用它进行队列操作。

@Service
public class RedisQueueService {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;public void enqueue(String queueName, Object value) {redisTemplate.opsForList().rightPush(queueName, value);}public Object dequeue(String queueName) {return redisTemplate.opsForList().leftPop(queueName);}// 其他队列操作...
}

这是一个基本的配置和使用RedisTemplate实现队列操作的流程。
根据你的具体需求，可能还需要配置连接池、密码认证、集群支持等高级特性。此外，对于延时队列和异步队列，可能需要使用Redisson或其他高级特性和库。

实现异步队列

在Spring Boot中，使用Redis实现异步队列通常可以通过发布/订阅模式或列表（List）数据结构来完成。以下是两种实现方式的示例：

使用发布/订阅模式实现异步队列

发布/订阅模式是一种消息通信模式，其中消息生产者（发布者）不会将消息直接发送到特定的接收者（订阅者），而是将消息发布到一个主题。

对这些消息感兴趣的接收者可以订阅这个主题，从而异步接收消息。

配置RedisTemplate：

首先，在Spring Boot应用中配置RedisTemplate，以便于操作Redis。

创建消息生产者：

使用RedisTemplate的convertAndSend方法发布消息到指定的频道。

@Service
public class MessageProducer {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void sendMessage(String channel, String message) {redisTemplate.convertAndSend(channel, message);}
}

创建消息消费者：

创建一个消息监听器，订阅频道并接收消息。

@Service
public class MessageListener {@RabbitListener(queues = "queue.name")public void receiveMessage(String message) {// 处理接收到的消息System.out.println("Received: " + message);}
}

使用列表（List）数据结构实现异步队列

列表是一种双向链表结构，可以作为队列使用。

生产者可以使用lpush或rpush将消息添加到列表的头部或尾部，消费者可以使用lpop或rpop从列表的头部或尾部取出消息。

配置RedisTemplate：

同样，首先需要配置RedisTemplate。

生产者添加消息到列表：

使用RedisTemplate的opsForList操作列表，将消息压入列表。

@Service
public class AsyncQueueProducer {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void pushMessage(String listKey, String message) {redisTemplate.opsForList().rightPush(listKey, message);}
}

消费者从列表中取出消息：

消费者可以从列表中弹出消息并处理。

@Service
public class AsyncQueueConsumer {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void consumeMessage(String listKey) {List<String> messages = redisTemplate.opsForList().range(listKey, 0, -1);for (String message : messages) {// 处理消息System.out.println("Processing: " + message);}}
}

在实际应用中，可以根据业务需求选择使用发布/订阅模式或列表数据结构来实现异步队列。
发布/订阅模式适用于消息广播的场景，而列表数据结构更适用于实现一个简单的任务队列

实现延时队列

在Spring Boot中使用Redis实现延时队列，可以通过Sorted Set数据结构来实现。以下是具体的实现步骤和示例代码：

步骤 1: 添加Redis依赖

首先，确保你的pom.xml文件中已经添加了Spring Boot的Redis依赖：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

步骤 2: 配置Redis

在application.properties或application.yml中配置Redis服务器的连接信息：

application.properties

spring.redis.host=localhost
spring.redis.port=6379

步骤 3: 创建延时队列服务

创建一个服务类来封装延时队列的逻辑：

@Service
public class DelayedQueueService {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void enqueue(String queueName, String message, long delaySeconds) {double score = System.currentTimeMillis() + delaySeconds * 1000;redisTemplate.opsForZSet().add(queueName, message, score);}public String dequeue(String queueName) {Set<String> keys = redisTemplate.opsForZSet().rangeByScore(queueName, 0, System.currentTimeMillis());if (!keys.isEmpty()) {String message = keys.iterator().next();redisTemplate.opsForZSet().remove(queueName, message);return message;}return null;}
}

步骤 4: 使用延时队列

在业务逻辑中使用DelayedQueueService来添加和消费延时消息：

@RestController
public class DelayedQueueController {@Autowiredprivate DelayedQueueService delayedQueueService;@PostMapping("/delayed-enqueue")public ResponseEntity<?> enqueueDelayedMessage(@RequestParam String message, @RequestParam long delaySeconds) {String queueName = "delayedQueue";delayedQueueService.enqueue(queueName, message, delaySeconds);return ResponseEntity.ok("Message enqueued with delay " + delaySeconds + " seconds");}@GetMapping("/delayed-dequeue")public ResponseEntity<?> dequeueDelayedMessage() {String queueName = "delayedQueue";String message = delayedQueueService.dequeue(queueName);if (message != null) {return ResponseEntity.ok(message);} else {return ResponseEntity.noContent().build();}}
}

说明

enqueue方法将消息和其预定的延迟时间戳（当前时间 + 延迟时间）作为分数（score）添加到Sorted Set中。

dequeue方法检索分数小于或等于当前时间的所有消息，从Sorted Set中移除并返回第一个消息。

定时任务（如果有）可以用于定期处理到期的消息。

这个简单的实现提供了一个基础的延时队列功能，适用于需要异步处理但具有特定延迟时间的任务。对于更高级的消息队列需求，可能需要考虑使用专业的MQ系统。

Redis Stream实现持久化队列

添加依赖：

确保pom.xml文件中已添加Spring Boot的Redis依赖。

配置Redis：

在application.properties或application.yml中配置Redis服务器的连接信息。
创建消息生产者：使用RedisTemplate的opsForStream()方法添加消息到Stream中。

创建消费者组：

使用XGROUP CREATE命令为Stream创建消费者组，可以指定从哪个消息ID开始消费。

配置消费者：

实现StreamListener接口，编写接收消息的逻辑。

配置Stream监听器容器：

使用StreamMessageListenerContainer来管理消费者，监听特定Stream的消息，并分配给消费者处理。

示例代码如下：

@Service
public class MessageProducer {@Autowiredprivate RedisTemplate<String, String> redisTemplate;public void sendMessage(String streamKey, Map<String, String> message) {redisTemplate.opsForStream().add(streamKey, message);}
}

@Service
public class MessageConsumer implements StreamListener<String, ObjectRecord<String, String>> {@Overridepublic void onMessage(ObjectRecord<String, String> message) {// 处理接收到的消息}
}

@Configuration
public class RedisStreamConfig {@Beanpublic StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer(RedisConnectionFactory connectionFactory, MessageConsumer messageConsumer) {StreamMessageListenerContainer<String, ObjectRecord<String, String>> container =new StreamMessageListenerContainer<>(connectionFactory, messageConsumer);container.start();return container;}
}

在上述代码中，MessageProducer用于发送消息到Redis
Stream，MessageConsumer实现了StreamListener接口来接收消息。
RedisStreamConfig配置了StreamMessageListenerContainer，它会启动并监听消息，将收到的消息分发给消费者处理。

注意，为了确保消息的可靠性，可以实现消息确认机制，在消息被成功处理后，通过XACK命令向Redis确认消息已被消费。同时，如果需要处理消息的持久化和回溯，Redis Stream提供了相应的命令来查询历史消息或未确认的消息 。

通过这种方式，Spring Boot应用可以利用Redis Stream构建一个高性能、持久化且支持消费者组的队列系统，适用于多种消息队列场景 。