RabbitMQ unacked messages
<p>I'm creating Tasks at a rate of 5 Tasks per second. I can see in RabbitMQ Message rates incoming an average of 5.2/s, I have 240 consumers distributed in 4 Virtual machines (60 per VM), each worker process a Task that last 20 seconds.
In theory I'm supposed to handle 100K task without queuing.</p>

<p>I see a large number of <strong>Unacked</strong> messages. How to get rid of Unacked messages or add a timer to kill them, does that point to be a problem in my worker side?</p>

<p><a href="http://stackoverflow.com/questions/7063224/how-can-i-recover-unacknowledged-amqp-messages-from-other-channels-than-my-conne">How can I recover unacknowledged AMQP messages from other channels than my connection&#39;s own?</a></p>

<p><strong>Queues</strong> tab </p>

<p>Ready Unacked Total incoming deliver / get ack</p>

<p>21,884  960     22,844  5.0/s   0.40/s  0.40/s</p>

<p><a href="http://i.stack.imgur.com/aJbuZ.png" rel="nofollow"><img src="http://i.stack.imgur.com/aJbuZ.png" alt="enter image description here"></a>
<strong>Exchange</strong> tab: stackoverflow direct  D   5.0/s   5.0/s</p>

<p>This is my celeryconfig file.</p>

<pre><code>CELERYD_CHDIR = settings.filepath
CELERY_ENABLE_UTC = True
CELERY_TIMEZONE = "US/Eastern"
CELERY_ACCEPT_CONTENT = ['json', 'pickle', 'yaml']
CELERY_IGNORE_RESULT = True
CELERY_RESULT_BACKEND = "amqp"
CELERY_RESULT_PERSISTENT = True
BROKER_URL = 'amqp://stackoverflow:stackoverflow@rabbitmq:5672'
BROKER_CONNECTION_TIMEOUT = 15
BROKER_CONNECTION_MAX_RETRIES = 5
CELERY_DISABLE_RATE_LIMITS = True
CELERY_TASK_RESULT_EXPIRES = 7200
CELERY_IMPORTS = ("cc.modules.stackoverflow")


CELERY_DEFAULT_QUEUE = "default"
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('gold', Exchange('stackoverflow'), routing_key='stackoverflow.gold'),
    Queue('silver', Exchange('stackoverflow'), routing_key='stackoverflow.silver'),
    Queue('bronze', Exchange('stackoverflow'), routing_key='stackoverflow.bronze'),
)
CELERY_DEFAULT_EXCHANGE = "stackoverflow"
CELERY_DEFAULT_EXCHANGE_TYPE = "topic"
CELERY_DEFAULT_ROUTING_KEY = "default"
CELERY_TRACK_STARTED = True

CELERY_ROUTES = {
    'process_call'     : {'queue': 'gold', 'routing_key': 'stackoverflow.gold', 'exchange': 'stackoverflow',},
    'process_recording': {'queue': 'silver', 'routing_key': 'stackoverflow.silver', 'exchange': 'stackoverflow',},
    'process_campaign' : {'queue': 'bronze', 'routing_key': 'stackoverflow.bronze', 'exchange': 'stackoverflow',}
}
</code></pre>

<p>Message acknowledgement acts like transaction in SQL like commit you have to acknowledge the message received from RabbitMq. This function help to avoid the message loss in your system.</p>

<p>Navigate to <strong>Message acknowledgment</strong> title
<a href="https://www.rabbitmq.com/tutorials/tutorial-two-python.html" rel="nofollow">https://www.rabbitmq.com/tutorials/tutorial-two-python.html</a></p>

python rabbitmq celery