Условная вставка очереди работ для beanstalkd?

Question

Я использую Perl-клиент beanstalkd. Мне нужен простой способ не ставить в очередь одну и ту же работу дважды.

Мне нужно что-то, что должно в основном ждать, пока не появятся элементы K, а затем группировать их вместе. Для этого у меня есть производитель:

insert item(s) into DB
insert a queue item into beanstalkd

И потребитель:

while ( 1 ) {
   beanstalkd.retrieve
   if ( DB items >= K )
       func_to_process_all_items
   kill job
}

Это линейно по количеству запросов / обработки, но в случае:

insert 1 item
... repeat many times ...
insert 1 item

Если предположить, что все эти вставки произошли до того, как было получено задание, это добавило бы N элементов очереди и сделало бы что-то подобное

check DB, process N items
check DB, no items
... many times ...
check DB, no items

Есть ли более разумный способ сделать это, чтобы он не вставлял / не обрабатывал последующие запросы на работу без необходимости?

Answer 1

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

Когда работник пытался добавить идентификатор задания в memcache, успешным было бы только первое - при невозможности добавить идентификатор задания работник удалял задание. Через несколько минут срок действия ключа истекает из memcache, и задание может быть обработано снова.

Не особенно элегантно, но работает.

Answer 2

Будет ли это работать для вас?:

1. Создать две трубки "буфер" и "вживую". Ваш производитель всегда добавляет только «буферную» пробирку.
2. Создайте двух рабочих, один наблюдает за «буфером», а другой наблюдает за «вживую», которые вызывают блокировку reserve(), вызов
3. Всякий раз, когда рабочий «буфера» возвращается в резерв, он зарывает задание, если имеется менее K элементов. Если есть точно K, то он «пинает» все K заданий и передает их в «живую» трубу.
4. "Живой" наблюдатель теперь вернется самостоятельно reserve()

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

Две отдельные очереди предназначены только для более чистого разделения. Вы можете сделать то же самое с одной очередью, хороня все задания до тех пор, пока не появится К-1, а затем по прибытии К-й работы, выполняя все их вживую.