Как нескольким worker'ам последовательно (в рамках topic) брать записи из таблицы сообщений?

Question

[Владимир]

Имеется таблица в которую приходят сообщения из RabbitMq.
Каждое сообщение содержит topic. Список topic'ов динамический.

CREATE TABLE messages (
	id bigserial PRIMARY KEY,
	message_id uuid NOT NULL,
	topic  character varying(128) NOT NULL,
	payload bytea NOT NULL,
	state int NOT NULL
);

Есть N worker'ов, которые должны обрабатывать эти сообщения параллельно.
НО, в рамках одного topic'а, сообщения должны обрабатываться последовательно в порядке поступления (ORDER BY id).

Я реализовал получение сообщения для обработки следующим запросом с использованием advisory locks:

WITH
h_topics AS (
	SELECT id, topic, hashtext(topic) AS h_topic 
	FROM (
		SELECT MIN(m.id) AS id, m.topic
		FROM messages m
		WHERE m.state = 0
		GROUP BY m.topic
	)
)

SELECT 
	m.id, m.message_id , m.topic , m.payload , m.state 
FROM 
(
	SELECT 
		pg_try_advisory_xact_lock(ht.h_topic, 0) AS lock_taken, ht.id
	FROM 
		h_topics ht
		LEFT JOIN pg_locks l_topics 
			ON 
				l_topics.locktype = 'advisory' 
				AND l_topics.classid = ht.h_topic 
				AND l_topics.objid = 0
	WHERE 
        l_topics.classid IS NULL
	ORDER BY ht.id
	LIMIT 1
) m_filter
JOIN messages m ON m.id  = m_filter.id AND m_filter.lock_taken = True;

Есть ли решение попроще?

Comments

[ky0]

А зачем в этой конструкции Постгрес-то?

[Владимир]

ky0, Inbox/Outbox паттерн. Сообщения для RabbitMQ сохраняются в Outbox-таблицу в рамках бизнес транзакции отправителя и потом доставляются в Inbox получателя.. Откуда они в рамках транзакции "забираются", обрабатываются и "удаляются".

[Akina]

Есть N worker'ов, которые должны обрабатывать эти сообщения параллельно.
НО, в рамках одного topic'а, сообщения должны обрабатываться последовательно в порядке поступления (ORDER BY id).

Да без шансов. Можно обеспечить, чтобы сообщения брались на обработку последовательно - то есть сообщение N не берётся на обработку, пока не взято на обработку сообщение N-1. Но на этом и всё, дальше постгресс вообще не при чём, это уже пусть воркеры твои договариваются, чтобы тот, что взял на обработку сообщение N, сидел и курил, пока не завершит обработку воркер, взявший сообщение N-1.

[Владимир]

Akina,

Да без шансов.

Разве SQL код в теле вопроса не делает как раз то что необходимо?
Просто при его использовании, если 2 воркера попытаются взять один и тот же topic, у одного будет ничего, а другой получит пустышку и ему заново надо обращаться к базе. Может есть более элегантное решение?!

[Akina]

Владимир,
Перечитайте ещё раз мой ответ. Я достаточно подробно объяснил, что и почему. Хотя ваш комментарий наводит на мысль, что вы спрашиваете не то, что хотите спросить. И, судя по комментарию, вам вообще похрен порядок именно обработки, вас волнует только порядок, в котором записи забираются на обработку.

Answer 1

[Сергей Соловьев]

BEGIN;
SELECT ... FOR UPDATE SKIP LOCKED;
COMMIT;

дока

Comments

[Владимир]

Если не ошибаюсь, использование FOR UPDATE увеличит нагрузку на диск + разрастание WAL ?!
Поэтому хотелось бы использовать именно advisory

[Сергей Соловьев]

Владимир,

Если не ошибаюсь, использование FOR UPDATE увеличит нагрузку на диск + разрастание WAL ?!

с чего бы?

[Владимир]

Разве FOR UPDATE не ставит флаг HEAP_XMAX_LOCK_ONLY в HeapTupleHeaderData ?

[Сергей Соловьев]

Владимир, этот флаг находится в памяти, иначе в чем смысл оставлять кортежи заблокированными в транзакции, если этой транзакции больше нет?

[Владимир]

Сергей Соловьев, Как я понял, не в памяти: Source

