cadence-workflow - 将外部事件发送到工作流

Question

在我们的节奏工作流程中，我们通常需要等待一定时间才能继续进行外部事件（即电子邮件阅读、链接点击等）。

我想知道将这些事件通知我们的工作流程的最佳方式是什么。信号是正确的方式，还是我们应该创建一个等待事件的活动？

从我所见，我们需要创建一个信号通道ch := workflow.GetSignalChannel(ctx, SignalName)，但是上下文在活动中不可用。

Answer 1

信令是向工作流发送事件的推荐方式。

Go 工作流的常用模式是使用 Selector 来等待多个信号通道以及一个计时器未来。

去样本：

sig1Ch := workflow.GetSignalChannel(ctx, "signal1")
sig2Ch := workflow.GetSignalChannel(ctx, "signal2")
timeout := workflow.NewTimer(ctx, time.Minute * 30)

s := workflow.NewSelector(ctx)

var signal1 *Signal1Struct
var signal2 *Signal2Struct
s.AddFuture(timeout, func(f Future) {
})
s.AddReceive(sig1Ch, func(c Channel, more bool) {
    c.Receive(ctx, signal1)
})
s.AddReceive(sig2Ch, func(c Channel, more bool) {
    c.Receive(ctx, signal2)
})

s.Select(ctx)

if signal1 == nil && signal2 == nil {
   // handle timeout
} else {
  // process signals
}

Java 示例：

public interface MyWorkflow {

    @WorkflowMethod
    void main();

    @SignalMethod
    void signal1(Signal1Struct signal);

    @SignalMethod
    void signal2(Signal2Struct signal);

}

public class MyWorkflowImpl implements MyWorkflow {

    private Signal1Struct signal1;
    private Signal2Struct signal2;

    @Override
    public void main() {
        Workflow.await(Duration.ofMinutes(30), 
            () -> signal1 != null || signal2 != null);

        if (signal1 == null && signal2 == null) {
            // handle timeout
        }
        // process signals
    }

    @Override
    public void signal1(Signal1Struct signal) {
        signal1 = signal;
    }

    @Override
    public void signal2(Signal2Struct signal) {
        signal2 = signal;
    }
}

请注意，考虑工作流工作者中断是一个好主意。例如，假设上述工作流已启动，并在启动后 40 分钟收到信号，而所有工作流工作人员都已关闭。在这种情况下，当工人被带回timeout未来并且signCh不会为空时。由于 Selector 不保证排序，因此信号可能在定时器之前传递，即使它是在之后接收的。所以你的代码逻辑应该考虑到这一点。例如，有一个硬性要求，即必须忽略工作流开始后 30 分钟后收到的信号。那么上面的示例必须修改为：

去样本：

...
start := workflow.Now(ctx); // must use workflow clock
s.Select(ctx)
duration := workflow.Now(ctx).Sub(start)
if duration.Minutes() >= 30 || (signal1 == nil && signal2 == nil) {
   // handle timeout
} else {
  // process signals
}

Java 示例：

public void main() {
    long start = Workflow.currentTimeMillis(); // must use workflow clock
    Duration timeout = Duration.ofMinutes(30);
    Workflow.await(timeout, () -> signal1 != null || signal2 != null);
    long duration = Workflow.currentTimeMillis() - start;
    if (timeout.toMillis() <= duration || (signal1 == null && signal2 == null)) {
        // handle timeout
    }
    // process signals
}

即使工作流执行延迟了一个小时，更新后的代码也能正常运行。

编辑：添加信号发送示例

去样本：

c, err := client.NewClient(client.Options{
    HostPort: client.DefaultHostPort,
})
if err != nil {
    log.Fatalln("Unable to create client", err)
}
defer c.Close()

err := c.SignalWorkflow(context.Background(), <workflowId>, "", "signal1", Signal1Struct{})

Java 示例：

WorkflowServiceStubs service = WorkflowServiceStubs.newInstance();
WorkflowClient client = WorkflowClient.newInstance(service);
GreetingWorkflow myWorkflow = 
client.newWorkflowStub(MyWorkflow.class, <workflowId>);
myWorkflow.signal1(new Signal1Struct());