现在，你最好的办法是关闭heartbeats，这将防止RabbitMQ关闭连接，如果你阻塞太长时间。我正在试验pika的核心连接管理和在后台线程中运行的IO循环，但它还不够稳定，无法发布。
在pika v1.1.0中，这是ConnectionParameters(heartbeat=0)

请不要禁用心跳！
从Pika 0.12.0开始，请使用this example code中描述的技术在单独的线程上运行长时间运行的任务，然后确认来自该线程的消息。

我遇到了和你一样的问题。
我的解决方案是：
1.在服务器端关闭心跳
1.评估任务可能花费的最长时间
1.将客户端心跳超时设置为从步骤2获得的时间
为什么？
正如我用以下案例测试的那样：
案例一
1.服务器心跳打开，19世纪
1.客户端未设置
我仍然得到错误时，任务运行了很长时间-- >1800
案例二
1.关闭服务器心跳
1.关闭客户端检测信号
客户端没有错误，除了一个问题--当客户端崩溃时（我的操作系统在某些故障下重新启动），在Rabbitmq管理插件上仍然可以看到tcp连接。这是令人困惑。
案三
1.关闭服务器心跳
1.打开客户端心跳，将其设置为可预见的最大运行时间
在这种情况下，我可以动态地改变每个客户端上的每一个心跳。事实上，我在经常崩溃的机器上设置了心跳。而且，我可以通过Rabbitmq管理插件看到离线机器。

环境

操作系统：Centos x86_64
鼠兔：0.9.13
rabbitmq：3.3.1

1.您可以在long_running_task(connection)中定期调用connection.process_data_events()，该函数将在调用时向服务器发送heartbeat，并使pika客户端远离关闭。
1.在pika BlockingConnection中将心跳值设置为大于call connection.process_data_events()周期。

不要禁用心跳。
最好的解决方案是在一个单独的线程中运行该任务，并将prefetch_count设置为1，这样使用者只会收到一条未确认的消息，类似于channel.basic_qos(prefetch_count=1)

• https://github.com/pika/pika/issues/753#issuecomment-318124510
• https://www.rabbitmq.com/consumer-prefetch.html

您还可以设置一个新线程，在这个新线程中处理消息，并在此线程处于活动状态时在连接上调用.sleep，以防止丢失心跳。下面是一个取自github中@gmr的示例代码块，以及一个指向该问题的链接以供将来参考。

import re
import json
import threading

from google.cloud import bigquery
import pandas as pd
import pika
from unidecode import unidecode

def process_export(url, tablename):
    df = pd.read_csv(csvURL, encoding="utf-8")
    print("read in the csv")
    columns = list(df)
    ascii_only_name = [unidecode(name) for name in columns]
    cleaned_column_names = [re.sub("[^a-zA-Z0-9_ ]", "", name) for name in ascii_only_name]
    underscored_names = [name.replace(" ", "_") for name in cleaned_column_names]
    valid_gbq_tablename = "test." + tablename
    df.columns = underscored_names

    # try:
    df.to_gbq(valid_gbq_tablename, "some_project", if_exists="append", verbose=True, chunksize=10000)
    # print("Finished Exporting")
    # except Exception as error:
    #     print("unable to export due to: ")
    #     print(error)
    #     print()

def data_handler(channel, method, properties, body):
    body = json.loads(body)

    thread = threading.Thread(target=process_export, args=(body["csvURL"], body["tablename"]))
    thread.start()
    while thread.is_alive():  # Loop while the thread is processing
        channel._connection.sleep(1.0)
    print('Back from thread')
    channel.basic_ack(delivery_tag=method.delivery_tag)

def main():
    params = pika.ConnectionParameters(host='localhost', heartbeat=60)
    connection = pika.BlockingConnection(params)
    channel = connection.channel()
    channel.queue_declare(queue="some_queue", durable=True)
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(data_handler, queue="some_queue")
    try:
        channel.start_consuming()
    except KeyboardInterrupt:
        channel.stop_consuming()
    channel.close()

if __name__ == '__main__':
    main()

网址：https://github.com/pika/pika/issues/930#issuecomment-360333837

这里有一个更简单的方法来处理这个线程。如果消费者应用程序在当前作业完成之前不应消费另一个作业，则特别有用。ACK可以在任何时候发送-在本例中，我选择仅在作业完成时（线程不再活动）发送它。
在自己的线程中启动长时间运行的进程，然后通过调用channel.process_data_events（）在循环中监视该线程。在主线程中保留对连接对象的引用，因为它不是线程安全的。基本上：

import time
import pika
from threading import Thread
from functools import partial

rmqconn = pika.BlockingConnection( ... )
rmqchan = rmqconn.channel()
rmqchan.basic_consume(
    queue='test',
    on_message_callback=partial(launch_process,rmqconn)
)
rmqchan.start_consuming()

def launch_process(conn,ch,method,properties,body):
    runthread = Thread(target=run_process,args=body)
    runthread.start()
    while runthread.is_alive():
        time.sleep(2)
        conn.process_data_events()
    ch.basic_ack(delivery_tag=method.delivery_tag)

def run_process(body):
    #do the long-running thing
    time.sleep(10)