python中pika模塊問(wèn)題的深入探究
更新時(shí)間：2018年10月13日 11:40:45 作者：python修行路
這篇文章主要給大家介紹了關(guān)于python中pika模塊問(wèn)題的相關(guān)資料，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家學(xué)習(xí)或者使用python具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧
前言
工作中經(jīng)常用到rabbitmq,而用的語(yǔ)言主要是python，所以也就經(jīng)常會(huì)用到python中的pika模塊，但是這個(gè)模塊的使用，也給我?guī)Я撕芏鄦?wèn)題，這里整理一下關(guān)于這個(gè)模塊我在使用過(guò)程的改變歷程已經(jīng)中間碰到一些問(wèn)題的解決方法
關(guān)于MQ：
MQ全稱(chēng)為Message Queue, 消息隊(duì)列（MQ）是一種應(yīng)用程序?qū)?yīng)用程序的通信方法。應(yīng)用程序通過(guò)讀寫(xiě)出入隊(duì)列的消息（針對(duì)應(yīng)用程序的數(shù)據(jù)）來(lái)通信，而無(wú)需專(zhuān)用連接來(lái)鏈接它們。消息傳遞指的是程序之間通過(guò)在消息中發(fā)送數(shù)據(jù)進(jìn)行通信，而不是通過(guò)直接調(diào)用彼此來(lái)通信，直接調(diào)用通常是用于諸如遠(yuǎn)程過(guò)程調(diào)用的技術(shù)。排隊(duì)指的是應(yīng)用程序通過(guò)隊(duì)列來(lái)通信。隊(duì)列的使用除去了接收和發(fā)送應(yīng)用程序同時(shí)執(zhí)行的要求。
剛開(kāi)寫(xiě)代碼的小菜鳥(niǎo)
在最開(kāi)始使用這個(gè)rabbitmq的時(shí)候，因?yàn)楸旧順I(yè)務(wù)需求，我的程序既需要從rabbitmq消費(fèi)消息，也需要給rabbitmq發(fā)布消息，代碼的邏輯圖為如下：
下面是我的模擬代碼：
#! /usr/bin/env python3
# .-*- coding:utf-8 .-*-
import pika
import time
import threading
import os
import json
import datetime
from multiprocessing import Process
# rabbitmq 配置信息
MQ_CONFIG = {
"host": "192.168.90.11",
"port": 5672,
"vhost": "/",
"user": "guest",
"passwd": "guest",
"exchange": "ex_change",
"serverid": "eslservice",
"serverid2": "airservice"
}
class RabbitMQServer(object):
_instance_lock = threading.Lock()
def __init__(self, recv_serverid, send_serverid):
# self.serverid = MQ_CONFIG.get("serverid")
self.exchange = MQ_CONFIG.get("exchange")
self.channel = None
self.connection = None
self.recv_serverid = recv_serverid
self.send_serverid = send_serverid
def reconnect(self):
if self.connection and not self.connection.is_closed():
self.connection.close()
credentials = pika.PlainCredentials(MQ_CONFIG.get("user"), MQ_CONFIG.get("passwd"))
parameters = pika.ConnectionParameters(MQ_CONFIG.get("host"), MQ_CONFIG.get("port"), MQ_CONFIG.get("vhost"),
credentials)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.exchange, exchange_type="direct")
result = self.channel.queue_declare(queue="queue_{0}".format(self.recv_serverid), exclusive=True)
queue_name = result.method.queue
self.channel.queue_bind(exchange=self.exchange, queue=queue_name, routing_key=self.recv_serverid)
self.channel.basic_consume(self.consumer_callback, queue=queue_name, no_ack=False)
def consumer_callback(self, channel, method, properties, body):
"""
消費(fèi)消息
:param channel:
:param method:
:param properties:
:param body:
:return:
"""
channel.basic_ack(delivery_tag=method.delivery_tag)
process_id = os.getpid()
print("current process id is {0} body is {1}".format(process_id, body))
def publish_message(self, to_serverid, message):
"""
發(fā)布消息
:param to_serverid:
:param message:
:return:
"""
message = dict_to_json(message)
self.channel.basic_publish(exchange=self.exchange, routing_key=to_serverid, body=message)
def run(self):
while True:
self.channel.start_consuming()
@classmethod
def get_instance(cls, *args, **kwargs):
"""
單例模式
:return:
"""
if not hasattr(cls, "_instance"):
with cls._instance_lock:
if not hasattr(cls, "_instance"):
cls._instance = cls(*args, **kwargs)
return cls._instance
def process1(recv_serverid, send_serverid):
"""
用于測(cè)試同時(shí)訂閱和發(fā)布消息
:return:
"""
# 線程1 用于去 從rabbitmq消費(fèi)消息
rabbitmq_server = RabbitMQServer.get_instance(recv_serverid, send_serverid)
rabbitmq_server.reconnect()
recv_threading = threading.Thread(target=rabbitmq_server.run)
recv_threading.start()
i = 1
while True:
# 主線程去發(fā)布消息
message = {"value": i}
rabbitmq_server.publish_message(rabbitmq_server.send_serverid,message)
i += 1
time.sleep(0.01)
class CJsonEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
return obj.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(obj, datetime.date):
return obj.strftime("%Y-%m-%d")
else:
return json.JSONEncoder.default(self, obj)
def dict_to_json(po):
jsonstr = json.dumps(po, ensure_ascii=False, cls=CJsonEncoder)
return jsonstr
def json_to_dict(jsonstr):
if isinstance(jsonstr, bytes):
jsonstr = jsonstr.decode("utf-8")
d = json.loads(jsonstr)
return d
if __name__ == '__main__':
recv_serverid = MQ_CONFIG.get("serverid")
send_serverid = MQ_CONFIG.get("serverid2")
# 進(jìn)程1 用于模擬模擬程序1
p = Process(target=process1, args=(recv_serverid, send_serverid, ))
p.start()
# 主進(jìn)程用于模擬程序2
process1(send_serverid, recv_serverid)
上面是我的將我的實(shí)際代碼更改的測(cè)試模塊，其實(shí)就是模擬實(shí)際業(yè)務(wù)中，我的rabbitmq模塊既有訂閱消息，又有發(fā)布消息的時(shí)候，同時(shí)，訂閱消息和發(fā)布消息用的同一個(gè)rabbitmq連接的同一個(gè)channel
但是這段代碼運(yùn)行之后基本沒(méi)有運(yùn)行多久就會(huì)看到如下錯(cuò)誤信息：
Traceback (most recent call last): File "/app/python3/lib/python3.6/multiprocessing/process.py", line 258, in _bootstrap self.run() File "/app/python3/lib/python3.6/multiprocessing/process.py", line 93, in run self._target(*self._args, **self._kwargs) File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 109, in process1 rabbitmq_server.publish_message(rabbitmq_server.send_serverid,message) File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 76, in publish_message self.channel.basic_publish(exchange=self.exchange, routing_key=to_serverid, body=message) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 2120, in basic_publish mandatory, immediate) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 2206, in publish immediate=immediate) File "/app/python3/lib/python3.6/site-packages/pika/channel.py", line 415, in basic_publish raise exceptions.ChannelClosed() pika.exceptions.ChannelClosed Traceback (most recent call last): File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 144, in <module> process1(send_serverid, recv_serverid) File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 109, in process1 rabbitmq_server.publish_message(rabbitmq_server.send_serverid,message) File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 76, in publish_message self.channel.basic_publish(exchange=self.exchange, routing_key=to_serverid, body=message) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 2120, in basic_publish mandatory, immediate) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 2206, in publish immediate=immediate) File "/app/python3/lib/python3.6/site-packages/pika/channel.py", line 415, in basic_publish raise exceptions.ChannelClosed() pika.exceptions.ChannelClosed Exception in thread Thread-1: Traceback (most recent call last): File "/app/python3/lib/python3.6/threading.py", line 916, in _bootstrap_inner self.run() File "/app/python3/lib/python3.6/threading.py", line 864, in run self._target(*self._args, **self._kwargs) File "/app/py_code/\udce5\udc85\udcb3\udce4\udcba\udc8erabbitmq\udce9\udc97\udcae\udce9\udca2\udc98/low_rabbitmq.py", line 80, in run self.channel.start_consuming() File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 1822, in start_consuming self.connection.process_data_events(time_limit=None) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 749, in process_data_events self._flush_output(common_terminator) File "/app/python3/lib/python3.6/site-packages/pika/adapters/blocking_connection.py", line 477, in _flush_output result.reason_text) pika.exceptions.ConnectionClosed: (505, 'UNEXPECTED_FRAME - expected content header for class 60, got non content header frame instead')
而這個(gè)時(shí)候你查看rabbitmq服務(wù)的日志信息，你會(huì)看到兩種情況的錯(cuò)誤日志如下：
情況一：
=INFO REPORT==== 12-Oct-2018::18:32:37 ===
accepting AMQP connection <0.19439.2> (192.168.90.11:42942 -> 192.168.90.11:5672)
=INFO REPORT==== 12-Oct-2018::18:32:37 ===
accepting AMQP connection <0.19446.2> (192.168.90.11:42946 -> 192.168.90.11:5672)
=ERROR REPORT==== 12-Oct-2018::18:32:38 ===
AMQP connection <0.19446.2> (running), channel 1 - error:
{amqp_error,unexpected_frame,
"expected content header for class 60, got non content header frame instead",
'basic.publish'}
=INFO REPORT==== 12-Oct-2018::18:32:38 ===
closing AMQP connection <0.19446.2> (192.168.90.11:42946 -> 192.168.90.11:5672)
=ERROR REPORT==== 12-Oct-2018::18:33:59 ===
AMQP connection <0.19439.2> (running), channel 1 - error:
{amqp_error,unexpected_frame,
"expected content header for class 60, got non content header frame instead",
'basic.publish'}
=INFO REPORT==== 12-Oct-2018::18:33:59 ===
closing AMQP connection <0.19439.2> (192.168.90.11:42942 -> 192.168.90.11:5672)
情況二：
=INFO REPORT==== 12-Oct-2018::17:41:28 ===
accepting AMQP connection <0.19045.2> (192.168.90.11:33004 -> 192.168.90.11:5672)
=INFO REPORT==== 12-Oct-2018::17:41:28 ===
accepting AMQP connection <0.19052.2> (192.168.90.11:33008 -> 192.168.90.11:5672)
=ERROR REPORT==== 12-Oct-2018::17:41:29 ===
AMQP connection <0.19045.2> (running), channel 1 - error:
{amqp_error,unexpected_frame,
"expected content body, got non content body frame instead",
'basic.publish'}
=INFO REPORT==== 12-Oct-2018::17:41:29 ===
closing AMQP connection <0.19045.2> (192.168.90.11:33004 -> 192.168.90.11:5672)
=ERROR REPORT==== 12-Oct-2018::17:42:23 ===
AMQP connection <0.19052.2> (running), channel 1 - error:
{amqp_error,unexpected_frame,
"expected method frame, got non method frame instead",none}
=INFO REPORT==== 12-Oct-2018::17:42:23 ===
closing AMQP connection <0.19052.2> (192.168.90.11:33008 -> 192.168.90.11:5672)
對(duì)于這種情況我查詢(xún)了很多資料和文檔，都沒(méi)有找到一個(gè)很好的答案，查到關(guān)于這個(gè)問(wèn)題的連接有：
https://stackoverflow.com/questions/49154404/pika-threaded-execution-gets-error-505-unexpected-frame
http://rabbitmq.1065348.n5.nabble.com/UNEXPECTED-FRAME-expected-content-header-for-class-60-got-non-content-header-frame-instead-td34981.html
這個(gè)問(wèn)題其他人碰到的也不少，不過(guò)查了最后的解決辦法基本都是創(chuàng)建兩個(gè)rabbitmq連接，一個(gè)連接用于訂閱消息，一個(gè)連接用于發(fā)布消息，這種情況的時(shí)候，就不會(huì)出現(xiàn)上述的問(wèn)題
在這個(gè)解決方法之前，我測(cè)試了用同一個(gè)連接，不同的channel，讓訂閱消息用一個(gè)channel, 發(fā)布消息用另外一個(gè)channel，但是在測(cè)試過(guò)程依然會(huì)出現(xiàn)上述的錯(cuò)誤。
有點(diǎn)寫(xiě)代碼能力了
最后我也是選擇了用兩個(gè)連接的方法解決出現(xiàn)上述的問(wèn)題，現(xiàn)在是一個(gè)測(cè)試代碼例子：
#! /usr/bin/env python3
# .-*- coding:utf-8 .-*-
import pika
import threading
import json
import datetime
import os
from pika.exceptions import ChannelClosed
from pika.exceptions import ConnectionClosed
# rabbitmq 配置信息
MQ_CONFIG = {
"host": "192.168.90.11",
"port": 5672,
"vhost": "/",
"user": "guest",
"passwd": "guest",
"exchange": "ex_change",
"serverid": "eslservice",
"serverid2": "airservice"
}
class RabbitMQServer(object):
_instance_lock = threading.Lock()
def __init__(self):
self.recv_serverid = ""
self.send_serverid = ""
self.exchange = MQ_CONFIG.get("exchange")
self.connection = None
self.channel = None
def reconnect(self):
if self.connection and not self.connection.is_closed:
self.connection.close()
credentials = pika.PlainCredentials(MQ_CONFIG.get("user"), MQ_CONFIG.get("passwd"))
parameters = pika.ConnectionParameters(MQ_CONFIG.get("host"), MQ_CONFIG.get("port"), MQ_CONFIG.get("vhost"),
credentials)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.exchange, exchange_type="direct")
if isinstance(self, RabbitComsumer):
result = self.channel.queue_declare(queue="queue_{0}".format(self.recv_serverid), exclusive=True)
queue_name = result.method.queue
self.channel.queue_bind(exchange=self.exchange, queue=queue_name, routing_key=self.recv_serverid)
self.channel.basic_consume(self.consumer_callback, queue=queue_name, no_ack=False)
class RabbitComsumer(RabbitMQServer):
def __init__(self):
super(RabbitComsumer, self).__init__()
def consumer_callback(self, ch, method, properties, body):
"""
:param ch:
:param method:
:param properties:
:param body:
:return:
"""
ch.basic_ack(delivery_tag=method.delivery_tag)
process_id = threading.current_thread()
print("current process id is {0} body is {1}".format(process_id, body))
def start_consumer(self):
while True:
self.reconnect()
self.channel.start_consuming()
@classmethod
def run(cls, recv_serverid):
consumer = cls()
consumer.recv_serverid = recv_serverid
consumer.start_consumer()
class RabbitPublisher(RabbitMQServer):
def __init__(self):
super(RabbitPublisher, self).__init__()
def start_publish(self):
self.reconnect()
i = 1
while True:
message = {"value": i}
message = dict_to_json(message)
self.channel.basic_publish(exchange=self.exchange, routing_key=self.send_serverid, body=message)
i += 1
@classmethod
def run(cls, send_serverid):
publish = cls()
publish.send_serverid = send_serverid
publish.start_publish()
class CJsonEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
return obj.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(obj, datetime.date):
return obj.strftime("%Y-%m-%d")
else:
return json.JSONEncoder.default(self, obj)
def dict_to_json(po):
jsonstr = json.dumps(po, ensure_ascii=False, cls=CJsonEncoder)
return jsonstr
def json_to_dict(jsonstr):
if isinstance(jsonstr, bytes):
jsonstr = jsonstr.decode("utf-8")
d = json.loads(jsonstr)
return d
if __name__ == '__main__':
recv_serverid = MQ_CONFIG.get("serverid")
send_serverid = MQ_CONFIG.get("serverid2")
# 這里分別用兩個(gè)線程去連接和發(fā)送
threading.Thread(target=RabbitComsumer.run, args=(recv_serverid,)).start()
threading.Thread(target=RabbitPublisher.run, args=(send_serverid,)).start()
# 這里也是用兩個(gè)連接去連接和發(fā)送，
threading.Thread(target=RabbitComsumer.run, args=(send_serverid,)).start()
RabbitPublisher.run(recv_serverid)
上面代碼中我分別用了兩個(gè)連接去訂閱和發(fā)布消息，同時(shí)另外一對(duì)訂閱發(fā)布也是用的兩個(gè)連接來(lái)執(zhí)行訂閱和發(fā)布，這樣當(dāng)再次運(yùn)行程序之后，就不會(huì)在出現(xiàn)之前的問(wèn)題
關(guān)于斷開(kāi)重連
上面的代碼雖然不會(huì)在出現(xiàn)之前的錯(cuò)誤，但是這個(gè)程序非常脆弱，當(dāng)rabbitmq服務(wù)重啟或者斷開(kāi)之后，程序并不會(huì)有重連接的機(jī)制，所以我們需要為代碼添加重連機(jī)制，這樣即使rabbitmq服務(wù)重啟了或者
rabbitmq出現(xiàn)異常我們的程序也能進(jìn)行重連機(jī)制
#! /usr/bin/env python3
# .-*- coding:utf-8 .-*-
import pika
import threading
import json
import datetime
import time
from pika.exceptions import ChannelClosed
from pika.exceptions import ConnectionClosed
# rabbitmq 配置信息
MQ_CONFIG = {
"host": "192.168.90.11",
"port": 5672,
"vhost": "/",
"user": "guest",
"passwd": "guest",
"exchange": "ex_change",
"serverid": "eslservice",
"serverid2": "airservice"
}
class RabbitMQServer(object):
_instance_lock = threading.Lock()
def __init__(self):
self.recv_serverid = ""
self.send_serverid = ""
self.exchange = MQ_CONFIG.get("exchange")
self.connection = None
self.channel = None
def reconnect(self):
try:
if self.connection and not self.connection.is_closed:
self.connection.close()
credentials = pika.PlainCredentials(MQ_CONFIG.get("user"), MQ_CONFIG.get("passwd"))
parameters = pika.ConnectionParameters(MQ_CONFIG.get("host"), MQ_CONFIG.get("port"), MQ_CONFIG.get("vhost"),
credentials)
self.connection = pika.BlockingConnection(parameters)
self.channel = self.connection.channel()
self.channel.exchange_declare(exchange=self.exchange, exchange_type="direct")
if isinstance(self, RabbitComsumer):
result = self.channel.queue_declare(queue="queue_{0}".format(self.recv_serverid), exclusive=True)
queue_name = result.method.queue
self.channel.queue_bind(exchange=self.exchange, queue=queue_name, routing_key=self.recv_serverid)
self.channel.basic_consume(self.consumer_callback, queue=queue_name, no_ack=False)
except Exception as e:
print(e)
class RabbitComsumer(RabbitMQServer):
def __init__(self):
super(RabbitComsumer, self).__init__()
def consumer_callback(self, ch, method, properties, body):
"""
:param ch:
:param method:
:param properties:
:param body:
:return:
"""
ch.basic_ack(delivery_tag=method.delivery_tag)
process_id = threading.current_thread()
print("current process id is {0} body is {1}".format(process_id, body))
def start_consumer(self):
while True:
try:
self.reconnect()
self.channel.start_consuming()
except ConnectionClosed as e:
self.reconnect()
time.sleep(2)
except ChannelClosed as e:
self.reconnect()
time.sleep(2)
except Exception as e:
self.reconnect()
time.sleep(2)
@classmethod
def run(cls, recv_serverid):
consumer = cls()
consumer.recv_serverid = recv_serverid
consumer.start_consumer()
class RabbitPublisher(RabbitMQServer):
def __init__(self):
super(RabbitPublisher, self).__init__()
def start_publish(self):
self.reconnect()
i = 1
while True:
message = {"value": i}
message = dict_to_json(message)
try:
self.channel.basic_publish(exchange=self.exchange, routing_key=self.send_serverid, body=message)
i += 1
except ConnectionClosed as e:
self.reconnect()
time.sleep(2)
except ChannelClosed as e:
self.reconnect()
time.sleep(2)
except Exception as e:
self.reconnect()
time.sleep(2)
@classmethod
def run(cls, send_serverid):
publish = cls()
publish.send_serverid = send_serverid
publish.start_publish()
class CJsonEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
return obj.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(obj, datetime.date):
return obj.strftime("%Y-%m-%d")
else:
return json.JSONEncoder.default(self, obj)
def dict_to_json(po):
jsonstr = json.dumps(po, ensure_ascii=False, cls=CJsonEncoder)
return jsonstr
def json_to_dict(jsonstr):
if isinstance(jsonstr, bytes):
jsonstr = jsonstr.decode("utf-8")
d = json.loads(jsonstr)
return d
if __name__ == '__main__':
recv_serverid = MQ_CONFIG.get("serverid")
send_serverid = MQ_CONFIG.get("serverid2")
# 這里分別用兩個(gè)線程去連接和發(fā)送
threading.Thread(target=RabbitComsumer.run, args=(recv_serverid,)).start()
threading.Thread(target=RabbitPublisher.run, args=(send_serverid,)).start()
# 這里也是用兩個(gè)連接去連接和發(fā)送，
threading.Thread(target=RabbitComsumer.run, args=(send_serverid,)).start()
RabbitPublisher.run(recv_serverid)
上面的代碼運(yùn)行運(yùn)行之后即使rabbitmq的服務(wù)出問(wèn)題了，但是當(dāng)rabbitmq的服務(wù)好了之后，我們的程序依然可以重新進(jìn)行連接，但是上述這種實(shí)現(xiàn)方式運(yùn)行了一段時(shí)間之后，因?yàn)閷?shí)際的發(fā)布消息的地方的消息是從其他線程或進(jìn)程中獲取的數(shù)據(jù)，這個(gè)時(shí)候你可能通過(guò)queue隊(duì)列的方式實(shí)現(xiàn)，這個(gè)時(shí)候你的queue中如果長(zhǎng)時(shí)間沒(méi)有數(shù)據(jù)，在一定時(shí)間之后來(lái)了數(shù)據(jù)需要發(fā)布出去，這個(gè)時(shí)候你發(fā)現(xiàn)，你的程序會(huì)提示連接被rabbitmq 服務(wù)端給斷開(kāi)了，但是畢竟你設(shè)置了重連機(jī)制，當(dāng)然也可以重連，但是這里想想為啥會(huì)出現(xiàn)這種情況，這個(gè)時(shí)候查看rabbitmq的日志你會(huì)發(fā)現(xiàn)出現(xiàn)了如下錯(cuò)誤：
=ERROR REPORT==== 8-Oct-2018::15:34:19 ===
closing AMQP connection <0.30112.1> (192.168.90.11:54960 -> 192.168.90.11:5672):
{heartbeat_timeout,running}
這是我之前測(cè)試環(huán)境的日志截取的，可以看到是因?yàn)檫@個(gè)錯(cuò)誤導(dǎo)致的，后來(lái)查看pika連接rabbitmq的連接參數(shù)中有這么一個(gè)參數(shù)
這個(gè)參數(shù)默認(rèn)沒(méi)有設(shè)置，那么這個(gè)heatbeat的心跳時(shí)間，默認(rèn)是不設(shè)置的，如果不設(shè)置的話，就是根絕服務(wù)端設(shè)置的，因?yàn)檫@個(gè)心跳時(shí)間是和服務(wù)端進(jìn)行協(xié)商的結(jié)果
當(dāng)這個(gè)參數(shù)設(shè)置為0的時(shí)候則表示不發(fā)送心跳，服務(wù)端永遠(yuǎn)不會(huì)斷開(kāi)這個(gè)連接，所以這里我為了方便我給發(fā)布消息的線程的心跳設(shè)置為0，并且我這里，我整理通過(guò)抓包，看一下服務(wù)端和客戶(hù)端的協(xié)商過(guò)程
從抓包分析中可以看出服務(wù)端和客戶(hù)端首先協(xié)商的是580秒，而客戶(hù)端回復(fù)的是：
這樣這個(gè)連接就永遠(yuǎn)不會(huì)斷了，但是如果我們不設(shè)置heartbeat這個(gè)值，再次抓包我們會(huì)看到如下
從上圖我們可以刪除最后服務(wù)端和客戶(hù)端協(xié)商的結(jié)果就是580，這樣當(dāng)時(shí)間到了之后，如果沒(méi)有數(shù)據(jù)往來(lái)，那么就會(huì)出現(xiàn)連接被服務(wù)端斷開(kāi)的情況了
特別注意
需要特別注意的是，經(jīng)過(guò)我實(shí)際測(cè)試python的pika==0.11.2 版本及以下版本設(shè)置heartbeat的不生效的，只有0.12.0及以上版本設(shè)置才能生效
總結(jié)
以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，如果有疑問(wèn)大家可以留言交流，謝謝大家對(duì)腳本之家的支持。
您可能感興趣的文章:
- python使用pika庫(kù)調(diào)用rabbitmq參數(shù)使用詳情
- Python+Pika+RabbitMQ環(huán)境部署及實(shí)現(xiàn)工作隊(duì)列的實(shí)例教程
- python操作RabbitMq的三種工作模式
- 詳解Python Celery和RabbitMQ實(shí)戰(zhàn)教程
- 如何通過(guò)Python實(shí)現(xiàn)RabbitMQ延遲隊(duì)列
- 基于python實(shí)現(xiàn)監(jiān)聽(tīng)Rabbitmq系統(tǒng)日志代碼示例
- python使用pika庫(kù)調(diào)用rabbitmq交換機(jī)模式詳解
相關(guān)文章
這篇文章主要介紹了Python繪制交通流折線圖詳情，文章基于python的相關(guān)資料展開(kāi)折線圖繪制的實(shí)現(xiàn)流程，感興趣的小伙伴可以參考一下2022-06-06
這篇文章主要介紹了djang常用查詢(xún)SQL語(yǔ)句的使用代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-02-02
代碼講解Python對(duì)Windows服務(wù)進(jìn)行監(jiān)控
本篇文章給大家分享了通過(guò)Python對(duì)Windows服務(wù)進(jìn)行監(jiān)控的實(shí)例代碼，對(duì)此有興趣的朋友可以學(xué)習(xí)參考下。2018-02-02
python3用PyPDF2解析pdf文件,用正則匹配數(shù)據(jù)方式
這篇文章主要介紹了python3用PyPDF2解析pdf文件,用正則匹配數(shù)據(jù)方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-05-05
這篇文章主要介紹了Python 中制作偶數(shù)列表的方法，本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2023-05-05
邊緣檢測(cè)是圖像處理和計(jì)算機(jī)視覺(jué)當(dāng)中的基本問(wèn)題，邊緣檢測(cè)的目的是標(biāo)識(shí)數(shù)字圖像中亮度變化明顯的點(diǎn)。本文將通過(guò)示例和大家介紹一下邊緣檢測(cè)的原理，希望對(duì)大家有所幫助2022-12-12
詳解python列表（list）的使用技巧及高級(jí)操作
這篇文章主要介紹了詳解python列表（list）的使用技巧及高級(jí)操作，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2019-08-08
Python 數(shù)值區(qū)間處理_對(duì)interval 庫(kù)的快速入門(mén)詳解
今天小編就為大家分享一篇Python 數(shù)值區(qū)間處理_對(duì)interval 庫(kù)的快速入門(mén)詳解，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-11-11
python dict 相同key 合并value的實(shí)例
今天小編就為大家分享一篇python dict 相同key 合并value的實(shí)例，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2019-01-01
python實(shí)現(xiàn)的Iou與Giou代碼
今天小編就為大家分享一篇python實(shí)現(xiàn)的Iou與Giou代碼，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-01-01