将外部事件发送到工作流

Question

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

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

根据我所看到的，我们需要创建一个信号通道ch：=workflow.GetSignalChannel（ctx， SignalName），但是上下文在活动中不可用。

Answer 1

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

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

Go示例：

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分钟收到信号，而所有工作流工作人员都已关闭。在这种情况下，当工作人员被带回时，timeoutFuture和signCh都不会为空。由于Selector不保证排序，因此信号可能会在计时器之前传递，即使它是在计时器之后收到的。因此您的代码逻辑应该考虑到这一点。例如，有一个硬性要求，必须忽略自工作流启动30分钟后收到的信号。然后必须将上述示例修改为：

Go示例：

...
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
}

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

Go示例：

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());