如果消息在过去 5 秒内未传递，如何检查 python rabbitmq

Question

您好，我正在尝试使用 rabbitmq 在 python 中编写代码。我有一个发送消息的队列，但我必须检查消费者是否在过去 5 秒内发送了消息，如果没有，我应该终止进程。我尝试在互联网上搜索此类功能，但没有相关答案，您能给我一些建议吗？

Answer 1

RabbitMQ 包含一个心跳来检测无响应的对等方/失败的消息

From the docs:

使用心跳检测死 TCP 连接

在某些类型的网络故障中，数据包丢失可能意味着中断 TCP 连接需要相当长的时间（大约 11 分钟） 例如，Linux 上的默认配置）将被 操作系统。 AMQP 0-9-1 提供心跳功能以确保 应用层迅速发现中断的连接 （以及完全没有反应的同伴）。心跳也保卫 针对可能终止“空闲”TCP 的某些网络设备 连接。

使用 Java 客户端启用 Hearbeats：

ConnectionFactory cf = new ConnectionFactory();

// set the heartbeat timeout to 5 seconds
cf.setRequestedHeartbeat(5);

与 .NET 客户端类似：

var cf = new ConnectionFactory();

// set the heartbeat timeout to 5 seconds
cf.RequestedHeartbeat = 5;

希望这会有所帮助。

（在 rabbitmq 文档中有更多关于 dead-letter exchanges 的内容，还有关于 nack 和 ack/(neg/pos) delivery/confirms on this page 但配置 Heartbeats 应该可以解决问题。）

编辑：抱歉，文档中还有一个 python remote procedure callback example！它需要'pika'.. 错过了！

服务器代码示例：

#!/usr/bin/env python
import pika

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

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)

def on_request(ch, method, props, body):
    n = int(body)

    print(" [.] fib(%s)" % n)
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')

print(" [x] Awaiting RPC requests")
channel.start_consuming()

客户端代码示例：

#!/usr/bin/env python
import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)

fibonacci_rpc = FibonacciRpcClient()

print(" [x] Requesting fib(30)")
response = fibonacci_rpc.call(30)
print(" [.] Got %r" % response)