使用Redis实现消息队列：List、Pub/Sub和Stream的实践

import redis
 
# 连接到Redis服务器
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)
 
# 使用List作为消息队列
def enqueue_task(queue_name, task):
    """将任务添加到指定的Redis队列中"""
    redis_client.lpush(queue_name, task)
 
def dequeue_task(queue_name):
    """从指定的Redis队列中移除并返回一个任务"""
    task = redis_client.brpop(queue_name, timeout=5)
    return task[1] if task else None
 
# 使用Pub/Sub进行消息传递
def publish_message(channel, message):
    """发布消息到指定的频道"""
    redis_client.publish(channel, message)
 
def subscribe_to_channel(channel):
    """订阅指定频道的消息"""
    pubsub = redis_client.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            yield message['data']
 
# 使用Streams作为消息队列和发布订阅的替代
def stream_enqueue_task(stream_name, task_id, task):
    """将任务作为消息添加到Stream中"""
    redis_client.xadd(stream_name, {'task': task}, id=task_id)
 
def stream_dequeue_task(stream_name):
    """从Stream中获取一个任务"""
    tasks = redis_client.xrange(stream_name, count=1)
    return tasks[0].decode('utf-8') if tasks else None
 
def stream_subscribe_to_tasks(stream_name):
    """订阅Stream中的消息"""
    for message in redis_client.xread({stream_name: 0}, count=1, block=5):
        yield message[1][0][1].decode('utf-8')

这段代码展示了如何使用Redis的List、Pub/Sub以及Streams数据类型来实现消息队列和消息传递的功能。代码中的enqueue_task和dequeue_task函数分别用于将任务推入和拉出List。publish_message和subscribe_to_channel函数用于发布和订阅消息。对于Streams，stream_enqueue_task和stream_dequeue_task用于消息队列操作，而stream_subscribe_to_tasks用于消息订阅。这些函数可以帮助开发者理解如何在实际应用中使用Redis的不同特性。