一、为什么选择Redis实现消息队列？
Redis凭借其内存级操作（微秒级响应）、丰富的数据结构以及持久化能力，成为构建高性能消息队列的热门选择。相比传统消息队列（如Kafka/RabbitMQ），Redis在以下场景表现突出：
• 轻量级任务调度：毫秒级任务分发
• 实时数据处理：日志采集、事件驱动架构
• 高并发队列：电商秒杀、API限流
• 实时广播：即时通知、实时数据推送
二、主流实现方案对比
方案对比维度
| 特性 | List结构队列 | Stream类型队列 | Sorted Set队列 | Pub/Sub |
|---|---|---|---|---|
| 消息持久化 | ❌（依赖Redis配置，配置RDB或AOF后可以持久化） | ✔️（内置持久化） | ❌（依赖Redis配置） | ❌（纯内存） |
| 消息广播 | ❌ | ❌ | ❌ | ✔️（一对多） |
| 离线消息 | ❌ | ✔️（存储未ACK消息） | ❌ | ❌（立即丢弃） |
| 订阅模式 | ❌ | ❌ | ❌ | ✔️（频道/模式匹配） |
| 典型延迟 | 0.8ms | 1.2ms | 2.7ms | 0.2ms |
| 适用场景 | 任务队列 | 可靠消息处理 | 定时任务 | 实时通知 |
三、核心方案实现详解
方案1：List结构队列（简单队列）
核心原理
// 生产者
jedis.lpush("task_queue", taskJson);
// 消费者（阻塞模式）
List<String> result = jedis.brpop(0, "task_queue");
String task = result.get(1);
持久化机制
• RDB持久化：定时生成内存快照（需配置save参数）
• AOF持久化：记录所有写操作命令（需配置appendonly yes）
• 验证方法：
# 查看当前持久化配置
CONFIG GET save
CONFIG GET appendonly
特性分析
• 优点：实现简单，性能极高（TPS 10万+）
• 缺点：无ACK机制，持久化依赖Redis配置
• 适用场景：日志采集、非关键任务队列
方案2：Stream类型队列（企业级队列）
核心原理
// 生产者
String messageId = jedis.xadd("order_stream", "*",
"status", "created",
"amount", "99.9");
// 消费者组消费
Map.Entry<String, String> entry = jedis.xreadGroup(
"order_group",
"consumer1",
XReadGroupParams.xReadGroupParams().count(1).streamOffset("order_stream", ">"),
"order_stream"
).get(0);
// 确认消息
jedis.xack("order_stream", "order_group", entry.getKey());
核心优势
• 消费者组：支持多消费者并行处理
• 消息确认：ACK机制保证消息不丢失
• 消息回溯：可查看历史消息（7天默认）
方案3：Sorted Set延迟队列
核心原理
// 投递延迟任务（延迟30分钟）
long delaySeconds = 1800;
jedis.zadd("delay_queue",
System.currentTimeMillis() + delaySeconds*1000,
taskJson);
// 轮询处理
Set<String> tasks = jedis.zrangeByScore(
"delay_queue",
0,
System.currentTimeMillis()
);
应用场景
• 订单超时处理
• 支付回调重试
• 定时任务调度
方案4：Pub/Sub实时消息系统
核心原理
// 发布者
jedis.publish("stock_updates",
JSON.toJSONString(stockData));
// 订阅者
JedisPubSub subscriber = new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
handleRealTimeUpdate(message);
}
};
jedis.subscribe(subscriber, "stock_updates");
核心特性
• 广播模式：一对多实时消息推送
• 模式匹配：支持通配符订阅（如news.*）
• 低延迟：微秒级消息传递
四、Java实战代码示例
4.1 List队列完整实现
public class ListQueue {
private static final String KEY = "list_queue";
private Jedis jedis;
public ListQueue() {
this.jedis = new Jedis("localhost", 6379);
}
// 生产者
public void produce(String task) {
jedis.lpush(KEY, task);
}
// 消费者（阻塞模式）
public String consume() {
while (true) {
List<String> result = jedis.brpop(0, KEY);
if (result != null && !result.isEmpty()) {
return result.get(1);
}
}
}
}
4.2 Stream队列消费者组
public class StreamQueue {
private static final String STREAM_KEY = "stream_queue";
private static final String GROUP_NAME = "order_group";
private Jedis jedis;
public StreamQueue() {
this.jedis = new Jedis("localhost", 6379);
createConsumerGroup();
}
private void createConsumerGroup() {
try {
jedis.xgroupCreate(STREAM_KEY, GROUP_NAME, "0");
} catch (Exception e) {
// 组已存在
}
}
// 消费者处理
public void processMessages() {
while (true) {
Map.Entry<String, String> entry = jedis.xreadGroup(
GROUP_NAME,
"consumer1",
XReadGroupParams.xReadGroupParams().count(1).streamOffset(STREAM_KEY, ">"),
STREAM_KEY
).get(0);
String msgId = entry.getKey();
Map<String, String> fields = EntryToMap(entry.getValue());
processTask(fields);
jedis.xack(STREAM_KEY, GROUP_NAME, msgId);
}
}
private Map<String, String> EntryToMap(String value) {
// 解析Stream消息格式
return Arrays.stream(value.split(","))
.map(entry -> entry.split("="))
.collect(Collectors.toMap(a -> a[0], a -> a[1]));
}
}
4.3 Pub/Sub实时通知
public class PubSubDemo {
public static void main(String[] args) {
// 发布者线程
new Thread(() -> {
try (Jedis jedis = new Jedis("localhost")) {
for (int i = 0; i < 1000; i++) {
jedis.publish("realtime_alerts",
String.format("{\"event\":\"alert\",\"id\":%d}", i));
Thread.sleep(100);
}
}
}).start();
// 订阅者线程
new Thread(() -> {
Jedis jedis = new Jedis("localhost");
jedis.subscribe(new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
System.out.printf("[实时通知] %s: %s%n", channel, message);
}
}, "realtime_alerts");
}).start();
}
}
五、性能测试对比
测试环境
• 硬件：4核8G CentOS 7.9
• Redis版本：6.2.6（混合持久化）
• 客户端：Jedis 4.2.3
• 并发量：500线程
测试结果（单位：TPS）
| 方案 | 吞吐量 | 平均延迟 | CPU占用 | 消息可靠性 |
|---|---|---|---|---|
| List队列（无持久化） | 122,300 | 0.8ms | 38% | ❌（重启丢失） |
| List队列（AOF） | 98,500 | 1.5ms | 45% | ✔️（AOF每秒同步） |
| Stream队列 | 85,600 | 1.2ms | 45% | ✔️（ACK机制） |
| Sorted Set队列 | 38,400 | 2.7ms | 29% | ✔️（定时轮询） |
| Pub/Sub | 182,450 | 0.4ms | 32% | ❌（离线丢失） |
六、生产环境配置建议
- List队列持久化配置
# Redis.conf 配置示例
save 900 1 # 900秒内至少1次修改触发保存
save 300 10 # 300秒内至少10次修改
save 60 10000 # 60秒内至少10000次修改
appendonly yes
appendfsync everysec # 每秒同步（性能与安全平衡）
- 混合持久化方案
// 关键业务数据双写保障
jedis.lpush("critical_task", taskJson); // 写List
jedis.xadd("critical_stream", "*", "data", taskJson); // 写Stream
七、选型决策树
八、关键注意事项
- List队列持久化陷阱
• 大Key风险：单List超过1GB会显著降低性能
• 持久化阻塞：AOF重写期间可能延迟飙升
• 解决方案：
// 拆分大List为多个子List
String listKey = "task_list_" + (taskId % 10);
jedis.lpush(listKey, taskJson);
- Stream消息过期策略
# 自动清理旧消息（保留最近1000条）
XTRIM order_stream MAXLEN ~ 1000
通过本文的完整分析，开发者可以明确：
• List队列的持久化能力完全依赖Redis服务端配置，需显式启用AOF/RDB
• Stream队列是唯一内置可靠持久化的方案，适合核心业务场景
• Pub/Sub仅适用于实时广播场景，需配合其他方案实现消息持久化
生产环境建议采用混合架构：
• 用Pub/Sub处理实时通知
• 用Stream处理关键业务数据
• 用List处理高吞吐量日志（需配置持久化）
• 用Sorted Set处理定时任务
