kombu 4.0 is slower than kombu 3.0.37 when pulish messaging

[Root-Fang]

It's my code:

    conn = Connection('amqp://admin:[email protected]')
    channel = conn.channel()
    exchange = Exchange('pengdong', 'direct', channel=channel, durable=True)
    producer = conn.Producer(channel=channel, exchange=exchange)
    while True:
        producer.publish("abc", routing_key="fang",delivery_mode=2)

when I run my code in kombu 4.0 ,I find the rate is only 80/s from the rabbitmq management site.
But if I use the kombu 3.0.37 , I find the rate is up to 4000/s.

Why are there so huge difference between the two version of kombu ?

I use pyamqp as the amqplib in both case.

[ask]

It seems like this is due to the lazy declare of the Producers default exchange.

With your code:

from kombu import Connection, Exchange
from kombu.five import monotonic
conn = Connection('amqp://')
channel = conn.channel()
exchange = Exchange('pengdong', 'direct', channel=channel, durable=True)
producer = conn.Producer(channel=channel, exchange=exchange)
start = monotonic()
for i in range(2000):
    producer.publish("abc", routing_key="fang", delivery_mode=2)
print(monotonic() - start)

It takes 4.6s to publish 2000 tasks, but if you change the code to not
use a default exchange:

exchange = Exchange('pengdong', 'direct', channel=channel, durable=True)
producer = conn.Producer(channel=channel)  # <-- no exchange argument
start = monotonic()
for i in range(2000):
    producer.publish("abc", routing_key="fang", delivery_mode=2,
                                  exchange=exchange)  # <- moves exchange here.
print(monotonic() - start)

the runtime is 0.046s

Strangely it does not get slower if you tell the operation to declare the exchange:

    producer.publish(
        "abc", routing_key="fang", delivery_mode=2,
        exchange=exchange, declare=[exchange],
    )

[ask]

It's really strange, since auto_declare=True was changed to be a shortcut to always adding the default exchange to declare:

https://github.com/celery/kombu/blob/32997b17be9b65e56399c837a39e5d72cd86f701/kombu/messaging.py#L169-L171

       if self.auto_declare and self.exchange.name:
            declare = [] if declare is None else declare
            declare.append(self.exchange)

[Root-Fang]

Maybe I know what the problem is. In the case , producer = conn.Producer(channel=channel, exchange=exchange), the producer will try to declare the exchange every time when producer invoke the method publish. And in the method publish, it will wait for exchange.declare() to complete.

@ask

[ask]

Yeah, but the declaration should be cached as AFAICT it uses maybe_declare in both cases, but for some reason I guess it's not cached when using the default exchange.

[EugeniuZ]

I am experiencing a similar problem. After many messages are published, the publish method gets slow. I am also passing a default exchange to the constructor
self.producer = kombu.Producer(self.connection, exchange=self.exchange, auto_declare=True) and using the default producer's exchange when publishing
self.producer.publish(task, routing_key=queue_name, delivery_mode=2, retry=True, priority=prio_level)

I noticed that constructor documentation says that
"""
auto_declare (bool): Automatically declare the default exchange
at instantiation. Default is :const:True.
"""
But it does not actually use the flag in constructor to declare the exchange. Is this a bug or am I missing something ? If this is implemented in the constructor then you don't need to declare the exchange in the publish method.

[EugeniuZ]

@Root-Fang
This worked for me to speed up publishing:
producer = kombu.Producer(connection, exchange=exchange, auto_declare=False) producer.declare()

[ask]

@EugeniuZ It used to declare the exchange in the constructor, but I changed it as

• side effects in constructor is a bad practice in general
• It forces the connection to be evaluated, and without catching connection errors
  so retrying as in this example won't work:

from kombu import Connection, Exchange, Producer

foo_exchange = Exchange('foo')

with Connection('amqp://') as conn:
    producer = Producer(conn, exchange=foo_exchange)  # <- may raise here
    producer.send({'hello': 'world'}, routing_key='foo', retry=True)

With auto_declare=True declaring in the constructor this code may raise a connection error,
and the publish operation won't be retried, even if retry=True is specified.

The solution to this is to pass the exchange in the declare argument:

from kombu import Connection, Exchange, Producer

foo_exchange = Exchange('foo')

with Connection('amqp://') as conn:
    producer = Producer(conn)
    producer.send(
        {'hello': 'world'},
        routing_key='foo',
        retry=True,
        declare=[foo_exchange],  # <-- declared for each retry
    )

So that's what auto_declare does now in 4.0: it makes sure the default exchange is always in declare.

I'm not sure why this will cause it to slow down though, I'm guessing it has something to do with the declaration cache and that for some reason this cache doesn't work, forcing it to be declared every time.

[ask]

Hah, I see what's going on now...

The list of entities to declare is growing for every publish call :)

[ask]

   def publish(self, body, routing_key=None, delivery_mode=None,
                mandatory=False, immediate=False, priority=0,
                content_type=None, content_encoding=None, serializer=None,
                headers=None, compression=None, exchange=None, retry=False,
                retry_policy=None, declare=[], expiration=None, **properties):

It's funny, because I have been arguing at work that it's fine for grown ups to use mutable default arguments, then I walk right into this wall.