一、从"排队买奶茶"说起：为什么Redis列表适合做消息队列
想象一下你去奶茶店排队买饮料的场景。每个人进去之后都在排队，先到的人先被服务，后到的人排在后面等着。这个场景其实就是一个非常典型的"先进先出"模型，也就是我们常说的FIFO（First In, First Out）。
而Redis的列表（List）数据结构，天生就具备这种能力。你可以把列表想象成一个可以两头操作的队列：你可以从左边塞东西进去，也可以从右边塞进去；同样，你可以从左边取出来，也可以从右边取出来。这种灵活性让它在消息队列的场景中非常好用。
很多开发者第一次接触消息队列，可能觉得它是一个很复杂的东西，需要专门的中间件比如RabbitMQ、Kafka。但实际上，对于一些中小规模的消息处理场景，Redis列表完全可以胜任，而且部署简单、性能优秀。
二、Redis列表的基本操作：push和pop就像往盒子里放东西和取东西
Redis列表其实就是一个链表，支持从两端进行push和pop操作。下面我们把常用的命令理一遍。
2.1 往列表两端塞东西
往列表左边塞东西用LPUSH，往右边塞东西用RPUSH。这和你在生活中整理收纳盒的操作很像——你可以从上盖放进去，也可以从底下抽出来。
# 往my_queue这个列表的左边放入一个任务
LPUSH my_queue "任务A"
# 往my_queue这个列表的右边放入一个任务
RPUSH my_queue "任务B"
# 查看所有任务，0表示起始位置，-1表示最后一个
LRANGE my_queue 0 -1
# 返回: ["任务A", "任务B"]
2.2 从列表两端取东西
取东西的时候，如果从左边取用LPOP，从右边取用RPOP。这里有个关键的组合：如果你用RPUSH来放消息，用LPOP来取消息，那就是标准的先进先出（FIFO）队列。
# 从右边放入新消息，模拟生产者连续推送
RPUSH task_queue "消息1"
RPUSH task_queue "消息2"
RPUSH task_queue "消息3"
# 从左边取出最早的消息（先放进去的先被取出）
LPOP task_queue
# 返回: "消息1"
LPOP task_queue
# 返回: "消息2"
2.3 阻塞式获取：没有消息时不着急
在实际的消息队列场景中，消费者不会一直空转着检查有没有新消息。Redis提供了阻塞式的BLPOP和BRPOP命令，意思是：如果没有消息，就先挂着等，一旦有新消息就立刻返回。这就像你去取快递，如果快递员还没到，你就不用一直盯着门口，等快递到了自然会有人叫你。
# 阻塞式从左边取，最多等30秒，超时返回nil
BLPOP task_queue 30
# 也可以同时监听多个队列，哪个有消息就先取哪个
BRPOP task_queue urgent_queue 10
三、用Redis列表搭建一个简单的消息队列
下面我们用Python来演示一个完整的生产者-消费者模型。整个示例统一使用Python技术栈，依赖redis-py库。
3.1 安装依赖
pip install redis
3.2 生产者代码
生产者负责往队列里塞消息。
# 技术栈: Python + redis-py
# 文件: producer.py
# 功能: 消息生产者，负责往Redis列表中推送消息
import redis
import json
import time
import uuid
# 连接Redis，使用本地默认配置
r = redis.Redis(host='localhost', port=6379, db=0)
# 队列名称，全局统一
QUEUE_NAME = 'task_queue'
def produce_message(task_type, payload):
"""
生产一条消息并推入队列
task_type: 任务类型，比如'发邮件'、'发短信'
payload: 任务的具体内容
"""
message = {
'id': str(uuid.uuid4()), # 消息唯一ID
'type': task_type, # 任务类型
'payload': payload, # 任务数据
'timestamp': time.time() # 创建时间戳
}
# 使用RPUSH从右侧推入消息，保证FIFO顺序
r.rpush(QUEUE_NAME, json.dumps(message))
print(f"[生产者] 消息已入队: id={message['id']}, type={task_type}")
return message
# 模拟生产多条消息
if __name__ == '__main__':
# 批量生产10条不同类型的任务
for i in range(10):
task_types = ['发邮件', '发短信', '生成报表', '数据同步']
task_type = task_types[i % len(task_types)]
payload = {'order_id': f'ORD{1000 + i}', 'user': f'user_{i}'}
produce_message(task_type, payload)
time.sleep(0.5) # 每隔0.5秒生产一条
# 查看队列中消息总数
queue_size = r.llen(QUEUE_NAME)
print(f"[生产者] 当前队列中消息总数: {queue_size}")
3.3 消费者代码
消费者从队列中取消息并处理。
# 技术栈: Python + redis-py
# 文件: consumer.py
# 功能: 消息消费者，负责从Redis列表中取出并处理消息
import redis
import json
import time
import logging
# 配置日志格式
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
# 连接Redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 队列名称，必须和生产者保持一致
QUEUE_NAME = 'task_queue'
# 阻塞超时时间，单位秒
BLOCK_TIMEOUT = 5
def process_message(message):
"""
处理单条消息
这里模拟不同的任务类型执行不同的逻辑
"""
task_type = message['type']
payload = message['payload']
if task_type == '发邮件':
# 模拟发邮件，耗时2秒
time.sleep(2)
logger.info(f"邮件已发送: to={payload.get('user')}")
elif task_type == '发短信':
# 模拟发短信，耗时1秒
time.sleep(1)
logger.info(f"短信已发送: to={payload.get('user')}")
elif task_type == '生成报表':
# 模拟生成报表，耗时3秒
time.sleep(3)
logger.info(f"报表已生成: order_id={payload.get('order_id')}")
elif task_type == '数据同步':
# 模拟数据同步，耗时1.5秒
time.sleep(1.5)
logger.info(f"数据已同步: order_id={payload.get('order_id')}")
else:
logger.warning(f"未知任务类型: {task_type}")
def consume_loop():
"""
消费循环：不断从队列中取消息并处理
使用BLPOP实现阻塞式消费，避免空转浪费CPU
"""
logger.info(f"消费者启动，监听队列: {QUEUE_NAME}")
while True:
# BLPOP: 从左边阻塞式弹出
# 参数1: 队列名称
# 参数2: 超时时间(秒)
result = r.blpop(QUEUE_NAME, timeout=BLOCK_TIMEOUT)
# 如果超时了没有消息，result为None
if result is None:
logger.debug("暂时没有新消息，继续等待...")
continue
# result是一个元组: (队列名, 消息内容)
queue_name, message_json = result
message = json.loads(message_json)
logger.info(f"消费消息: id={message['id']}, type={message['type']}")
try:
# 执行具体的业务逻辑
process_message(message)
logger.info(f"消息处理成功: id={message['id']}")
except Exception as e:
# 处理失败时记录错误
logger.error(f"消息处理失败: id={message['id']}, error={e}")
if __name__ == '__main__':
try:
consume_loop()
except KeyboardInterrupt:
logger.info("消费者被手动停止")
3.4 多消费者并行消费
实际生产环境中，往往需要多个消费者并行消费，提高处理效率。Redis列表天然支持多个消费者同时从同一队列取消息。
# 技术栈: Python + redis-py
# 文件: multi_consumer.py
# 功能: 多消费者并行消费示例
import redis
import json
import time
import threading
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(threadName)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
QUEUE_NAME = 'task_queue'
def worker(name):
"""
每个工作线程独立消费消息
线程名会体现在日志中，方便排查问题
"""
logger.info(f"工作线程 {name} 启动")
processed_count = 0
while True:
# 每个线程独立调用blpop，不会互相干扰
result = r.blpop(QUEUE_NAME, timeout=10)
if result is None:
continue
queue_name, message_json = result
message = json.loads(message_json)
processed_count += 1
# 模拟处理耗时
time.sleep(1)
logger.info(
f"线程{name} 处理完成 第{processed_count}条: "
f"type={message['type']}"
)
if __name__ == '__main__':
# 创建3个消费者线程
threads = []
for i in range(3):
t = threading.Thread(
target=worker,
args=(f'worker_{i}',),
name=f'Worker-{i}'
)
t.daemon = True # 设置为守护线程，主线程退出时自动结束
t.start()
threads.append(t)
# 主线程保持运行，防止进程退出
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("所有消费者已停止")
四、消息队列场景中的核心挑战
用一个Redis列表来做消息队列，跑起来确实很快，但一旦放到生产环境，就会遇到一些实际问题。下面这些是我们在实践中踩过坑的地方，也是大家在使用时最容易忽略的。
4.1 消息丢失问题
最典型的场景是：消费者取出了消息，但是在处理过程中程序崩了或者网络断了，这条消息就永远消失了。因为在Redis列表模型里，LPOP是"取走"操作，消息一旦被弹出就不再存在于队列中。
这个问题就像你去银行取钱，ATM机吐了钱出来，但你在钱到手之前突然停电了，这笔钱就"蒸发"了——银行不知道你已经取走了，但你手里也没有。
4.2 消息重复问题
如果我们为了防止消息丢失而做了重试机制，那又会引入新问题：一条消息可能被重复处理多次。比如消费者取到了消息，处理成功了，但是在给服务端回复"已处理"之前断网了。下次重连后，它会再次取到同样的消息。
在消息队列的术语里，这叫"至少一次投递"（at-least-once delivery）。要解决重复问题，通常需要在业务层面做幂等处理。
4.3 消息积压问题
当生产者的速度远快于消费者的处理速度时，队列会越堆越长。Redis列表是存储在内存中的，如果消息积压严重，内存会迅速被占满，最终导致Redis无法响应甚至崩溃。
五、优化实践：让消息队列更可靠
知道了问题在哪里，接下来我们来看看怎么解决。下面这些都是我们在实际项目中验证过的方案。
5.1 用两个队列实现可靠消费
思路很简单：用一个"就绪队列"和一个"处理中队列"。消费者取消息时，先从就绪队列取，同时把消息放到处理中队列。处理成功后从处理中队列移除。如果消费者挂了，可以通过检查处理中队列来找回丢失的消息。
# 技术栈: Python + redis-py
# 文件: reliable_consumer.py
# 功能: 基于双队列的可靠消息消费实现
import redis
import json
import time
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
# 就绪队列：存放待消费的消息
READY_QUEUE = 'task_queue:ready'
# 处理中队列：存放已经被取走但还没处理完的消息
PROCESSING_QUEUE = 'task_queue:processing'
# 消息超时时间，单位秒。超过这个时间还没处理完，就认为消费者挂了
MESSAGE_TIMEOUT = 60
def acquire_message():
"""
原子性地从就绪队列获取消息并标记为处理中
使用Lua脚本保证原子性，避免两个消费者同时取到同一条消息
"""
lua_script = """
-- 从就绪队列右边弹出消息，同时放入处理中队列的左边
local msg = redis.call('RPOPLPUSH', KEYS[1], KEYS[2])
if msg then
-- 获取消息ID
local message = cjson.decode(msg)
-- 记录处理开始时间，方便后续超时检测
redis.call('HSET', 'processing_info', message.id, tostring(tonumber(ARGV[1])))
end
return msg
"""
# 执行Lua脚本
# KEYS[1] = 就绪队列
# KEYS[2] = 处理中队列
# ARGV[1] = 当前时间戳
result = r.eval(
lua_script,
2,
READY_QUEUE,
PROCESSING_QUEUE,
str(time.time())
)
return result if result else None
def commit_message(message_id):
"""
消息处理成功后，从处理中队列中移除
表示这条消息已经被可靠地消费了
"""
# 删除处理信息
r.hdel('processing_info', message_id)
def recover_stale_messages():
"""
恢复超时的消息：把处理中超时的消息放回就绪队列
这个函数应该由一个定时任务定期调用
"""
now = time.time()
info = r.hgetall('processing_info')
for msg_id, start_time_str in info.items():
start_time = float(start_time_str)
# 如果处理时间超过了超时阈值，就认为消费者挂了
if now - start_time > MESSAGE_TIMEOUT:
# 从处理中队列中找到对应的消息
processing_msgs = r.lrange(PROCESSING_QUEUE, 0, -1)
for msg_json in processing_msgs:
message = json.loads(msg_json)
if message['id'] == msg_id:
# 把消息放回就绪队列（用LPUSH放到左边，保持顺序）
r.lpush(READY_QUEUE, msg_json)
# 从处理中队列移除
r.lrem(PROCESSING_QUEUE, 0, msg_json)
# 删除处理信息
r.hdel('processing_info', msg_id)
logger.warning(
f"恢复超时消息: id={msg_id}, "
f"已处理时间={int(now - start_time)}秒"
)
break
return len(info)
def safe_consume_loop():
"""
可靠的消费循环
每条消息都会被可靠地处理，不会丢失也不会无限重试
"""
logger.info("可靠消费者启动")
while True:
# 原子性地获取消息
result = acquire_message()
if result is None:
# 没有消息，短暂休眠后重试
time.sleep(1)
continue
message = json.loads(result)
logger.info(f"开始处理: id={message['id']}, type={message['type']}")
try:
# 模拟业务处理
time.sleep(2)
logger.info(f"业务处理完成: id={message['id']}")
# 处理成功，标记为完成
commit_message(message['id'])
logger.info(f"消息已确认: id={message['id']}")
except Exception as e:
# 处理失败，不做任何操作
# 消息仍然留在处理中队列
# 等超时后会被自动恢复
logger.error(f"处理失败，等待超时恢复: id={message['id']}, error={e}")
if __name__ == '__main__':
try:
safe_consume_loop()
except KeyboardInterrupt:
logger.info("可靠消费者已停止")
5.2 消息去重：用Redis的Set做幂等
为了防止消息被重复处理，我们可以用Redis的Set结构来记录已经处理过的消息ID。
# 技术栈: Python + redis-py
# 文件: dedup_consumer.py
# 功能: 带幂等检查的消息消费者
import redis
import json
import time
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
QUEUE_NAME = 'task_queue'
# 已处理消息ID的集合，用于去重
PROCESSED_SET = 'processed_messages'
# 去重记录的过期时间，单位秒（保留7天的记录）
DEDUP_EXPIRE_TIME = 7 * 24 * 3600
def is_duplicate(message_id):
"""
检查消息是否已经被处理过
使用Redis的SADD命令的原子性来同时检查和标记
SADD返回1表示新增成功（即第一次处理），返回0表示已存在（即重复）
"""
# 添加消息ID到去重集合
added = r.sadd(PROCESSED_SET, message_id)
# 设置整个集合的过期时间（简化处理，实际可以更精细）
r.expire(PROCESSED_SET, DEDUP_EXPIRE_TIME)
# added=1 表示是新消息，added=0 表示已处理过
return added == 0
def consume_with_dedup():
"""
带幂等检查的消费循环
"""
logger.info("去重消费者启动")
while True:
result = r.blpop(QUEUE_NAME, timeout=5)
if result is None:
continue
queue_name, message_json = result
message = json.loads(message_json)
# 幂等检查
if is_duplicate(message['id']):
logger.warning(f"消息重复，跳过处理: id={message['id']}")
continue
# 执行业务逻辑
logger.info(f"处理消息: id={message['id']}, type={message['type']}")
time.sleep(1)
logger.info(f"处理完成: id={message['id']}")
if __name__ == '__main__':
try:
consume_with_dedup()
except KeyboardInterrupt:
logger.info("去重消费者已停止")
5.3 限制队列长度：防止内存溢出
生产者往队列里塞消息的速度如果不加控制，队列可能会无限增长。我们需要在入队时设置最大长度限制。
# 技术栈: Python + redis-py
# 文件: bounded_producer.py
# 功能: 带长度限制的队列生产者
import redis
import json
import time
import logging
import threading
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
QUEUE_NAME = 'task_queue'
# 队列最大长度，超过这个值就拒绝入队
MAX_QUEUE_LENGTH = 10000
def safe_produce(message_json):
"""
安全地往队列里添加消息
如果队列已满，返回False表示入队失败
"""
# 先检查队列当前长度
current_length = r.llen(QUEUE_NAME)
if current_length >= MAX_QUEUE_LENGTH:
logger.warning(
f"队列已满: 当前长度={current_length}, "
f"最大长度={MAX_QUEUE_LENGTH}, 拒绝入队"
)
return False
# 推入消息
r.rpush(QUEUE_NAME, message_json)
return True
# 更好的方案：用Lua脚本保证检查和推入的原子性
lua_enqueue = """
local queue_key = KEYS[1]
local max_length = tonumber(ARGV[1])
local message = ARGV[2]
-- 获取当前队列长度
local current_length = redis.call('LLEN', queue_key)
if current_length >= max_length then
-- 队列已满
return 0
end
-- 推入消息
redis.call('RPUSH', queue_key, message)
return 1
"""
def atomic_produce(message_json):
"""
原子性地检查队列长度并推入消息
使用Lua脚本避免并发问题
"""
result = r.eval(lua_enqueue, 1, QUEUE_NAME, MAX_QUEUE_LENGTH, message_json)
return result == 1
if __name__ == '__main__':
# 模拟高并发场景下生产消息
def producer_thread(thread_id):
for i in range(100):
msg = json.dumps({
'id': f'thread_{thread_id}_msg_{i}',
'type': '测试任务',
'payload': {'thread': thread_id, 'index': i}
})
success = atomic_produce(msg)
if not success:
logger.warning(f"线程{thread_id} 第{i}条消息入队失败")
break
time.sleep(0.01)
# 启动5个生产者线程
threads = []
for tid in range(5):
t = threading.Thread(target=producer_thread, args=(tid,))
t.start()
threads.append(t)
for t in threads:
t.join()
# 打印最终队列长度
final_length = r.llen(QUEUE_NAME)
logger.info(f"所有生产者完成后，队列长度: {final_length}")
5.4 延迟消息：用ZSet实现延迟队列
有时候我们需要消息延迟一段时间再被消费，比如用户下单后30分钟未支付就取消订单。Redis的列表本身不支持延迟，但可以结合ZSet（有序集合）来实现。
# 技术栈: Python + redis-py
# 文件: delayed_queue.py
# 功能: 基于ZSet的延迟消息队列实现
import redis
import json
import time
import threading
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
# 延迟消息存储在ZSet中，score是到期时间戳
DELAYED_ZSET = 'delayed_queue'
# 到期后移入的普通队列
READY_QUEUE = 'task_queue'
def schedule_delayed_message(message, delay_seconds):
"""
将消息放入延迟队列
message: 消息字典
delay_seconds: 延迟多少秒后消息到期
"""
message_json = json.dumps(message)
# 计算到期时间戳（秒级）
expire_at = time.time() + delay_seconds
# 用ZADD把消息放入有序集合
# score = 到期时间戳
# member = 消息JSON字符串
r.zadd(DELAYED_ZSET, {message_json: expire_at})
logger.info(
f"延迟消息已调度: id={message['id']}, "
f"延迟={delay_seconds}秒, 到期时间={expire_at}"
)
def promote_delayed_messages():
"""
将到期的延迟消息移入就绪队列
这个函数需要定时调用，比如每秒调用一次
"""
now = time.time()
# 查找所有score小于等于当前时间的消息
# 即所有已经到期的消息
expired_messages = r.zrangebyscore(
DELAYED_ZSET,
0,
now,
withscores=True
)
for msg_json, score in expired_messages:
# 使用Lua脚本保证原子性：先从ZSet删除，再放入队列
lua_promote = """
local msg = ARGV[1]
local score = tonumber(ARGV[2])
-- 检查是否还在延迟队列中
local member_score = redis.call('ZSCORE', KEYS[1], msg)
if member_score == false then
-- 已经被其他进程处理了，跳过
return 0
end
-- 从延迟队列中移除
redis.call('ZREM', KEYS[1], msg)
-- 放入就绪队列
redis.call('RPUSH', KEYS[2], msg)
return 1
"""
result = r.eval(lua_promote, 2, DELAYED_ZSET, READY_QUEUE, msg_json, str(score))
if result == 1:
msg = json.loads(msg_json)
logger.info(f"延迟消息已激活: id={msg['id']}")
return len(expired_messages)
def delay_queue_worker():
"""
延迟队列工作线程
定期检查并激活到期的延迟消息
"""
logger.info("延迟队列工作线程启动")
while True:
promoted = promote_delayed_messages()
if promoted > 0:
logger.info(f"本轮激活了{promoted}条延迟消息")
time.sleep(1) # 每秒检查一次
if __name__ == '__main__':
# 启动延迟队列工作线程
worker = threading.Thread(target=delay_queue_worker, daemon=True)
worker.start()
# 模拟调度几条延迟消息
for i in range(5):
delay = (i + 1) * 10 # 10秒, 20秒, 30秒, 40秒, 50秒
msg = {
'id': f'delay_msg_{i}',
'type': '取消订单',
'payload': {'order_id': f'ORD_{1000 + i}'}
}
schedule_delayed_message(msg, delay)
time.sleep(0.5)
# 等待所有延迟消息到期
time.sleep(60)
logger.info("演示结束")
六、应用场景
6.1 订单处理系统
电商平台中，用户下单后会产生各种任务：扣库存、发通知、生成物流单、更新积分等。这些任务可以异步处理，用Redis列表做队列完全可以应对。
6.2 数据导出任务
用户点击"导出报表"按钮后，后台生成文件需要几十秒甚至几分钟。可以把导出请求放入队列，由后台worker慢慢处理，用户稍后去下载。
6.3 日志采集与处理
微服务产生的日志需要采集、解析、存储。用Redis列表做中转队列，前端服务负责写入，后端worker负责消费处理。
6.4 秒杀场景的任务削峰
秒杀活动瞬时涌入大量请求，直接打数据库肯定扛不住。把请求先放入Redis队列缓冲，后端按照数据库能承受的速率慢慢消费，起到"削峰填谷"的作用。
七、技术优缺点
7.1 优点
部署简单，运维成本低。 Redis本身就是一个轻量级的缓存服务，很多项目已经部署了Redis，直接用列表就能做消息队列，不需要额外搭建RabbitMQ或Kafka这样的重型中间件。
性能优秀，吞吐量高。 Redis是纯内存操作，单个实例的吞吐量可以达到每秒10万级。对于中小规模的消息处理，Redis列表完全够用。
API简洁，上手快。 Redis列表的核心操作就那几个命令：LPUSH、RPUSH、LPOP、RPOP、BLPOP。开发者不需要学习复杂的消息中间件概念，半天就能上手。
支持多种消费模式。 可以通过组合不同的命令，实现一对一消费、一对多消费、阻塞消费等模式，灵活性很高。
7.2 缺点
消息可靠性有限。 原生Redis列表没有消息确认机制，消费者取走消息后如果处理失败，消息就丢失了。需要自己实现可靠消费逻辑。
不支持消息回溯。 消息一旦被消费就没了，不能像Kafka那样按offset回退重新消费。这在数据对账、故障恢复场景下是个硬伤。
内存消耗大。 消息都存储在Redis内存中，如果消息量大或者积压严重，内存压力会非常大。而且Redis的持久化机制在极端情况下也可能导致数据丢失。
不适合超大规模场景。 当消息量达到每天上亿级别、或者需要跨机房高可用的时候，Redis列表就不够了，这时候应该选择Kafka、Pulsar等专业消息队列。
八、注意事项
8.1 队列命名要规范
建议给队列加上业务前缀和环境的区分，避免多个项目共用Redis时产生冲突。比如用prod:order:task_queue而不是简单的task_queue。
8.2 设置队列过期策略
如果队列长时间没有新消息，应该设置过期时间自动清理，避免僵尸队列占用内存。可以用Redis的EXPIRE命令给队列key设置TTL。
8.3 监控队列长度
一定要对队列长度做监控告警。当队列长度超过阈值时，应该及时通知运维人员，避免消息积压导致服务异常。
# 技术栈: Python + redis-py
# 文件: queue_monitor.py
# 功能: 队列长度监控示例
import redis
import time
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)
# 需要监控的队列列表
WATCHED_QUEUES = [
'task_queue:ready',
'task_queue:processing',
'delayed_queue'
]
# 告警阈值
WARN_THRESHOLD = 5000
CRITICAL_THRESHOLD = 10000
def monitor_queues():
"""
定期检查队列长度
超过阈值时发出警告
"""
while True:
for queue_name in WATCHED_QUEUES:
length = r.llen(queue_name)
if length >= CRITICAL_THRESHOLD:
logger.critical(
f"[紧急] 队列 {queue_name} 长度: {length}, "
f"已超过紧急阈值 {CRITICAL_THRESHOLD}!"
)
elif length >= WARN_THRESHOLD:
logger.warning(
f"[警告] 队列 {queue_name} 长度: {length}, "
f"已超过警告阈值 {WARN_THRESHOLD}"
)
# 每5秒检查一次
time.sleep(5)
if __name__ == '__main__':
try:
monitor_queues()
except KeyboardInterrupt:
logger.info("监控已停止")
8.4 避免大Key
不要把单条消息做得太大。Redis列表的每个元素都是string类型，如果单条消息很大（比如几MB），会影响Redis的性能。建议单条消息控制在1KB以内，如果数据量大，可以把大对象存储在文件存储或数据库中，队列里只存引用ID。
8.5 生产者和消费者要解耦
消息的JSON格式要做好版本管理，新增字段时不要影响已有消费者。建议在消息体中包含版本号字段。
# 技术栈: Python + redis-py
# 文件: versioned_message.py
# 功能: 带版本管理的消息格式示例
import redis
import json
import time
import uuid
r = redis.Redis(host='localhost', port=6379, db=0)
QUEUE_NAME = 'task_queue'
def create_message_v2(task_type, payload):
"""
创建带版本号的消息
v2版本新增了priority字段和trace_id
"""
message = {
'version': 2, # 消息格式版本
'id': str(uuid.uuid4()), # 消息唯一ID
'type': task_type, # 任务类型
'payload': payload, # 任务数据
'priority': 'normal', # 优先级（v2新增）
'trace_id': str(uuid.uuid4()), # 链路追踪ID（v2新增）
'timestamp': time.time(), # 创建时间戳
}
r.rpush(QUEUE_NAME, json.dumps(message))
return message
def create_message_v1(task_type, payload):
"""
创建v1版本的消息（保持向后兼容）
"""
message = {
'version': 1, # 消息格式版本
'id': str(uuid.uuid4()), # 消息唯一ID
'type': task_type, # 任务类型
'payload': payload, # 任务数据
'timestamp': time.time(), # 创建时间戳
}
r.rpush(QUEUE_NAME, json.dumps(message))
return message
def consume_versioned_message(message_json):
"""
根据消息版本选择对应的处理逻辑
"""
message = json.loads(message_json)
version = message.get('version', 1)
if version == 1:
# v1版本处理逻辑
print(f"处理v1消息: type={message['type']}")
elif version == 2:
# v2版本处理逻辑，可以获取priority和trace_id
print(
f"处理v2消息: type={message['type']}, "
f"priority={message['priority']}, "
f"trace_id={message['trace_id']}"
)
else:
print(f"未知版本: {version}")
if __name__ == '__main__':
# 生产一条v2消息
msg = create_message_v2('发邮件', {'to': 'user@example.com'})
consume_versioned_message(json.dumps(msg))
# 生产一条v1消息
msg = create_message_v1('发短信', {'to': '13800000000'})
consume_versioned_message(json.dumps(msg))
九、文章总结
Redis列表在消息队列场景中的应用，本质上就是利用它"两端操作"的特性来实现FIFO的队列模型。对于中小规模的消息处理需求，它是一个简单高效的方案，不需要引入重型中间件，也不需要复杂的运维部署。
但在实际落地时，有几个关键点需要特别注意：第一，消息可靠性不能靠Redis列表本身保证，需要自己设计可靠消费机制，比如双队列加超时恢复；第二，消息去重必须做，用Redis的Set记录已处理的消息ID是最简单的方案；第三，队列长度必须做限制，防止内存溢出；第四，消息格式要有版本管理，方便后续扩展。
如果业务规模比较小、对可靠性要求不是特别高、团队对Redis比较熟悉，那用Redis列表做消息队列是一个非常务实的选择。但如果你的业务需要消息回溯、需要跨机房高可用、或者消息量非常大，那还是建议上专业的消息队列中间件。
说到底，技术选型没有银弹，关键是根据你的实际场景来权衡。Redis列表做消息队列，就像是骑自行车上班——对于大多数场景来说，它够快、够稳、够方便。但如果你需要日行千里，那还是得上火车或者飞机。
评论
围绕“Redis列表数据结构在消息队列场景中的适配与优化实践”参与讨论