Использование таблиц SQL Server в качестве очередей сообщений

Пример практического использования технологии вложил на GitHub.

Для начала хочу констатировать тот факт, что любое обращение к внешним по отношению к СУБД ресурсам внутри транзакции СУБД - это зло. Например, никому из нас не приходит в голову, что в процедуре "ОбработкаПроведения" можно записать сообщение обмена в файл или выполнить обмен через http-сервис с другой информационной базой. Очевидно, что так делать нельзя.

Некоторые считают, что подобный код вполне безобиден, ну или, в крайнем случае, несёт в себе незначительные риски.

Процедура ОбработкаПроведения(Отказ, РежимПроведения)

	// Прикладной код 1С …

	ОтправитьСообщениеВОчередь(ЭтотОбъект); // обращение к внешнему ресурсу

КонецПроцедуры

Мол мы быстренько дёрнем брокер сообщений, потом сразу же зафиксируем транзакцию и всё будет просто супер. Не будет. И вот почему — транзакия проведения документа может стать вложенной. Кто её, где и каким образом сделает вложенной неизвестно. Например, посмотрите на следующий код:

НачатьТранзакцию();

СписокДокументов = СоздатьСвязанныеДокументы();

Для Каждого Док Из СписокДокументов Цикл

	// обращение к внешнему ресурсу в процедуре "ОбработкаПроведения"
	ВсёПровелось = ПровестиСвязанныйДокумент(Док);

	Если Не ВсёПровелось Тогда
		Прервать;
	КонецЕсли;

КонецЦикла;

Если ВсёПровелось Тогда
	ЗафиксироватьТранзакцию();
Иначе
	ОтменитьТранзакцию(); // внешний ресурс уже получил наши документы !
КонецЕсли;

Это типичный пример "Хотели как лучше, а получилось как всегда" (С). Так делать не надо!

На вопрос "как надо" фирма 1С уже ответила - планы обмена. Кроме этого в типовых библиотеках интеграции от 1С можно встретить регистры сведений, название которых начинается на слово "Очередь". Ещё один пример - регистры сведений для отложенного проведения документов. Другими словами для организации асинхронной обработки данных в контексте СУБД необходимо использовать таблицы-очереди. Запись в эти вспомогательные таблицы выполняется одновременно с записью в основные таблицы данных в локальной транзакции СУБД. Таким образом, в том числе, реализуется так называемый транзакционный обмен сообщениями.

Казалось бы, что на этом мою статью можно закончить, но нет. Практика использования планов обмена и регистров сведений в качестве очередей говорит о том, что оба этих механизма имеют существенные недостатки. О планах обмена я предлагаю почитать мою статью на Инфостарте "Планы обмена 1С".

Существенным недостатком регистров сведений при использовании их в качестве таблиц-очередей, я бы отметил тот факт, что для чтения записей (сообщений) из этой таблицы с последующим их удалением (типичный сценарий обработки таблицы как очереди) потребуется выполнить две команды СУБД. Сначала нужно прочитать записи (сообщения), а затем, в случае их успешной обработки, удалить.

При этом очень важно в момент потребления сообщения заблокировать запись регистра сведений таким образом, чтобы другие транзакции не смогли её прочитать и одновременно с этим не были бы заблокированы ожиданием на чтение. Это необходимо для того, чтобы, например, не отправить одно и то же сообщение дважды. К сожалению достичь обеих целей одновременно средствами платформы 1С невозможно.

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

Таким образом доступ к регистру сведений в целях обработки сообщений должен быть монопольным. В принципе, если быть объективным, то это не такая уж и большая проблема. Одно фоновое задание читает сообщения и распределяет их по дочерним фоновым заданиям - обработчикам этих сообщений. Это работает.

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

После того, как асинхронная задача в виде сообщения или просто записи в таблице-очереди успешно сохранена, можно заняться её обработкой или доставкой во внешнюю информационную систему. На этом этапе можно использовать все доступные вам средства: регламентные задания 1С (по отношению к СУБД это внешний процесс), файловый обмен, web и http сервисы, брокер сообщений RabbitMQ и так далее.

Если говорить о доставке сообщений во внешнюю СУБД, например, информационную базу 1С, то я считаю хорошей практикой создание в базе-приёмнике аналогичной входящей очереди, чтобы опять же обрабатывать её в контексте локальной транзакции базы-приёмника.

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

Далее я предлагаю ознакомиться с советами эксперта по SQL Server Ремуса Русану - одного из программистов ядра SQL Server. Он написал очень известную статью "Использование таблиц в качестве очередей". В этой статье он рассматривает различные виды таблиц-очередей. Вариантов таких таблиц может быть множество в зависимости от решаемых прикладных задач. Я опишу лишь некоторые из них.

Ключевым моментом нижеприведённых скриптов является техника так называемого "деструктивного чтения", при которой происходит удаление записи таблицы с одновременным её чтением. Эта возможность впервые появилась в SQL Server 2005. В коде SQL это выглядит так: DELETE … OUTPUT …

Вторым ключевым моментом является использование хинтов ROWLOCK и READPAST для параллельной обработки записей таблиц-очередей несколькими транзакциями одновременно. Про хинт ROWLOCK я писал здесь. Про хинт READPAST я писал тут.

Таблица-очередей "Куча" (heap)

–- Создание таблицы-очереди
CREATE TABLE [HeapQueue]
(
    Payload varbinary(max)
);
GO

