在Cadence/Temporal工作流程中处理信号的最佳方式/模式是什么？

Question

当使用像文档建议的那样的信号时：

public class MyWorkflow{
   public Output myWorkflwMethod(Input input){
      ...
   }

   public void mySignalMethod(request){
     // do actual processing here. 
     ...
   }
}

我可能会遇到以下问题:

1. 我想保证一次处理一个FIFO
2. 我想处理signalWithStart的“赛车状态”，其中信号方法调用得太早
3. 我想安全地重置工作流。重置后，可以在历史早期重新应用信号
4. 我想确保工作流不会在信号处理之前提前完成

Answer 1

1. 保证FIFO按顺序一次处理一个
2. 在信号方法过早调用的情况下处理signalWithStart的“竞争条件”。或者在现实中，对于没有signalWithStart的常规信号，信号可能在工作流准备好处理之前来得太早。
3. 安全地重置工作流。重置后，信号可以在历史记录的早期重新应用
4. 确保在处理信号之前不会提前完成工作流程

这四个是在Cadence/Temporal工作流中使用signal时最常见的错误。

您可以应用一种设计模式来一起解决所有问题。

其思想是简化信号处理程序，总是将信号放入一个队列中，而工作流方法将启动另一个工作流线程来处理该队列。

它基于样本（节奏

public class MyWorkflow{
   private Queue<SignalRequest> signalRequestQueue = new LinkedList<>(); 

   public void mySignalMethod(SignalRequest req){
       signalRequestQueue.add(req);
   }

   public Output myWorkflwMethod(Input input){
      //1. do everything necessary/needed before actually processing a signal
      ...

      //2. spin up a workflow thread to process 
      Async.procedure(
      () -> {
          while (true) {
              Workflow.await(() -> !signalRequestQueue.isEmpty());
              final SignalRequest request = signalRequestQueue.poll();
              processSignal(request);
          }
      });


      //3. always wait for queue to be empty before completing/failing/continueAsNew the workflow
      Workflow.await(() -> signalRequestQueue.isEmpty());
      return output
   }

   private void processSignal(request){
     // do your actual processing here. 
     // If a process a single signal may take too much time and you don't care about FIFO, you could also start another workflow thread to process signals in parallel.
     ...
   }
}

您应该使用版本控制来进行迁移。

假设您有这样的现有代码；

public class MyWorkflow{
   public Output myWorkflwMethod(Input input){
      ...
   }

   public void mySignalMethod(request){
     // do your actual processing here. 
     ...
   }
}

然后应使用如下版本控制：

public class MyWorkflow{
   private Queue<SignalRequest> signalRequestQueue = new LinkedList<>(); 

   public void mySignalMethod(SignalRequest req){
       int version = Workflow.getVersion("useSignalQueue", Workflow.DEFAULT_VERSION, 1);
       if( version == 1){
          signalRequestQueue.add(req);
       }else{
          processSignal(req);
       }
   }

   public Output myWorkflwMethod(Input input){
      //1. do everything necessary/needed before actually processing a signal
      ...

       int version = Workflow.getVersion("useSignalQueue", Workflow.DEFAULT_VERSION, 1);
       if( version == 1){
         //2. spin up a workflow thread to process 
         Async.procedure(
         () -> {
             while (true) {
                 Workflow.await(() -> !signalRequestQueue.isEmpty());
                 final SignalRequest request = signalRequestQueue.poll();
                 processSignal(request);
             }
         });
       }

      //3. always wait for queue to be empty before completing/failing/continueAsNeww the workflow
      Workflow.await(() -> signalRequestQueue.isEmpty());
      return output
   }

   private void processSignal(request){
     // do your actual processing here. 
     // If a process a single signal may take too much time and you don't care about FIFO, you could also start another workflow thread to process signals in parallel.
     ...
   }
}

Golang SDK没有1/2/3的相同问题。这是因为Golang SDK提供了一个完全不同的API来处理信号。

Golang SDK不是将信号方法定义为处理程序，而是需要工作流侦听通道来处理信号，这正是这个答案在Java中建议做的事情。请参阅如何发出 API 信号的示例。（参见节奏/时间）

但它有问题#4——工作流可能会在处理信号之前提前完成。这是Golang SDK的一个常见错误。

建议在完成或继续作为新工作流之前，始终排空信号通道。请参阅此示例，了解如何在Golang中排空信号通道。

它类似于使用工作流。在Java中等待，等待处理所有信号。但是因为通道没有api来获取大小，所以我们必须使用“默认”分支来检查空值。

感谢@Maxim指出颞态 html" target="_blank">go sdk 中的 API - 或者，在颞态 go-sdk 中使用“已挂起”API 来检查是否使用了所有信号。

此外，建议监控“unhandledSignal”指标。