RabbitMQ的消息确认机制_rabbitmq消息发送确认机制

import pika

# 连接到RabbitMQ服务器

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

# 启用消息发布确认模式

channel.confirm_select()

# 定义消息未能被路由到队列时的回调

def on_return(channel, method, properties, body):

print('Message returned: {}'.format(body))

channel.add_on_return_callback(on_return)

# 发布消息到RabbitMQ

channel.basic_publish(exchange='', routing_key='my_queue', body='Hello, RabbitMQ!')

# 等待消息发布确认

if channel.wait_for_confirms():

print('Message published successfully.')

else:

print('Message publishing failed.')

# 关闭连接

connection.close()

import pika

# 连接到RabbitMQ服务器

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

# 设置消费者的预取值

channel.basic_qos(prefetch_count=1)

# 定义消息处理函数

def process_message(channel, method, properties, body):

print('Received message: {}'.format(body.decode()))

# 处理消息

# 发送消息确认给RabbitMQ

channel.basic_ack(delivery_tag=method.delivery_tag)

# 订阅队列并开始消费消息

channel.basic_consume(queue='my_queue', on_message_callback=process_message)

# 开始消费消息

channel.start_consuming()

# 关闭连接

connection.close()