rabbitmq unacked что это

12.08.202303.07.2023 admin 0 Comments

RabbitMQ How to Remove Unacked Message

When using RabbitMQ there are situations where you might want to remove unacked messages. You could end up with a polluted environment if you donвЂ™t get things cleaned up. It can also be an indicator of something that is broken such as a stuck consumer.

To remove unacked messages in RabbitMQ:

WeвЂ™re going to show you how to do this from the command line and from the management GUI.

Show queues with unacked connections like this.

To find the channel with unacked messages you can use the following command. It will also show the connection that is associated with the channel.

You can close the channel like this. Specify the connection that was associated with the channel found above. The last parameter is a message that will be sent to the consumer and will appear in the consumerвЂ™s error message.

As an alternative to the step above you could just manually kill the process for the consumer that is associated with the channel.

You can use the following command to verify that you no longer have any unacked messages but these same messages should be in the вЂњReadyвЂќ state.

If you want the messages to be removed from the queue this can be done by purging them with either of the following commands.

You can complete these same steps using the management GUI.

You can view channels and unacked messages here on the вЂњChannelsвЂќ tab but you will want to go to the вЂњConnectionsвЂќ tab to actually close them.

To start off you can see all of your connections on the вЂњConnectionsвЂќ tab like this. Just click on one of them to view it.

When you view a specific connection you can see the number of unacked messages on each of its channels. Just click the вЂњForce CloseвЂќ button to close the connection.

Next you can go to the вЂњQueuesвЂќ tab. Here you should be able to see that there are no unacked messages. The messages that were previously in an unacked state should now be in a вЂњReadyвЂќ state. Select a queue to purge.

Click on the вЂњPurge MessagesвЂќ button to clear all messages from the queue.

RabbitMQ Unacked Message Cause

One thing that can cause unacked messages to build up for a consumer is forgetting to ack messages. If auto_ack is not turned on then an ack needs to be sent manually. If this is forgotten or just not done it will cause unacked messages to build up. Here is an example of broken consumer code that omits the ack line from the call back function.

You can fix this by placing a call to вЂњbasic_ackвЂќ at the end of the callback function like this:

You can also fix this by enabling вЂњauto_ackвЂќ which may or may not be the right thing to do depending on the situation:

Extra Info

You can close a channel programmatically like this:

When you close a connection all channels on that connection will also be closed. You can close a connection programmatically like this:

Источник

Rabbitmq unacked что это

This tutorial assumes RabbitMQ is installed and running on localhost on the standard port ( 5672 ). In case you use a different host, port or credentials, connections settings would require adjusting.

Where to get help

If you’re having trouble going through this tutorial you can contact us through the mailing list or RabbitMQ community Slack.

In the first tutorial we wrote programs to send and receive messages from a named queue. In this one we’ll create a Work Queue that will be used to distribute time-consuming tasks among multiple workers.

The main idea behind Work Queues (aka: Task Queues) is to avoid doing a resource-intensive task immediately and having to wait for it to complete. Instead we schedule the task to be done later. We encapsulate a task as a message and send it to a queue. A worker process running in the background will pop the tasks and eventually execute the job. When you run many workers the tasks will be shared between them.

This concept is especially useful in web applications where it’s impossible to handle a complex task during a short HTTP request window.

Preparation

We will slightly modify the Send program from our previous example, to allow arbitrary messages to be sent from the command line. This program will schedule tasks to our work queue, so let’s name it NewTask :

Like tutorial one we need to generate two projects.

Some help to get the message from the command line argument:

Our old Receive.cs script also requires some changes: it needs to fake a second of work for every dot in the message body. It will handle messages delivered by RabbitMQ and perform the task, so let’s copy it to the Worker project and modify:

Our fake task to simulate execution time:

Round-robin dispatching

One of the advantages of using a Task Queue is the ability to easily parallelise work. If we are building up a backlog of work, we can just add more workers and that way, scale easily.

First, let’s try to run two Worker instances at the same time. They will both get messages from the queue, but how exactly? Let’s see.

In the third one we’ll publish new tasks. Once you’ve started the consumers you can publish a few messages:

Let’s see what is delivered to our workers:

By default, RabbitMQ will send each message to the next consumer, in sequence. On average every consumer will get the same number of messages. This way of distributing messages is called round-robin. Try this out with three or more workers.

Message acknowledgment

Doing a task can take a few seconds. You may wonder what happens if one of the consumers starts a long task and dies with it only partly done. With our current code, once RabbitMQ delivers a message to the consumer it immediately marks it for deletion. In this case, if you kill a worker we will lose the message it was just processing. We’ll also lose all the messages that were dispatched to this particular worker but were not yet handled.

But we don’t want to lose any tasks. If a worker dies, we’d like the task to be delivered to another worker.

In order to make sure a message is never lost, RabbitMQ supports message acknowledgments. An ack(nowledgement) is sent back by the consumer to tell RabbitMQ that a particular message has been received, processed and that RabbitMQ is free to delete it.

If a consumer dies (its channel is closed, connection is closed, or TCP connection is lost) without sending an ack, RabbitMQ will understand that a message wasn’t processed fully and will re-queue it. If there are other consumers online at the same time, it will then quickly redeliver it to another consumer. That way you can be sure that no message is lost, even if the workers occasionally die.

A timeout (30 minutes by default) is enforced on consumer delivery acknowledgement. This helps detect buggy (stuck) consumers that never acknowledge deliveries. You can increase this timeout as described in Delivery Acknowledgement Timeout.

Manual message acknowledgments are turned on by default. In previous examples we explicitly turned them off by setting the autoAck («automatic acknowledgement mode») parameter to true. It’s time to remove this flag and manually send a proper acknowledgment from the worker, once we’re done with a task.

Using this code we can be sure that even if you kill a worker using CTRL+C while it was processing a message, nothing will be lost. Soon after the worker dies all unacknowledged messages will be redelivered.

Acknowledgement must be sent on the same channel that received the delivery. Attempts to acknowledge using a different channel will result in a channel-level protocol exception. See the doc guide on confirmations to learn more.

Forgotten acknowledgment

In order to debug this kind of mistake you can use rabbitmqctl to print the messages_unacknowledged field:

On Windows, drop the sudo:

Message durability

We have learned how to make sure that even if the consumer dies, the task isn’t lost. But our tasks will still be lost if RabbitMQ server stops.

When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren’t lost: we need to mark both the queue and messages as durable.

First, we need to make sure that the queue will survive a RabbitMQ node restart. In order to do so, we need to declare it as durable:

This QueueDeclare change needs to be applied to both the producer and consumer code.

Note on Message Persistence

Fair Dispatch

You might have noticed that the dispatching still doesn’t work exactly as we want. For example in a situation with two workers, when all odd messages are heavy and even messages are light, one worker will be constantly busy and the other one will do hardly any work. Well, RabbitMQ doesn’t know anything about that and will still dispatch messages evenly.

This happens because RabbitMQ just dispatches a message when the message enters the queue. It doesn’t look at the number of unacknowledged messages for a consumer. It just blindly dispatches every n-th message to the n-th consumer.

In order to change this behavior we can use the BasicQos method with the prefetchCount = 1 setting. This tells RabbitMQ not to give more than one message to a worker at a time. Or, in other words, don’t dispatch a new message to a worker until it has processed and acknowledged the previous one. Instead, it will dispatch it to the next worker that is not still busy.

Note about queue size

If all the workers are busy, your queue can fill up. You will want to keep an eye on that, and maybe add more workers, or have some other strategy.

Putting It All Together

Open two terminals.

Run the consumer (worker) first so that the topology (primarily the queue) is in place. Here is its complete code:

Now run the task publisher. Its final code is:

Using message acknowledgments and BasicQos you can set up a work queue. The durability options let the tasks survive even if RabbitMQ is restarted.

Now we can move on to tutorial 3 and learn how to deliver the same message to many consumers.

Production [Non-]Suitability Disclaimer

Please keep in mind that this and other tutorials are, well, tutorials. They demonstrate one new concept at a time and may intentionally oversimplify some things and leave out others. For example topics such as connection management, error handling, connection recovery, concurrency and metric collection are largely omitted for the sake of brevity. Such simplified code should not be considered production ready.

Please take a look at the rest of the documentation before going live with your app. We particularly recommend the following guides: Publisher Confirms and Consumer Acknowledgements, Production Checklist and Monitoring.

Getting Help and Providing Feedback

If you have questions about the contents of this tutorial or any other topic related to RabbitMQ, don’t hesitate to ask them on the RabbitMQ mailing list.

Help Us Improve the Docs 1 «Hello World!»

The simplest thing that does something

2 Work queues

Distributing tasks among workers (the competing consumers pattern)

3 Publish/Subscribe

Sending messages to many consumers at once

4 Routing

Receiving messages selectively

5 Topics

Receiving messages based on a pattern (topics)

6 RPC

7 Publisher Confirms

Reliable publishing with publisher confirms

Источник

RabbitMQ tutorial 1 — Hello World

RabbitMQ позволяет взаимодействовать различным программам при помощи протокола AMQP. RabbitMQ является отличным решением для построения SOA (сервис-ориентированной архитектуры) и распределением отложенных ресурсоемких задач.

Под катом перевод первого из шести уроков официального сайта. Примеры на python, но его знание вовсе не обязательно. Аналогичные примеру программы можно воспроизвести практически на любом популярном ЯП. [так выглядят комментарии переводчика, т.е. меня]

Вступление

RabbitMQ ‒ это брокер сообщений. Его основная цель ‒ принимать и отдавать сообщения. Его можно представлять себе, как почтовое отделение: когда Вы бросаете письмо в ящик, Вы можете быть уверены, что рано или поздно почтальон доставит его адресату [видимо, автор ни разу не имел дела с Почтой России]. В этой аналогии RabbitMQ является одновременно и почтовым ящиком, и почтовым отделением, и почтальоном.

Наибольшее отличие RabbitMQ от почтового отделения в том, что он не имеет дела с бумажными конвертами ‒ RabbitMQ принимает, хранит и отдает бинарные данные ‒ сообщения.

В RabbitMQ, а также обмене сообщениями в целом, используется следующая терминология:

Поставщик, подписчик и брокер не обязаны находиться на одной физической машине, обычно они находятся на разных.

Hello World!

Первый пример не будет особо сложным ‒ давайте просто отправим сообщение, примем его и выведем на экран. Для этого нам потребуется две программы: одна будет отправлять сообщения, другая ‒ принимать и выводить их на экран.
Общая схема такова:

Поставщик отправляет сообщения в очередь с именем «hello», а подписчик получает сообщения из этой очереди.

Библиотека RabbitMQ

RabbitMQ использует протокол AMQP. Для использования RabbitMQ необходима библиотека, поддерживающая этот протокол. Такие библиотеки можно найти практически для каждого языка программирования. Python ‒ не исключение, для него есть несколько библиотек:

Отправка сообщений

Наша первая программа send.py будет просто отправлять одно сообщение в очередь.

Мы подключились к брокеру сообщений, находящемуся на локальном хосте. Для подключения к брокеру, находящемуся на другой машине, достаточно заменить «localhost» на IP адрес этой машины.

Перед отправкой сообщения мы должны убедиться, что очередь, получающая сообщение, существует. Если отправить сообщение в несуществующую очередь, RabbitMQ его проигнорирует. Давайте создадим очередь, в которую будет отправлено сообщение, назовем ее «hello»:

Теперь все готово для отправки сообщения. Наше первое сообщение будет содержать строку и будет отправлено в очередь с именем «hello».

Вообще, в RabbitMQ сообщения не отправляются непосредственно в очередь, они должны пройти через exchange (точка обмена). Но сейчас мы не будем заострять на этом внимание, точки обмена будут рассмотрены в третьем уроке. Сейчас достаточно знать, что точку обмена по-умолчанию можно определить, указав пустую строку. Это специальная точка обмена ‒ она позволяет определять, в какую именно очередь отправлено сообщение. Имя очереди должно быть определено в параметре routing_key:

Перед выходом из программы необходимо убедиться, что буфер был очищен и сообщение дошло до RabbitMQ. В этом можно быть уверенным, если использовать безопасное закрытие соединения с брокером.

Получение сообщений

Наша вторая программа receive.py будет получать сообщения из очереди и выводить их на экран.

Также как и в первой программе сначала необходимо подключиться к RabbitMQ. Для этого следует использовать тот же код, как и ранее. Следующий шаг, как и прежде ‒ убедиться, что очередь существует. Команда queue_declare не будет создавать новую очередь, если она уже существует, поэтому сколько бы раз не была вызвана эта команда, все-равно будет создана только одна очередь.

Вы можете задаться вопросом, почему мы объявляем очередь снова, ведь она уже была объявлена в первой программе. Это нужно, чтобы быть уверенным в существовании очереди, так будет, если сначала будет запущена программа send.py. Но мы не знаем, какая программа будет запущена раньше. В таких случаях лучше объявить очередь в обеих программах.

Мониторинг очередей

Если Вы хотите посмотреть, какие очереди существуют в RabbitMQ на данный момент, Вы можете сделать это с помощью команды rabbitmqctl (потребуются права суперпользователя):

(для Windows ‒ без sudo)

[в нашей компании используют более удобный скрипт мониторинга:]

[скрипт выводит и обновляет каждые 2 секунды таблицу со списком очередей: имя очереди; количество сообщений в обработке; количество сообщений готовых к обработке; общее количество сообщений; устойчивость очереди к перезагрузке сервиса; является ли временной очередью; количество подписчиков]

Получение сообщений из очереди более сложный процесс, чем отправка. Получение осуществляется при помощи подписки с использованием callback функции. При получении каждого сообщения библиотека Pika вызывает эту callback функцию. В нашем примере она будет выводить на экран текст сообщения.

Далее, нам нужно обозначить, что callback функция будет получать сообщения из очереди с именем «hello»:

Здесь необходимо быть уверенным в том, что очередь, на которую мы хотим подписаться, была объявлена. Мы сделали это ранее с помощью команды queue_declare.

Параметр no_ack будет рассмотрен позже [во втором уроке].
И, наконец, запуск бесконечного процесса, который ожидает сообщения из очереди и вызывает callback функцию, когда это необходимо.

Ну а теперь все вместе

Полный код receive.py:

Теперь мы можем попробовать запустить наши программы в терминале. Сначала отправим сообщение при помощи программы send.py:

Выполнение этой программы будет завершаться после отправки каждого сообщения. Теперь сообщение нужно получить:

Отлично! Мы отправили наше первое сообщение через RabbitMQ. Как Вы могли заметить, выполнение программы receive.py не завершилось. Она будет ожидать следующих сообщений, а остановить ее можно, нажав Ctrl+C.
Попробуйте запустить send.py снова в новом окне терминала.

Мы изучили, как отправлять и получать сообщения через именованные очереди. В следующем уроке мы создадим простую очередь задач [ресурсоемких].

UPD: библиотеку, работающую с RabbitMQ, для своего любимого ЯП Вы можете найти на официальном сайте тут.

Источник