何时使用selector.AddReceive，selector.Select

Question

希望能澄清一下我什么时候应该使用selector.AddReceive和selector.Select。这可能不是卡丹斯的问题，但也许我错过了一些关于戈朗的知识。

对于selector.Select，我认为基本的思想是等待一个通道的下一个输出。不完全确定selector.AddRecieve做什么。

例如，在韵律示例中，local_activity 链接并粘贴到下面：

func signalHandlingWorkflow(ctx workflow.Context) error {
    logger := workflow.GetLogger(ctx)
    ch := workflow.GetSignalChannel(ctx, SignalName)
    for {
        var signal string
        if more := ch.Receive(ctx, &signal); !more {
            logger.Info("Signal channel closed")
            return cadence.NewCustomError("signal_channel_closed")
        }

        logger.Info("Signal received.", zap.String("signal", signal))

        if signal == "exit" {
            break
        }

        cwo := workflow.ChildWorkflowOptions{
            ExecutionStartToCloseTimeout: time.Minute,
            // TaskStartToCloseTimeout must be larger than all local activity execution time, because DecisionTask won't
            // return until all local activities completed.
            TaskStartToCloseTimeout: time.Second * 30,
        }
        childCtx := workflow.WithChildOptions(ctx, cwo)

        var processResult string
        err := workflow.ExecuteChildWorkflow(childCtx, processingWorkflow, signal).Get(childCtx, &processResult)
        if err != nil {
            return err
        }
        logger.Sugar().Infof("Processed signal: %v, result: %v", signal, processResult)
    }

    return nil
}

我们不使用任何selector.AddReceive

但是，在这里的示例中，它也使用信号通道：基于外部输入改变uber韵律睡眠时间

我也会在这里粘贴代码

func SampleTimerWorkflow(ctx workflow.Context, timerDelay time.Duration) error 
{
    logger := workflow.GetLogger(ctx)
    resetCh := workflow.GetSignalChannel(ctx, "reset")

    timerFired := false
    delay := timerDelay
    for ;!timerFired; {
        selector := workflow.NewSelector(ctx)

        logger.Sugar().Infof("Setting up a timer to fire after: %v", delay)
        timerCancelCtx, cancelTimerHandler := workflow.WithCancel(ctx)
        timerFuture := workflow.NewTimer(timerCancelCtx, delay)
        selector.AddFuture(timerFuture, func(f workflow.Future) {
            logger.Info("Timer Fired.")
            timerFired = true
        })

        selector.AddReceive(resetCh, func(c workflow.Channel, more bool) {
            logger.Info("Reset signal received.")
            logger.Info("Cancel outstanding timer.")
            cancelTimerHandler()

            var t int
            c.Receive(ctx, &t)
            logger.Sugar().Infof("Reset delay: %v seconds", t)
            delay = time.Second * time.Duration(t)
        })

        logger.Info("Waiting for timer to fire.")
        selector.Select(ctx)
    }

    workflow.GetLogger(ctx).Info("Workflow completed.")
    return nil
}

您可以看到有selector.AddReceive，我不完全确定目的是什么，也不知道什么时候应该使用它。

我试图向我的工作流程发送一个信号，允许我延长过期时间。也就是说，它会延迟ExpirationActivity的调用。

当遵循这个示例(结合我当前的代码)时，一旦我发送信号进行重置，timerFired就会立即被设置为true。

我的当前代码如下(我已经删除了一些无关的if语句)，以前，我只使用了一个selector.Select实例，但在某个地方，我的代码没有正常运行。

