Best practices and whether should I have long lived publishers or not

[Aleksandre Bregadze]

Hi guys,

I have been working with Pika for some time, used it in multiple projects, but still unsure if I am doing things right.

I am almost always using python, thus Pika for publishing and never for consuming, therefore in short I need to send messages consistently and hopefully with as little errors as possible on the way. For this, is it better to use short lived connections or long lived?

Before hands, I was using the short lived approach:
Connect -> send -> close -> repeat for the each message
For some weird reasons I was getting a lot of warnings
"2024-01-23 14:23:52.088532+00:00 [warning] <0.1048873.0> client unexpectedly closed TCP connection"
Also I was getting missed heartbeats, but I have no clue why, connections were short lived.

Now I decided to make long lived connection and I think I broke stuff in production environment -_-...
As a result I decided to rewrite the whole RMQ utilities, I followed the example of yours at https://github.com/pika/pika/blob/main/examples/long_running_publisher.py
and have built the following.

I would really appreciate if you could identify the problems in my code and help me to design the well working version.

1. I initiate a single instance of RMQ per application or worker through django settings:
```

from project.utils.rmq_utils import RMQPublisher

# One connection per app for now
RMQPublisherInstance = RMQPublisher()
RMQPublisherInstance.start()
```

2. Here is the publisher
https://bitbucket.org/shecemenko/workspace/snippets/7qqBEj
I decided to create snippet for it so that you could view all conveniently

Thank you sooo much!

[Luke Bakken]

Yes, in every case, you should use long-lived connections.

Your publisher code seems fine. I can't explain why you're having issues with Django. The only way I can assist you is if you provide a simple Django project, using your Pika code, that exhibits the issue.

Share the project via a git repository that I can clone.

[Aleksandre Bregadze]

Thank you for your message, I have been testing this code past 9 days in prod and so far I have noticed 0 problems. Which is wonderful I guess. But what I did notice is that the long lived threaded publisher did not work with Celery Beat.
Let me elaborate:
1 - I used this class in one of the celery tasks, then called apply() and celery worker worked well. No problems at all, messages were published.
2 - Then instead of apply(), I used Celery Beats scheduler, I think its called PeriodicTask. Here I noticed that self.connection.add_callback_threadsafe was not being executed.

I believe problem is coming from threading, tried to find a work around, but nothing made sense. Thus I added a short lived publisher as well, just for celery beat. Are there any specific configurations when we want to use long lived published with Celery Beat?

[Nishant Varma]

Won't connection/channel get closed after initializtion (__init__)? What does handle_rmq_connection_error do to recover? I have something similar: https://bpa.st/W6ZA. The main difference is: I am trying to ensure_channel before publish.

[Luke Bakken]

Hi everyone,

It's not very productive to talk about code that you can't actually see and run.

Aleks - as I said before, if you would like assistance with your Celery Beat code, provide a simple project that I can review.

Based on the name, "PeriodicTask", it seems like Celery may not keep your long-lived connection around, and that you should open a new connection every time that task executes. That's not a big deal unless your task runs multiple times per second.

Thanks,

Luke

On Tuesday, June 11, 2024 at 3:15:36 AM UTC-7 nishan...@gmail.com wrote:

Won't connection/channel get closed after initializtion (__init__)? What does handle_rmq_connection_error do to recover? I have something similar: https://bpa.st/W6ZA. The main difference is: I am trying to ensure_channel before publish.

On Thursday 2 May 2024 at 21:45:37 UTC+5:30 al...@itsy.dev wrote:

Thank you for your message, I have been testing this code past 9 days in prod and so far I have noticed 0 problems. Which is wonderful I guess. But what I did notice is that the long lived threaded publisher did not work with Celery Beat.
Let me elaborate:
1 - I used this class in one of the celery tasks, then called apply() and celery worker worked well. No problems at all, messages were published.
2 - Then instead of apply(), I used Celery Beats scheduler, I think its called PeriodicTask. Here I noticed that self.connection.add_callback_threadsafe was not being executed.

I believe problem is coming from threading, tried to find a work around, but nothing made sense. Thus I added a short lived publisher as well, just for celery beat. Are there any specific configurations when we want to use long lived published with Celery Beat?

