[Questions] RabbitMQ keeps closing AMQP connection after missed heartbeats. #13627
-
Community Support Policy
RabbitMQ version used4.0.7 Erlang version used27.2.x Operating system (distribution) usedrabbitmq:4.0.7-management How is RabbitMQ deployed?Community Docker image rabbitmq-diagnostics status outputSee https://www.rabbitmq.com/docs/cli to learn how to use rabbitmq-diagnostics Logs from node 1 (with sensitive values edited out)See https://www.rabbitmq.com/docs/logging to learn how to collect logs Logs from node 2 (if applicable, with sensitive values edited out)See https://www.rabbitmq.com/docs/logging to learn how to collect logs Logs from node 3 (if applicable, with sensitive values edited out)See https://www.rabbitmq.com/docs/logging to learn how to collect logs rabbitmq.confSee https://www.rabbitmq.com/docs/configure#config-location to learn how to find rabbitmq.conf file location Steps to deploy RabbitMQ clusterSteps to reproduce the behavior in questionwait for 90s and the heartbeat dies. advanced.configSee https://www.rabbitmq.com/docs/configure#config-location to learn how to find advanced.config file location Application codeservice1: import os
import time
import socket
import pika.exceptions
from shared_logging.logging import logger
import pika
SCAN_DIR = "/mnt/scans"
# Connect to RabbitMQ
def connect_rabbitmq():
for i in range(10):
try:
connection = pika.BlockingConnection(pika.ConnectionParameters(host="rabbitmq", heartbeat=30))
channel = connection.channel()
channel.queue_declare(queue="ocr_queue", durable=True)
return connection, channel
except (socket.gaierror, pika.exceptions.AMQPConnectionError):
time.sleep(2)
logger.critical("Couldn't connect to RabbitMQ.")
exit(2)
connection, channel = connect_rabbitmq()
channel = connection.channel()
channel.queue_declare(queue="ocr_queue", durable=True)
known_files = set(os.listdir(SCAN_DIR))
while True:
try:
current_files = set(os.listdir(SCAN_DIR))
new_files = current_files - known_files
if new_files:
for new_file in new_files:
logger.info(f"Found new file: {new_file}")
channel.basic_publish(
exchange="",
routing_key="ocr_queue",
body=new_file,
properties=pika.BasicProperties(delivery_mode=2)
)
known_files = current_files
except Exception as e:
logger.error(f"Failed scanning {SCAN_DIR}: {e}")
time.sleep(1)service2: import pika
from shared_logging.logging import logger
import time
import socket
logger.info("Starting OCR service...")
def callback(ch, method, properties, body):
logger.info(f"Received message: {body}")
ch.basic_ack(delivery_tag=method.delivery_tag)
# Connect to RabbitMQ
def connect_rabbitmq():
for i in range(10):
try:
connection = pika.BlockingConnection(pika.ConnectionParameters(host="rabbitmq", heartbeat=30))
channel = connection.channel()
channel.queue_declare(queue="ocr_queue", durable=True)
return connection, channel
except (socket.gaierror, pika.exceptions.AMQPConnectionError):
time.sleep(2)
logger.critical("Couldn't connect to RabbitMQ.")
exit(2)
connection, channel = connect_rabbitmq()
channel = connection.channel()
channel.queue_declare(queue="ocr_queue", durable=True)
channel.basic_consume(queue="ocr_queue", on_message_callback=callback)
logger.info("OCR service ready!")
channel.start_consuming()Kubernetes deployment file# Relevant parts of K8S deployment that demonstrate how RabbitMQ is deployed
# PASTE YAML HERE, BETWEEN BACKTICKSWhat problem are you trying to solve?Hello! RabbitMQ runs fine for the first seconds. Messages can be successfully delivered, see docker-compose log here: but then after two minutes of inactivity, I suddenly receive the error and then when creating new files, it says the channel is closed. What did I miss? Thanks!! |
Beta Was this translation helpful? Give feedback.
Answered by
lukebakken
Mar 26, 2025
Replies: 1 comment 2 replies
-
|
Your Try changing to this: If this still doesn't work, start a discussion here: https://github.com/pika/pika/discussions If you start a discussion there, I expect a code sample I can clone and run to see the same behavior as you. |
Beta Was this translation helpful? Give feedback.
2 replies
-
|
this actually did solve it, thank you. although I am not sure if this is the best solution, as others quote:
Am I using the loop incorrectly? |
Beta Was this translation helpful? Give feedback.
-
|
Ha, yes, I said that about The 100% correct way to implement your code would be to have one |
Beta Was this translation helpful? Give feedback.
Answer selected by
michaelklishin
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Your
service1code blocks Pika's I/O loop, thus causing heartbeats to fail.Try changing to this: