Transparent transition PgQ -> RabbitMQ

Dear habrachitatel, I want to share my experience regarding the transparent for the application transition from the PgQ queue to amqp. Maybe this will seem like a bicycle, maybe some thoughts will come in handy. This article introduces you to the basics of PgQ and rabbitmq.

Background

It so happened historically that PgQ is very actively used in our project. With all its drawbacks, PgQ has an undeniable advantage - transactional with the database, which our code actively used. That is, you can be sure that the event will be in the queue and the database will be updated. Or neither of these will happen. And this advantage should be transferred to the new queue engine.

I will not describe in detail the reasons for leaving PgQ, this is a topic for another article. I will focus only on the transition itself.

Thinking Line, pg_amqp

Googling leads to an extension for PostgreSQL - pg_amqp . It provides stored procedures in PostgreSQL for submission to amqp. The extension works perfectly at the application logic level: by rolling back the transaction in PostgreSQL - the data will not get into amqp. And if they commit, they will.

BEGIN;
INSERTINTO some_table (...) VALUES (...);
SELECT amqp.publish(broker_id, 'amqp.direct', 'foo', 'bar');
ROLLBACK; //Данные и в базу не вставились, и в amqp не отправились

In fact, the extension does not guarantee that the message will go to amqp. Inside, there is only a sequential transaction commit, first in PostgreSQL and then in amqp. And if the connection with amqp between the two commits is lost, the message will be lost. Despite the fact that the probability of such an event is small, there will be lost packets. And given that we work with real money and trading accounts, this is unacceptable.

For those for whom a loss of 0.01% of packages is acceptable - the rest of the article can be left out. Just use pg_amqp if you want to leave from PgQ to amqp.

Start building a bike

* Next, instead of abstract amqp, there will be a specific rabbitmq.

But we still have PostgreSQL, inside which the transactions are full. And we can transactionally insert all the packages into some kind of table, and then somehow send to amqp what did not get there.

No sooner said than done.

All the work with PgQ in our application was done using one stored procedure, which I could freely change.

I created a table

amqp.message(
    id bigint default nextval('amqp_message_id_sequence') primary key,
    pid bigint,
    queue varchar(128),
    message text
)

and a trigger that, when inserted into the table, sent this data to amqp. And the insert storage in pgq was replaced by data insertion into this table. In this case, Overhead is only in sending data to amqp, since in PgQ, too, an insertion into the table occurs at each event. Why do I need a pid I will explain later.

Now there are messages in the table and the receiver from rabbitmq. Messages are written to the table with guarantee as part of the PostgreSQL transaction, and almost all messages are sent to amqp using pg_amqp. But how to understand which messages came and which did not? And how to keep this table in sane sizes (preferably tens or hundreds of rows) so as not to lose performance?

Here rabbitmq comes to the rescue. After all, he knows how to duplicate messages in several queues



So let's give one turn to our business code, and the second we will use to confirm receipt of the package?
No sooner said than done. We create exchange, 2 queues and a messenger, which simply deletes the received message from the amqp.message table.

As a result, there is a table in which only those messages that are "in transit" are stored. The size of the table is always small, since messages are deleted immediately after insertion. The size of the table can be monitored. And the business code of our application now works only with rabbitmq and does not know anything about magic under the hood.

This is how the final scheme of work looks like.

But now an important question arises: how to understand that some package has not arrived? After all, the row in the amqp.message table does not guarantee that the message is lost - it can simply be “on the way”. We need to be sure of this to send the package, otherwise we can create a duplicate of the package, and someone will credit $ 200 instead of $ 100 :) And at the same time, determine that the package did not come and you need to send it as quickly as possible, to minimize disrupt the order of packets in the queue.

This is where basic shamanism begins

All our packages are numbered in ascending order, but the system is multi-threaded, and packages are not required to arrive in rabbitmq in the order in which they are in the table. But within the framework of one process that sends messages to amqp, they must be strictly ordered. PostgreSQL provides the ability to view the pid of the current process (pg_backend_pid ()). And we can expect that within the framework of one pg_backend_pid () the packages will be strictly ordered in ascending order (recall that we generate the package id using nextval). Therefore, when receiving a packet with id N, all packets from the same pg_backend_pid with id below N are not delivered and must be sent.

In total, we need to make a queue messenger, which does only 2 things:

• Listens to the queue of the messenger
• Checks if there are messages in the amqp.message table with the current pid and id less than the current one. If there is - sends them (sends again transactionally, through the table amqp.message)
• Deletes messages by id from the amqp.message table

Profit! We have all the messages reach the recipient, and at the same time we completely got rid of PgQ. The main application code has not changed.

Overhead for everything:

• The amqp.message system table, into which we insert each message, and then delete
• A messenger who removes lines from amqp.message and sends messages

I draw attention to the fact that the logic of the messenger is completely resistant to falls. You can kill him at any time, he will start up again and continue working without any problems.

The system does not take into account the case when the postgres process sent a packet to amqp, which did not reach, and no longer sends packets. I would be grateful if someone could tell me how to handle this situation automatically. Now this is being solved simply by monitoring, but no such events have occurred so far. In general, the fact of sending a message is a very rare event. We use pgbouncer, which reduces the many different pg_backend_pid.