The problem is that transactions might want to lock large numbers of tuples at
any one time, so it's not possible to keep the locks objects in shared memory.

[Сергей Соловьев]

Владимир, in shared memory, это во-первых.
А во-вторых, говорится об огромном количестве заблокированных кортежей, это не 10 или 100 - это уже десятки тысяч кортежей. У тебя на воркер такое количество тасок будет уходить? Очень сомневаюсь, т.к. у тебя в изначальном запросе уже есть LIMIT 1

[Владимир]

Сергей Соловьев, Я выполнил параллельно (3 подключения) этот запрос:

DO $$
BEGIN
    FOR count IN 1..100000
    LOOP
        PERFORM m.topic FROM messages m
		WHERE m.id = floor(random() * 1376000 + 1)
		FOR UPDATE SKIP LOCKED;
		PERFORM pg_sleep(0.1);
    END LOOP;
END; 
$$

Process Monitor'ом отловил события файловой системы:


P.S. Блин, заголовок обрезал. 4ый справа столбец - Write Bytes.
И как мы видим, Postgres пишет в heap и в wal.
Причём чтений данных из heap нет (только pg_internal), так как всё в памяти.

[Владимир]

Сергей Соловьев А вот тоже самое, но без FOR UPDATE SKIP LOCKED:



Чистый read с парочкой служебных write

[Владимир]

Сергей Соловьев, Если добавить LIMIT 1 к FOR UPDATE , то запись у меня ведётся только в WAL:

[Владимир]

Сергей Соловьев, Судя по тестам... FOR UPDATE пишет на диск как минимум в WAL.

[Сергей Соловьев]

Владимир, да, ошибся

Answer 2

[GTD01]

У Вас есть RabbitMQ, зачем тут таблица?

Comments

[Владимир]

RabbitMQ для передачи сообщений между сервисами.
У каждого сервиса есть своя база Inbox. (Single Active Consumer кладёт сообщения в Inbox)
У сервиса может быть 1+ обработчиков.
Работа в рамках одного topic'а должна выполняться последовательно.

[GTD01]

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

т.е. в вашем случае делаем на каждый топик кролика по воркеру

[Владимир]

GTD01, список топиков не статичный... Например: топиком выступают id документа, над которым выполняются опред последовательные действия...

Answer 3

[Владимир]

Решил сделать так:
1. Воркер блокирует топик для работы:

WITH
h_topics AS (
	SELECT 
		topic_order_by,
		topic,
		hashtext(topic) AS h_topic
	FROM 
	(
		SELECT 
			MIN(m.id) AS topic_order_by,
			m.topic
		FROM messages m
		WHERE m.state = 0
		GROUP BY m.topic
	)
)
SELECT 
	m_selected.topic 
FROM 
	h_topics ht_cte
	CROSS JOIN
	(
		SELECT 
			m_filter.topic
		FROM 
		(
			SELECT 
				pg_try_advisory_lock('messages'::regclass::oid::int, ht.h_topic) AS lock_taken,
				ht.topic_order_by,
				ht.topic
			FROM 
				h_topics ht
				LEFT JOIN pg_locks l_topics 
					ON 
						l_topics.locktype = 'advisory' 
						AND l_topics.classid = 'messages'::regclass::oid::int 
						AND l_topics.objid = ht.h_topic
			WHERE l_topics.classid IS NULL
			ORDER BY ht.topic_order_by
			LIMIT 1
		) m_filter
		WHERE m_filter.lock_taken = TRUE
	) m_selected
WHERE m_selected.topic IS NOT NULL
LIMIT 1

2. Потом в транзакции берёт сообщение на обработку:

SELECT 
	m.id,
	m.message_id,
	m.topic,
	m.payload
FROM messages m
WHERE 
	m.topic = @topic
	AND m.state = 0
ORDER BY m.id
LIMIT 1

3. Выполняет работу
4. Обновляет статус сообщения и "публикует" результаты обработки.
5. Коммитит транзакцию
6. Повторяет с п.2 х N-раз (например 100)
7. Разблокирует топик:

PERFORM pg_advisory_unlock('messages'::regclass::oid::int, hashtext(@topic))

В результате, если топиков больше чем worker'ов, то за счёт лимита N-сообщений за одну блокировку топика, не заставим "старые" сообщения простаивать...