Лучшая практика для обработки нескольких строк из базы данных в другом потоке

Question

Я хотел бы спросить, как лучше всего сделать следующее, в настоящее время у меня есть много строк, которые вставляются в базу данных, с некоторым статусом, например 'NEW'

One thread ( ThreadA ) читает 20 строк данных из таблицы со следующим запросом: select * from TABLE where status = 'NEW' order by some_date asc и помещает прочитанные данные в очередь. Он считывает данные только тогда, когда количество элементов в очереди меньше 20.

Другой поток ( ThreadB ) читает данные из очереди и обрабатывает их, во время процесса он меняет статус строки на что-то вроде 'IN PROGRESS' .

Я опасаюсь, что пока ThreadB обрабатывает одну строку, но все еще не обновляет свой статус, если количество элементов в очереди сокращается до менее 20, он будет извлекать еще 20 элементов и помещать их в очередь, поэтому существует вероятность наличия дубликатов в очереди.

Данные могут вернуться со статусом типа 'NEW' Я подумал, что могу обновить прочитанные данные с помощью какого-то флага (что-то вроде извлеченных) и установить флаг как не прочитанный после обработки.

Я чувствую как будто я что-то упускаю. Поэтому я хотел бы спросить, есть ли какие-нибудь передовые методы решения подобных задач.

PS. Количество потоков, которые читают данные, может быть увеличено в будущем, это то, что я стараюсь иметь в виду

Answer 1

Верно, поскольку никто, похоже, не поднимает этот вопрос, я продолжу здесь то, что было начато в комментариях:

Есть много решений для этого. В вашем случае с одним потоком обработки вы можете, например, захотеть сохранить только идентификаторы записей в очереди. Затем ThreadB может получить саму строку, чтобы убедиться, что статус действительно равен NEW. Или используйте optimisti c блокировку с update table set status='IN_PROGRESS' where id=rowId and status='NEW' и прекратите обработку этой строки при исключении.

Оптимизация блокировки - это весело, и вы также можете использовать ее, чтобы полностью избавиться от потока производителя. Представьте себе несколько потоков, обрабатывающих записи из базы данных. Каждый из них может выбрать запись и попытаться установить статус с блокировкой optimisti c, как в первом примере. Таким образом, вполне возможно получить много конфликтов за записи, так что каждый поток может получить N строк, где N - количество потоков, или вдвое больше. А затем попробуйте обработать первую строку, для которой удалось установить IN_PROGRESS. Это решение делает систему менее сложной, и на одну вещь меньше нужно заботиться / синхронизироваться.

И вы можете заставить поток собирать не только записи, которые имеют NEW, но и те, которые IN_PROGRESS и started_date < sysdate = timeout, которые будут включать записи, которые не были обработаны из-за сбоя системы (например, потоку удалось установить для одной строки значение IN_PROGRESS, а затем ваша система вышла из строя. Таким образом, вы получите некоторую устойчивость здесь.