–- Процедура помещения сообщения в очередь
CREATE PROCEDURE [usp_EnqueueMessage]
    @payload varbinary(max)
AS
    INSERT [HeapQueue] (Payload) VALUES @payload;
GO

–- Процедура потребления сообщения из очереди
CREATE PROCEDURE [usp_DequeueMessage]
AS
    DELETE TOP(1)
        [HeapQueue] WITH(rowlock, readpast)
    OUTPUT
        deleted.Payload;
GO

Таблица-очередь "куча" не имеет индексов. Хинты ROWLOCK и READPAST позволяют нескольким транзакциям одновременно потреблять сообщения не блокируя друг друга. Это решение очень хорошо масштабируется за счёт добавления необходимого количества потребителей. При этом очередность потребления сообщений не гарантируется.

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

Таблица-очередь "Чтение по времени" (pending)

–- Создание таблицы-очереди
CREATE TABLE [PendingQueue]
(
    WaitForTime datetime NOT NULL,
    Payload varbinary(max)
);
CREATE CLUSTERED INDEX [cdxPendingQueue] ON [PendingQueue] (WaitForTime);
GO

–- Процедура помещения сообщения в очередь
CREATE PROCEDURE [usp_EnqueueMessage]
    @waitForTime datetime,
    @payload varbinary(max)
AS
    INSERT [PendingQueue] (WaitForTime, Payload) VALUES @waitForTime, @payload;
GO

–- Процедура потребления сообщения из очереди
CREATE PROCEDURE [usp_DequeueMessage]
AS
    WITH [CTE] AS
    (
        SELECT TOP(1)
            [Payload]
        FROM
            [PendingQueue] WITH(rowlock, readpast)
        WHERE
            [WaitForTime] < GETUTCDATE()
        ORDER BY
            [WaitForTime]
    )
    DELETE
        [CTE]
    OUTPUT
        deleted.Payload;
GO

Таблица-очередь "чтение по времени" имеет кластерный индекс по полю "WaitForTime", которое используется для определения времени потребления сообщений. Хинты ROWLOCK и READPAST позволяют нескольким транзакциям одновременно потреблять сообщения не блокируя друг друга. Такой вид таблицы-очереди можно использовать для балансировки нагрузки потребления сообщений по времени. При этом очередность потребления сообщений гарантируется всё тем же кластерным индексом и предложением ORDER BY в запросе, но есть один нюанс.

Предположим транзакция № 1 прочитала из очереди первые 10 сообщений, а вторая транзакция № 2 прочитала следующие 10 сообщений. Транзакция № 2 успешно обработала сообщения и завершилась без ошибок. В этот момент транзакция № 1 была отменена и все её сообщения вернулись в очередь. В таком случае случится нарушение очерёдности потребления сообщений.

Если очерёдность потребления и обработки сообщений должна быть гарантирована, то необходимо из запроса деструктивного чтения убрать хинт READPAST. В таком случае только одна транзакция сможет потреблять сообщения из очереди - остальные будут ожидать на блокировке. Следующий пример демонстрирует реализацию подобного требования.

Таблица-очередей "FIFO" (FIFO strict order required)

(соблюдение очерёдности обработки сообщений обязательно)

–- Создание таблицы-очереди
CREATE TABLE [FIFOQueue]
(
    ConsumeOrder bigint NOT NULL IDENTITY(1,1),
    Payload varbinary(max)
);
CREATE CLUSTERED INDEX [cdxFIFOQueue] ON [FIFOQueue] (ConsumeOrder);
GO

–- Процедура помещения сообщения в очередь
CREATE PROCEDURE [usp_EnqueueMessage]
    @payload varbinary(max)
AS
    INSERT [FIFOQueue] (Payload) VALUES @payload;
GO

–- Процедура потребления сообщения из очереди
CREATE PROCEDURE [usp_DequeueMessage]
AS
    WITH [CTE] AS
    (
        SELECT TOP(1)
            [Payload]
        FROM
            [FIFOQueue] WITH(rowlock)
        ORDER BY
            [ConsumeOrder]
    )
    DELETE
        [CTE]
    OUTPUT
        deleted.Payload;
GO

Таблица-очередь "FIFO" имеет кластерный индекс по полю "ConsumeOrder", которое фиксирует порядок потребления сообщений в сочетании с предложением ORDER BY. Отсутствие хинта READPAST определяет потребление сообщений только одной транзакцией в один и тот же момент времени. Таким образом гарантируется соблюдение очередности потребления сообщений. В контексте 1С это, например, может быть необходимо для соблюдения очерёдности проведения документов.

Справедливости ради следует отметить, что этот вариант мало чем отличается от монопольного использования регистров сведений 1С одним потребителем. Разница заключается только в количестве запросов, выполняемых в СУБД (1 против 2) и возможности создавать очереди динамически в коде 1С ("можно" против "нельзя").

Вывод.

Реализовать по-настоящему надёжную с точки зрения обеспечения целостности данных событийно-ориентированную интеграцию или асинхронную обработку данных в распределённых информационных системах, использующих СУБД SQL Server, возможно только двумя способами:

1. Использовать внешний процесс, который постоянно с какой-то периодичностью будет опрашивать таблицу-очередь, например, регламентное задание 1С. Это мы все умеем делать.

2. Использовать функционал "Activation" SQL Server Service Broker. Это отдельная тема.

Видео-презентацию использования Service Broker в контексте 1С можно посмотреть здесь.

На этом пока всё. Спасибо за внимание!