Aby zrealizować prostą kolejkę w redis, która może być użyta do ponownego przesłania uszkodzonych zadań, spróbowałbym czegoś takiego:
- 1 lista „up_for_grabs”
- 1 lista „pracuje_na_pracą”
- automatycznie wygasające blokady
pracownik starający się o pracę zrobiłby coś takiego:
timeout = 3600
#wrap this in a transaction so our cleanup wont kill the task
#Move the job away from the queue so nobody else tries to claim it
job = RPOPLPUSH(up_for_grabs, being_worked_on)
#Set a lock and expire it, the value tells us when that job will time out. This can be arbitrary though
SETEX('lock:' + job, Time.now + timeout, timeout)
#our application logic
do_work(job)
#Remove the finished item from the queue.
LREM being_worked_on -1 job
#Delete the item's lock. If it crashes here, the expire will take care of it
DEL('lock:' + job)
A od czasu do czasu mogliśmy po prostu chwycić naszą listę i sprawdzić, czy wszystkie zadania, które tam są, rzeczywiście mają blokadę. Jeśli znajdziemy jakieś zadania, które NIE mają blokady, oznacza to, że wygasły i nasz pracownik prawdopodobnie się zawiesił. w tym przypadku prześlemy ponownie.
To byłby pseudokod do tego:
loop do
items = LRANGE(being_worked_on, 0, -1)
items.each do |job|
if !(EXISTS("lock:" + job))
puts "We found a job that didn't have a lock, resubmitting"
LREM being_worked_on -1 job
LPUSH(up_for_grabs, job)
end
end
sleep 60
end
