довольно простая временная очередь в MongoDB

Question

Я пытаюсь реализовать довольно простую очередь, используя MongoDB. У меня есть коллекция, которую нужно обработать нескольким тупым работникам. Каждый работник должен найти в коллекции необработанную работу, а затем выполнить ее.

То, как я решаю, какая работа не обрабатывается, основано на простом расчете.

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

Пример документа (поле _id опущено):

{
  updated: 0360,
  interval: 60,
  work: "an object representing the work"
}

Мне нужен атомарный / блокирующий запрос (есть несколько рабочих), который возвращает пакет документов, где updated + interval < currentTime, где currentTime - время на сервере базы данных, а также устанавливает поле updated. до currentTime.

Другими словами:

1. найти: обновлено + интервал
2. вернуть партию из них, скажем, 30
3. set: updated = currentTime

Любая помощь очень ценится!

Answer 1

Поскольку MongoDB не поддерживает транзакции, вы не можете безопасно установить пессимистическую блокировку для пакета элементов, если у вас нет отдельного документа для этого - подробнее об этом в конце.

Давайте начнемс запросом: Вы не можете запросить sth.как «где x + y < z» в MongoDB.Вместо этого вам придется использовать поле для следующего срока оплаты, например, nextDue:

{
  "nextDue": "420",
  "work": { ... }
}

Теперь каждый работник может получить пару элементов (ПРИМЕЧАНИЕ: это весь псевдокод, а неконкретный язык программирования):

var result = db.queue.find( { "nextDue": { $gt, startTime } }).limit(50);
// hint: you can do a random skip here to decrease the chances of collisions
// between workers.

foreach(rover in result)
{
    // pessimistic locking: '-1' indicates this is in progress. 
    // I'd recommend a flag instead, however...

    var currentItem = db.queue.findAndModify({ "_id" : rover.id, "nextDue" : {$gt, startTime}}, {$set : {"nextDue" : -1}});

    if(currentItem == null)
        continue; // hit a lock: another worker is processing this already

    // ... process job ...

    db.queue.findAndModify({ "_id" : rover.id, "nextDue" : "-1"}, {$set : {"nextDue" : yourNextDue }});
}

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

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

Псевдокод, который я представил выше, работал очень хорошо в двух приложениях, но в обоих приложениях отдельные задания выполнялись довольно долго (от полсекунды до нескольких часов).1017 *