[Nishant Varma]

Is this fine for a publisher that runs in a normal threaded web app? It is supposed to:

a) publish to multiple queues

b) reconnect automatically

It uses threading.local and creates a *new* connection upon failure.

I have included a small performance test to check reconnection.

```

import logging
import threading
import sys
import time

import pika
import simplejson as json

#### CONFIG ####

HOST = "rabbitmq"
PORT = 5672
USER = "guest"
PASS = "guest"
VHOST = "vhost"
EXCHANGE = "exchange"

TEST_DURATION = 60
LOG_FILE = "rmq.log"

#### CONFIG ####

logger = logging.getLogger(__name__)
logger.addHandler(logging.FileHandler(LOG_FILE))

thread = threading.local()


def retry(func, retries=0, delay=0, exceptions=None, *args, **kwargs):
    retry = 0
    while retry < retries:
        try:
            return func(*args, **kwargs)
        except exceptions or Exception:
            logger.exception(f"Retry attempt: {retry + 1}")
            retry += 1
            time.sleep(delay)
    return func(*args, **kwargs)


class RMQ:
    def __init__(self):
        self.parameters = {
            "host": HOST,
            "port": PORT,
            "virtual_host": VHOST,
            "socket_timeout": 5
        }
        self.parameters["credentials"] = pika.PlainCredentials(
            USER,
            PASS
        )
        self.exchange = EXCHANGE
        self.ack = True

    def new_connection(self):
        parameters = pika.ConnectionParameters(**self.parameters)
        self.connection = pika.BlockingConnection(parameters)

    def new_channel(self):
        self.channel = self.connection.channel()
        if self.ack:
            self.channel.confirm_delivery()
        self.channel.exchange_declare(
            exchange=self.exchange,
            exchange_type="direct",
            durable=True,
            auto_delete=False
        )

    def ensure_channel(self):
        if not hasattr(self, "connection") or self.connection.is_closed:
            self.new_connection()
            self.new_channel()

    def publish(self, queue, message, options=None):
        properties = {"content_type": "application/json", "delivery_mode": 2}
        if isinstance(options, dict):
            properties.update(**options)
        self.ensure_channel()
        self.channel.basic_publish(
            exchange=self.exchange,
            routing_key=queue,
            body=message,
            properties=pika.BasicProperties(**properties)
        )

    def publish_retry(
        self,
        queue,
        message,
        retries=3,
        delay=2,
        options=None,
    ):
        retry(
            self.publish,
            retries=retries,
            delay=delay,
            queue=queue,
            message=message,
            options=options,
        )

    def close(self):
        self.channel.close()
        self.connection.close()

    @staticmethod
    def object(config=None):
        if not getattr(thread, "rmq", None):
            thread.rmq = RMQ()
        return thread.rmq


if __name__ == "__main__":
    success = 0
    failure = 0
    rmq = RMQ.object()
    rmq.ensure_channel()
    try:
        rmq.channel.queue_delete("test")
    except:
        pass
    rmq.channel.queue_declare("test", durable=True)
    rmq.channel.queue_bind("test", EXCHANGE)
    def send_msg(name):
        global success
        global failure
        while True:
            try:
                rmq = RMQ.object()
                rmq.publish_retry(
                    queue="test",
                    message=json.dumps("test")
                )
                body = rmq.channel.basic_get(queue="test")[2]
                assert json.loads(body) == "test"
                sys.stdout.write(".")
                sys.stdout.flush()
                success += 1
            except:
                logger.exception("Exception encountered:")
                sys.stdout.write("x")
                sys.stdout.flush()
                failure += 1
            time.sleep(0.25)
    try:
        threads = 4
        ts = []
        for each in range(0, threads):
            t = threading.Thread(target=send_msg, args=[each])
            t.daemon = True
            t.start()
            ts.append(t)
        time.sleep(TEST_DURATION)
        print()
    except KeyboardInterrupt:
        print(" recieved, stopping.")
    finally:
        print("Published: {}".format(success))
        print("Failed: {}".format(failure))
        print("Total: {}".format(success + failure))
        try:
            rmq.channel.queue_delete("test")
        except:
            pass
        sys.exit()

```