rabbitmq 避免重复消费的具体代码

在 RabbitMQ 中避免重复消费，可以通过以下两种方式实现：

1. 使用消息的全局唯一标识符（Message ID）来判断重复消费

当消费者从队列中获取一条消息时，可以获取到消息的 Message ID。消费者可以将 Message ID 存储到数据库或缓存中，每次获取消息时先检查数据库或缓存中是否存在该 Message ID，如果存在，则说明该消息已被消费过，可以直接丢弃或做其他处理。否则，将 Message ID 存储到数据库或缓存中，并处理该消息。

以下是基于 Python 的 RabbitMQ 消费者示例代码，实现了基于 Message ID 的去重：

import pika
import redis

# 连接 RabbitMQ 和 Redis
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

# 定义消息处理函数
def callback(ch, method, properties, body):
    message_id = properties.message_id
    if redis_client.get(message_id) is not None:
        # Message already consumed, discard it
        print('Message already consumed, discarding...')
    else:
        # Process the message
        print('Processing message:', body)
        # Store the message ID to Redis
        redis_client.set(message_id, 1)

# 启动消费者
channel.basic_consume(queue='my_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

2. 使用 RabbitMQ 自带的消息去重机制（Deduplication）

RabbitMQ 从 3.8.0 版本开始支持消息去重机制，可以在生产者端设置消息的唯一标识符，RabbitMQ 将会根据该标识符去重。如果消费者在消费过程中出现异常，RabbitMQ 将自动将该消息重新加入队列，等待下一次消费。

以下是基于 Python 的 RabbitMQ 生产者和消费者示例代码，实现了基于 Deduplication 的去重：

import pika

# 连接 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 定义 Deduplication header
deduplication_id = 'my_message_id'

# 发送消息
channel.basic_publish(
    exchange='',
    routing_key='my_queue',
    body='Hello, RabbitMQ!',
    properties=pika.BasicProperties(
        delivery_mode=2,  # make message persistent
        headers={
            'deduplication_id': deduplication_id  # set Deduplication header
        }
    )
)

# 定义消息处理函数
def callback(ch, method, properties, body):
    message_id = properties.headers['deduplication_id']
    # Process the message
    print('Processing message:', body)

# 启动消费者
channel.basic_consume(queue='my_queue', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

需要注意的是，如果使用 Deduplication 去重机制，需要保证 Deduplication ID 的唯一性。如果 Deduplication ID 不唯一，可能会导致消息无法被正确去重。