Как транзакционно опрашивать таблицу очередей базы данных с откатом? - PullRequest
Новая
       24

Как транзакционно опрашивать таблицу очередей базы данных с откатом?

2 голосов
/ 10 августа 2010

Я хочу настроить таблицу базы данных, которая будет использоваться в качестве очереди сообщений.В таблицу будут вставлены сообщения с уникальным идентификатором, отметкой времени и состоянием «ОЖИДАНИЕ».

Если предположить, что вставка в эту таблицу выполнена правильно, я хочу знать, каков наилучший способобрабатывать сообщения из этой таблицы с использованием простой базы данных HSQLDB 2.0 (хотя этот вопрос должен применяться ко всем базам данных, поддерживающим транзакции).

Я хочу прочитать следующее сообщение со статусом «в ожидании» и убедиться, что нетдругой обработчик очереди также может обработать ту же запись, а затем либо зафиксировать, либо выполнить откат.

Я включил несколько фрагментов кода, чтобы узнать, как я планирую добиться этого с использованием старого старого JDBC.

  • Будет ли это работать?
  • Есть ли лучшие альтернативы?

DDL: 

create table message_queue (
    qidx integer,
    message varchar(120),
    status varchar(20),
    inserted_date timestamp,
    inserted_by varchar(20),
    processed_date timestamp,
    processed_by varchar(20),
)

insert into message_queue values (1,'Important message here','PENDING','2010-08-10 00:01:00', 'BOB', null,null)

Вот моя очередь, читающая SQL: 

SET AUTOCOMMIT FALSE
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE
START TRANSACTION
    DECLARE nextID INTEGER DEFAULT 0
    SET nextID = select max(qidx) from message_queue where status = 'PENDING' 
    update message_queue set status = 'CONSUMED' where QIDX = nextID
    select * from message_queue where QIDX = nextID
ROLLBACK

Вотмои фрагменты кода подключения: 

    conn.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);
    try {
        String message = getNextMessage(conn); // uses sql in snippet
        processMessage(message);
        conn.commit(); // should commit
    } catch (Exception e) {
        conn.rollback(); // should rollback update
    }

Ответы [ 2 ]

3 голосов
/ 18 ноября 2010

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

  1. один из нескольких рабочих выбирает из таблицы очередей, где статус «ожидает»,Таблица содержит столбец version_number (целое число).
  2. работник просматривает список выбранных им элементов и пытается обновить каждый отдельный элемент до состояния «в процессе» (или любого другого).Предложение where оператора update включает «where version_number =» и, очевидно, первичный ключ для элемента.
  3. Если оператор обновления вернул 1 в качестве числа обновленных строк, этот работник успешно зарезервировал элемент и можетприступить к работе.Если оператор обновления вернул 0 обновленных строк - этот элемент был выбран другим работником, поэтому этот работник должен просто пропустить его без уведомления.
2 голосов
/ 20 октября 2010

Очередь - это, безусловно, нетривиальная архитектура, когда нужно учитывать многопользовательскую природу баз данных.Вообще говоря, вы не хотите реализовывать подобные вещи самостоятельно.Проблема, как я уверен, вы быстро обнаружили, заключается в том, что вы хотите, чтобы только один процесс получил одно сообщение, когда несколько читателей пытаются прочитать из одной и той же очереди.То есть вы должны убедиться, что сообщение обрабатывается только один раз.Итак, вы начинаете думать: «Хорошо, я обновлю эту запись, чтобы другие процессы не получили ее», но затем вы понимаете, что пока вы не зафиксируете другие процессы, вы не увидите ваши изменения.Если вы заблокируете запись, у вас не будет возможности прочитать другой процесс из очереди, пока вы не закончили обработку первого, по существу, сериализовав все это.

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

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

...