- c - 在位数组中找到第一个零
- linux - Unix 显示有关匹配两种模式之一的文件的信息
- 正则表达式替换多个文件
- linux - 隐藏来自 xtrace 的命令
我想用 ViewFlow library 实现以下用例:
由用户启动的特定 Flow 的进程必须在队列中等待,然后才能执行 celery 作业。每个用户都有一个这些进程的队列。根据时间表或手动触发,允许队列中的下一个进程继续进行。
我的流程中的一个节点进入一个命名队列。应用程序中的其他逻辑为每个队列确定何时允许执行下一个任务。选择队列中的下一个任务并调用其激活的 done() 方法。
示例流程可能如下所示:
class MyFlow(Flow):
start = flow.Start(...).Next(queue_wait)
queue_wait = QueueWait("myQueue").Next(job)
job = celery.Job(...).Next(end)
end = flow.End()
实现排队的最佳方法是什么?在上面的例子中,我不知道“QueueWait”应该是什么。
我已经通读了文档和 viewflow 代码,但我还不清楚这是否可以使用内置的 Node 和 Activation 类(例如 func.Function)来完成,或者我是否需要使用自定义类进行扩展.
最佳答案
经过多次试验,我找到了一个可行且简单的解决方案:
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
我试图让它尽可能通用,并支持多个命名队列和子队列的定义,例如每用户队列。
关于python - 查看流.io : implementing a queue task,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31658996/
我经常在 C 标准文档中看到“实现定义”的说法,并且非常将其作为答案。 然后我在 C99 标准中搜索它,并且: ISO/IEC 9899/1999 (C99) 中第 §3.12 条规定: 3.12 I
“依赖于实现”中的“实现”是什么意思? “依赖于实现”和“依赖于机器”之间有什么区别? 我使用C,所以你可以用C解释它。 最佳答案 当 C 标准讨论实现时,它指的是 C 语言的实现。因此,C 的实现就
我刚刚在 Android-studio 中导入了我的项目,并试图在其中创建一个新的 Activity。但我无法在 android-studio 中创建 Activity 。我指的是here我看不到将目
我想知道您对为什么会发生此错误的意见。在陆上生产环境中,我们使用 CDH4。在我们的本地测试环境中,我们只使用 Apache Hadoop v2.2.0。当我运行在 CDH4 上编译的同一个 jar
我正在尝试集成第三方 SDK (DeepAR)。但是当我构建它时,它会显示一个错误。我试图修复它。如果我创建一个简单的新项目,它就可以正常工作。但是我现有的应用程序我使用相机和 ndk。请帮我找出错误
我很好奇为什么我们有 @Overrides 注释,但接口(interface)没有类似的习惯用法(例如 @Implements 或 @Implementation)。这似乎是一个有用的功能,因为您可能
我对 DAODatabase(适用于 Oracle 11 xe)的 CRUD 方法的实现感到困惑。问题是,在通常存储到 Map 集合的情况下,“U”方法(更新)会插入新元素或更新它(像 ID:Abst
Java-API 告诉我特定类实现了哪些接口(interface)。但有两种不同类型的信息,我不太确定这意味着什么。例如,对于“TreeSet”类:https://docs.oracle.com/en
我有一个接口(interface) MLService,它具有与机器学习算法的训练和交叉验证相关的基本方法,我必须添加两个接口(interface)分类和预测,它们将实现 MLService 并包含根
我一直想知道如何最好地为所有实现相同接口(interface)的类系列实现 equals()(并且客户端应该只使用所述接口(interface)并且永远不知道实现类)。 我还没有编写自己的具体示例,但
我有一个接口(interface)及其 2 个或更多实现, public interface IProcessor { default void method1() { //logic
我有同一个应用程序的免费版和高级版(几乎相同的代码,相同的类,到处都是“if”, list 中的不同包, list 中的进程名称相同)。主要 Activity 使用 IMPLICIT Intent 调
这是我为我的应用程序中的错误部分编写的代码 - (id)initWithData:(NSData *)data <-------- options:(NSUInteger)opti
请查找随附的代码片段。我正在使用此代码将文件从 hdfs 下载到我的本地文件系统 - Configuration conf = new Configuration(); FileSys
我想在 MongoDB 中使用 Grails2.5 中的“ElasticSearch”插件。我的“BuildConfig.groovy”文件是: grails.servlet.version = "3
我收到一条错误消息: fatal error: init(coder:) has not been implemented 对于我的自定义 UITableViewCell。该单元格未注册,在 Stor
得到这个错误 kotlin.NotImplementedError: An operation is not implemented: not implemented 我正在实现一个 ImageBut
typedef int Element; typedef struct { Element *stack; int max_size; int top; } Stack; //
Playground 代码 here 例子: interface IFoo { bar: number; foo?: () => void; } abstract class Abst
我想知道如何抑制警告: Category is implementing a method which will also be implemented by its primary class. 我
我是一名优秀的程序员,十分优秀!