rabbitmq消费者做幂等性处理示例

当我们使用 RabbitMQ 作为消息队列时，消费者需要处理重复消费的情况。实现幂等性处理可以避免重复消费带来的问题。以下是一个示例，演示如何使用 Redis 实现 RabbitMQ 消费者的幂等性处理：

1. 在消费者应用中引入 Redis 的依赖。

2. 在消费者处理消息前，先查询 Redis 中是否已经处理过该消息。如果已经处理过，则不再处理；如果未处理，则处理消息，并将消息处理的标识存储到 Redis 中。

示例代码：

import redis
import pika

# Redis 配置
redis_host = 'localhost'
redis_port = 6379
redis_db = 0
redis_key_prefix = 'rabbitmq_msg_processed:'

# 连接 Redis
redis_conn = redis.Redis(host=redis_host, port=redis_port, db=redis_db)

# RabbitMQ 配置
rabbitmq_host = 'localhost'
rabbitmq_port = 5672
rabbitmq_user = 'guest'
rabbitmq_password = 'guest'
rabbitmq_exchange = 'example_exchange'
rabbitmq_queue = 'example_queue'
rabbitmq_routing_key = 'example_routing_key'

# 连接 RabbitMQ
credentials = pika.PlainCredentials(rabbitmq_user, rabbitmq_password)
connection = pika.BlockingConnection(pika.ConnectionParameters(host=rabbitmq_host, port=rabbitmq_port, credentials=credentials))
channel = connection.channel()

# 声明 exchange 和 queue
channel.exchange_declare(exchange=rabbitmq_exchange, exchange_type='direct', durable=True)
channel.queue_declare(queue=rabbitmq_queue, durable=True)
channel.queue_bind(queue=rabbitmq_queue, exchange=rabbitmq_exchange, routing_key=rabbitmq_routing_key)

# 定义消息处理函数
def callback(ch, method, properties, body):
    msg_id = properties.message_id
    redis_key = redis_key_prefix + msg_id
    
    # 查询 Redis 中是否已经处理过该消息
    if not redis_conn.exists(redis_key):
        # 处理消息
        print('Received message: %r' % body)
        
        # 将消息处理的标识存储到 Redis 中
        redis_conn.set(redis_key, 1)
        print('Processed message: %r' % body)
    else:
        print('Message already processed: %r' % body)
    
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 消费消息
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue=rabbitmq_queue, on_message_callback=callback)

print('Waiting for messages...')
channel.start_consuming()

在上述示例中，我们定义了一个 Redis 的连接对象 `redis_conn`，并在消息处理函数 `callback` 中使用该对象查询并存储消息处理的标识。如果消息处理过，则不再处理；如果未处理，则处理消息并存储标识。这样可以保证每条消息只被处理一次，避免了重复消费的问题。