rabbitm中如何实现幂等性- 面试宝典

pythonCopy codeimport pika
# 消费者端
def callback(ch, method, properties, body):
    # 检查消息是否已经处理过，这里以文件记录已处理的消息为例
    processed_msgs = read_processed_msgs_from_file()
    if body in processed_msgs:
        print("消息已处理，跳过：", body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
        return
    # 处理消息的逻辑
    process_message(body)
    # 将已处理的消息记录到文件中
    record_processed_msg_to_file(body)
    # 手动确认消息已被消费
    ch.basic_ack(delivery_tag=method.delivery_tag)
def read_processed_msgs_from_file():
    # 读取已处理的消息记录
    processed_msgs = []
    with open("processed_msgs.txt", "r") as file:
        for line in file:
            processed_msgs.append(line.strip())
    return processed_msgs
def record_processed_msg_to_file(msg):
    # 将已处理的消息记录到文件中
    with open("processed_msgs.txt", "a") as file:
        file.write(msg + "\n")
def process_message(msg):
    # 处理消息的逻辑
    print("处理消息：", msg)
def consume():
    # 连接到RabbitMQ
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    # 声明要消费的队列
    channel.queue_declare(queue='my_queue')
    # 设置为手动确认消息模式
    channel.basic_qos(prefetch_count=1)
    # 注册消息处理回调函数
    channel.basic_consume(queue='my_queue', on_message_callback=callback)
    # 开始消费消息
    channel.start_consuming()
# 生产者端
def produce(msg):
    # 连接到RabbitMQ
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    # 声明要发送消息的队列
    channel.queue_declare(queue='my_queue')
    # 发送消息
    channel.basic_publish(exchange='', routing_key='my_queue', body=msg)
    # 关闭连接
    connection.close()
if __name__ == '__main__':
    # 生产者发送消息
    produce("Hello RabbitMQ!")
    # 消费者消费消息
    consume()