经过大量实验,我得出了一个可行且简单的解决方案:
from viewflow.flow import base
from viewflow.flow.func import FuncActivation
from viewflow.activation import STATUS
class Queue(base.NextNodeMixin,
base.UndoViewMixin,
base.CancelViewMixin,
base.DetailsViewMixin,
base.Event):
"""
Node that halts the flow and waits in a queue. To process the next waiting task
call the dequeue method, optionally specifying the task owner.
Example placing a job in a queue::
class MyFlow(Flow):
wait = Queue().Next(this.job)
job = celery.Job(send_stuff).Next(this.end)
end = flow.End()
somewhere in the application code:
MyFlow.wait.dequeue()
or:
MyFlow.wait.dequeue(process__myprocess__owner=user)
Queues are logically separated by the task_type, so new queues defined in a
subclass by overriding task_type attribute.
"""
task_type = 'QUEUE'
activation_cls = FuncActivation
def __init__(self, **kwargs):
super(Queue, self).__init__(**kwargs)
def dequeue(self, **kwargs):
"""
Process the next task in the queue by created date/time. kwargs is
used to add task filter arguments, thereby effectively splitting the queue
into subqueues. This could be used to implement per-user queues.
Returns True if task was found and dequeued, False otherwise
"""
filter_kwargs = {'flow_task_type': self.task_type, 'status': STATUS.NEW}
if kwargs is not None:
filter_kwargs.update(kwargs)
task = self.flow_cls.task_cls.objects.filter(**filter_kwargs).order_by('created').first()
if task is not None:
lock = self.flow_cls.lock_impl(self.flow_cls.instance)
with lock(self.flow_cls, task.process_id):
task = self.flow_cls.task_cls._default_manager.get(pk=task.pk)
activation = self.activation_cls()
activation.initialize(self, task)
activation.prepare()
activation.done()
return True
return False
我试图使其尽可能通用并支持多个命名队列以及子队列的定义,例如每个用户的队列。