希望能澄清一下我应该何时使用selector.AddReceiveand selector.Select。这可能不是节奏问题，但也许我缺少一些关于 Golang 的知识。
因为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
但是，在此处的示例中，它也使用了信号通道：根据外部输入更改超级节奏睡眠时间
我也会在这里粘贴代码
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立即设置为真。
我当前的代码如下（我已经删除了一些不相关的 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
}
}
}