python之Amazon ec2 中 pika-rabbitmq 的良好心跳间隔_编程设计

我正在为 rabbitmq 使用最新的 pika 库(0.9.9+)。我对rabbitmq和pika的用法如下:

作为工作人员，我有长时间运行的任务(大约 5 分钟)。这些任务从 rabbitmq 获取它们的请求。请求很少出现，即请求之间有很长的空闲时间。
我之前遇到的问题与空闲连接有关(由于空闲连接导致的连接关闭)。所以，我在 pika 中启用了心跳。
现在heartbeat的选择是个问题。 Pika 似乎是一个单线程库，其中心跳接收和确认恰好在请求时间范围内完成。
因此，如果设置的心跳间隔小于回调函数用于执行其长时间运行计算的时间，服务器将不会收到任何心跳确认并关闭连接。
因此，我假设最小心跳间隔应该是阻塞连接中回调函数的最大计算时间。

amazon ec2 防止其关闭空闲连接的良好心跳值是多少？

此外，有些人建议使用 rabbitmq keepalive(或 libkeepalive)来维护 tcp 连接。我认为在 tcp 层管理心跳要好得多，因为应用程序不需要管理它们。这是真的吗？与 RMQ 心跳相比，keepalive 是一种好方法吗？

我看到有人建议对长时间运行的任务使用多线程和队列。但这是长时间运行任务的唯一选择吗？非常令人失望的是，必须为这种情况使用另一个队列。

提前致谢。我想我已经详细说明了问题。让我知道是否可以提供更多详细信息。

请您参考如下方法:

如果您不习惯使用鼠兔，这 thread使用 kombu 帮助我实现了你想要做的事情:

#!/usr/bin/env python 
import time, logging, weakref, eventlet 
from kombu import Connection, Exchange, Queue 
from kombu.utils.debug import setup_logging 
from kombu.common import eventloop 
from eventlet import spawn_after 
 
eventlet.monkey_patch() 
 
log_format = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) ' 
              '-35s %(lineno) -5d: %(message)s') 
logging.basicConfig(level=logging.INFO, format=log_format) 
logger = logging.getLogger('job_worker') 
logger.setLevel(logging.INFO) 
 
 
def long_running_function(body): 
    time.sleep(300) 
 
def job_worker(body, message): 
    long_running_function(body) 
    message.ack() 
 
def monitor_heartbeats(connection, rate=2): 
    """Function to send heartbeat checks to RabbitMQ. This keeps the 
       connection alive over long-running processes.""" 
    if not connection.heartbeat: 
        logger.info("No heartbeat set for connection: %s" % connection.heartbeat) 
        return 
    interval = connection.heartbeat 
    cref = weakref.ref(connection) 
    logger.info("Starting heartbeat monitor.") 
 
    def heartbeat_check(): 
        conn = cref() 
        if conn is not None and conn.connected: 
            conn.heartbeat_check(rate=rate) 
            logger.info("Ran heartbeat check.") 
            spawn_after(interval, heartbeat_check) 
    return spawn_after(interval, heartbeat_check) 
 
def main(): 
    setup_logging(loglevel='INFO') 
 
    # process for heartbeat monitor 
    p = None 
 
    try: 
        with Connection('amqp://guest:guest@localhost:5672//', heartbeat=300) as conn: 
            conn.ensure_connection() 
            monitor_heartbeats(conn) 
            queue = Queue('job_queue', 
                          Exchange('job_queue', type='direct'), 
                          routing_key='job_queue') 
            logger.info("Starting worker.") 
            with conn.Consumer(queue, callbacks=[job_worker]) as consumer: 
                consumer.qos(prefetch_count=1) 
                for _ in eventloop(conn, timeout=1, ignore_timeouts=True): 
                    pass 
    except KeyboardInterrupt: 
        logger.info("Worker was shut down.") 
 
if __name__ == "__main__": 
    main()

我删除了我的领域特定代码，但本质上这是我使用的框架。