I followed the example code here: https://github.com/pika/pika/blob/1.1.0/examples/basic_consumer_threaded.py
But after few hours, my consumer stops with the below traceback. No activity or too many missed heartbeats in the last 5 seconds (I set 5 seconds because of following example code)
Traceback (most recent call last):
File "/usr/lib/python3.7/threading.py", line 926, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
File "main.py", line 44, in do_word
connection.add_callback_threadsafe(cb)
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 744, in add_callback_threadsafe
'BlockingConnection.add_callback_threadsafe() called on '
pika.exceptions.ConnectionWrongStateError: BlockingConnection.add_callback_threadsafe() called on closed or closing connection.
Traceback (most recent call last):
File "main.py", line 66, in <module>
rabbit.channel.start_consuming()
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 1866, in start_consuming
self._process_data_events(time_limit=None)
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 2027, in _process_data_events
self.connection.process_data_events(time_limit=time_limit)
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 825, in process_data_events
self._flush_output(common_terminator)
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 522, in _flush_output
raise self._closed_result.value.error
pika.exceptions.AMQPHeartbeatTimeout: No activity or too many missed heartbeats in the last 5 seconds
Exception in thread Thread-42749:
Traceback (most recent call last):
File "/usr/lib/python3.7/threading.py", line 926, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
File "main.py", line 44, in do_word
connection.add_callback_threadsafe(cb)
File "/opt/cdp/at-cdp-product-analysis-consumer/venv/lib/python3.7/site-packages/pika/adapters/blocking_connection.py", line 744, in add_callback_threadsafe
'BlockingConnection.add_callback_threadsafe() called on '
pika.exceptions.ConnectionWrongStateError: BlockingConnection.add_callback_threadsafe() called on closed or closing connection.
Here is my code:
import json
import config
from app import rabbit, elastic
import functools
import threading
# --------------------------------------------
# Ack message
# --------------------------------------------
def ack_message(channel, delivery_tag):
if channel.is_open:
channel.basic_ack(delivery_tag)
else:
pass
# --------------------------------------------
# Parse message to transactions
# --------------------------------------------
def do_word(connection, channel, delivery_tag, body):
# Do something
cb = functools.partial(ack_message, channel, delivery_tag)
connection.add_callback_threadsafe(cb)
# --------------------------------------------
# Callback function for RabbitMQ Consuming
# --------------------------------------------
def callback(channel, method, properties, body, args):
connection, threads = args
t = threading.Thread(target=do_word,
args=(connection, channel, method.delivery_tag, body))
t.start()
threads.append(t)
if __name__ == "__main__":
# Consuming config
threads = []
# I declare exchange, channel in another python file and call as rabbit object.
message_callback = functools.partial(callback, args=(rabbit.connection, threads))
rabbit.channel.basic_consume(queue=config.RABBITMQ_TRANSACTION_QUEUE, on_message_callback=message_callback)
# Start Consuming
rabbit.channel.start_consuming()
# Wait for all to complete
for thread in threads:
thread.join()
rabbit.connection.close()
What if my heartbeat configuration was too low (5 seconds). Should I disable heartbeat?
- pika: 1.1.0