func Workflow(ctx workflow.Context) (string, error) {
    // local state per bonus workflow
    bonusAcceptanceState := pending
    logger := workflow.GetLogger(ctx).Sugar()
    logger.Info("Bonus workflow started")
    timerCreated := false
    timerFired := false
    delay := timerDelay

    // To query state in Cadence GUI
    err := workflow.SetQueryHandler(ctx, "bonusAcceptanceState", func(input []byte) (string, error) {
        return bonusAcceptanceState, nil
    })
    if err != nil {
        logger.Info("SetQueryHandler failed: " + err.Error())
        return "", err
    }
    info := workflow.GetInfo(ctx)
    executionTimeout := time.Duration(info.ExecutionStartToCloseTimeoutSeconds) * time.Second
    // decisionTimeout := time.Duration(info.TaskStartToCloseTimeoutSeconds) * time.Second
    decisionTimeout := time.Duration(info.ExecutionStartToCloseTimeoutSeconds) * time.Second
    maxRetryTime := executionTimeout // retry for the entire time

    retryPolicy := &cadence.RetryPolicy{
        InitialInterval:          time.Second,
        BackoffCoefficient:       2,
        MaximumInterval:          executionTimeout,
        ExpirationInterval:       maxRetryTime,
        MaximumAttempts:          0, // unlimited, bound by maxRetryTime
        NonRetriableErrorReasons: []string{},
    }
    ao := workflow.ActivityOptions{
        TaskList:               taskList,
        ScheduleToStartTimeout: executionTimeout, // time until a task has to be picked up by a worker
        ScheduleToCloseTimeout: executionTimeout, // total execution timeout
        StartToCloseTimeout:    decisionTimeout,  // time that a worker can take to process a task
        RetryPolicy:            retryPolicy,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    selector := workflow.NewSelector(ctx)
    timerCancelCtx, cancelTimerHandler := workflow.WithCancel(ctx)

    var signal *singalType

    for {
        signalChan := workflow.GetSignalChannel(ctx, signalName)
        // resetCh := workflow.GetSignalChannel(ctx, "reset")

        selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {
            c.Receive(ctx, &signal)
        })

        selector.Select(ctx)

        if signal.Type == "exit" {
            return "", nil
        }

        // We can check the age and return an appropriate response
        if signal.Type == "ACCEPT" {
            if bonusAcceptanceState == pending {
                logger.Info("Bonus Accepted")
                bonusAcceptanceState = accepted

                var status string
                future := workflow.ExecuteActivity(ctx, AcceptActivity)
                if err := future.Get(ctx, &status); err != nil {
                    logger.Errorw("Activity failed", "error", err)
                }
                // Start expiration timer
                if !timerCreated {
                    timerCreated = true
                    timerFuture := workflow.NewTimer(timerCancelCtx, delay)
                    selector.AddFuture(timerFuture, func(f workflow.Future) {
                        logger.Info("Timer Fired.")
                        timerFired = true
                    })
                }

            }
        }
        
        
        if signal.Type == "ROLLOVER_1X" && bonusAcceptanceState == accepted {
            var status string
            future := workflow.ExecuteActivity(ctx, Rollover1x)
            if err := future.Get(ctx, &status); err != nil {
                logger.Errorw("Activity failed", "error", err)
            }
            selector.Select(ctx)
        }
        if signal.Type == "ROLLOVER_COMPLETE" && bonusAcceptanceState == accepted {
            var status string
            future := workflow.ExecuteActivity(ctx, RolloverComplete)
            if err := future.Get(ctx, &status); err != nil {
                logger.Errorw("Activity failed", "error", err)
                return "", err
            }
            // Workflow is terminated on return result
            return status, nil
        }
        for; !timerFired && bonusAcceptanceState == accepted && signal.Type == "RESET" {
            cancelTimerHandler()

            i, err := strconv.Atoi(signal.Value)
            if err != nil {
                logger.Infow("error in converting")
            }

            logger.Infof("Reset delay: %v seconds", i)
            delay = time.Minute * time.Duration(i)
            timerFuture := workflow.NewTimer(timerCancelCtx, delay)
            selector.AddFuture(timerFuture, func(f workflow.Future) {
                logger.Info("Timer Fired.")
                timerFired = true
            })
            selector.Select(ctx)
        }
        if timerFired {
            var status string
            future := workflow.ExecuteActivity(ctx, ExpirationActivity)
            if err := future.Get(ctx, &status); err != nil {
                logger.Errorw("Activity failed", "error", err)
            }
            return status, nil
        }
    }

}

Answer 1

TL;DR：

• 只有当您需要让选择器在频道上侦听时才使用selector.AddReceive，就像在您的第二个代码片段中一样。如果您只需要在没有选择器的情况下直接处理来自信道的信号，那么您就不需要使用它。
• selector.Select是让代码等待某些事件发生。因为你不想用繁忙的循环来等待。

关于何时使用它们的更多详细信息

本质上，这是与Golang select语句完全相同的概念。Golang select允许您等待定时器和频道。除了Golang没有selector.Select()仅仅是因为它是被语言本身烤熟的，但是Cadence是一个库。

因此，和在golang中一样，您不必使用select语句来使用计时器或通道。只有当您必须编写一些代码来侦听多个事件源时，您才需要它。

例如，如果您有两个通道，您需要编写一些通用的逻辑来处理这两个通道，例如增加一个计数器。这个计数器不属于任何频道。这是个普通的柜台。那么使用selector看起来会很不错。

chA := workflow.GetSignalChannel(ctx, SignalNameA)
chB := workflow.GetSignalChannel(ctx, SignalNameB)
counter := 0

selector.AddReceive(chA)
selector.AddReceive(chB)
For {
  selector.Select()
  counter += 1
}

带有选择器的工作流代码看起来与Golang中的类似：

    counter := 0
    for {
        select {
        case _ := <- chA:
            counter += 1
        case _ := <- chB:
            counter += 1
        }
    }

否则，您可能需要使用两个峡谷在每个频道上收听，并进行计数。戈朗代码如下所示：

counter := 0
go func(){
   for{
      _ := <- chA
      counter += 1
   }
}()

go func(){
   for{
      _ := <- chB
      counter += 1
   }
}()

这可能是种族状况的问题。除非计数器被很好地实现为线程安全。

在Cadence工作流代码中，它是这样的：

chA := workflow.GetSignalChannel(ctx, SignalNameA)
chB := workflow.GetSignalChannel(ctx, SignalNameB)
counter := 0

Workflow.Go(ctx){
   for{
     chA.Receive(ctx,nil)
     counter +=1
   }
}

Workflow.Go(ctx){
   for{
     chB.Receive(ctx,nil)
     counter +=1
   }
}

然而，在卡登斯没有这样的比赛条件，因为卡登斯的协同线(由Workflow.Go()启动)并不是真正的并发。上面的两个工作流代码都应该完美地工作。

但是Cadence仍然提供了与Golang相同的selector，主要是因为第一个更适合编写代码。

Answer 2

检查未来的返回结果

selector.AddFuture(timerFuture, func(f workflow.Future) {
    err := f.Get(ctx, nil)
    if err == nil {
        logger.Info("Timer Fired.")
        timerFired = true
    }
})

参考文献：包装#L 128-L 129