go select语句的强制优先级

Posted 2023-03-14

【中文标题】go select语句的强制优先级

【英文标题】：Force priority of go select statement

【发布时间】：2018-02-22 07:55:57

【问题描述】：

我有以下代码：

func sendRegularHeartbeats(ctx context.Context) 
    for 
        select 
        case <-ctx.Done():
            return
        case <-time.After(1 * time.Second):
            sendHeartbeat()
        
    

这个函数在一个专门的 go-routine 中执行并且每秒发送一个心跳消息。当上下文被取消时，整个过程应该立即停止。

现在考虑以下场景：

ctx, cancel := context.WithCancel(context.Background())
cancel()
go sendRegularHeartbeats(ctx)

这会以封闭的上下文启动心跳例程。在这种情况下，我不希望传输任何心跳。所以应该立即输入select中的第一个case块。

但是，似乎无法保证评估 case 块的顺序，并且代码有时会发送心跳消息，即使上下文已经取消。

实现这种行为的正确方法是什么？

我可以在第二个 case 中添加一个“isContextclosed”检查，但这看起来更像是一个丑陋的解决方法。

【问题讨论】：

虽然标题和***.com/questions/11117382/…很相似，但这不是重复的。另一个问题的解决方案处理的是消费者/生产者问题，而不是上下文。

请参阅本文中的第 2 点。 tapirgames.com/blog/golang-concurrent-select-implementation ，您所经历的似乎是正常的 select 行为。如果您想要更多控制权，您可能必须在没有上下文的情况下重新实现并继续使用consumer/producer 解决方案

问题不一样，但答案是：先在ctx.Done上选择，如果没有取消（也选择），默认情况下发送心跳。环形。简单的。您不能强制选择优先级。不要再看下去了，这是可以撤消的。你必须嵌套选择。

@Volker 好点，发送到默认值应该可以解决问题

@Volker 您可以通过代码示例将此作为答案发布吗？我真的不明白这应该如何工作。如果上下文在 1 秒超时期间关闭，我也想中止（超时是可配置的，可以是一分钟或更长时间）

【参考方案1】：

如果保持操作的优先级绝对关键，您可以：

在单独的 goroutine 中从每个通道消费 让每个 goroutine 向共享的第三个通道写入一条消息，指明其类型 让第三个 goroutine 从该通道消费，读取它接收到的消息以确定它是一个滴答，应该sendHeartbeat，或者它是一个取消，它应该退出

这样，在其他通道上接收到的消息将（可能 - 你不能保证并发例程的执行顺序）按照它们被触发的顺序进入第三个通道，允许您可以适当地处理它们。

然而，值得注意的是，这可能不是必需的。 select 不能保证如果多个案例同时成功会执行哪个case。这可能是一个罕见的事件；在select 处理任何一个之前，cancel 和ticker 都必须触发。大多数情况下，在任何给定的循环迭代中，只有一个或另一个会触发，因此它的行为将完全符合预期。如果您可以容忍在取消后触发一个额外的心跳的罕见情况，那么您最好保留现有的代码，因为它更有效且更具可读性。

【讨论】：

我理解你的回答的实际性质，但承认“比赛条件有时发生的频率并不高，所以没关系”并不能在口中留下很好的味道。

@dcow 我从来没有说过，所以我不确定你指的是什么？这里没有数据竞争，所有 goroutine 之间的通信都是通过通道同步的。

我想我无法消化你的最后一段。严格来说，在取消定时器后触发它是不正确的，我很奇怪，即使定时器在不应该触发的时候触发，你也会设计可能没问题的代码。我一直在这样的大型代码库中，它开始让你发疯。我知道防御性编码是实用的，但你的回答实际上并没有展示一种方法来做 OP 所要求的，（你的 probably 有序频道并没有真正帮助任何事情，因为它只是移动问题）。

不能保证并发操作的顺序。您可以使用锁定强制它们按顺序排列，但您仍然无法强制它们的顺序。如果你想要严格有序的操作，它们不应该是并发的。

我认为我们在这一点上没有分歧。然而，问题不在于严格的排序。问题是，“如果我有两个通道都准备好使用数据，我如何将一个通道优先于另一个通道？”这个问题与实际发生的事件无关首先，正如您所指出的，这通常没有意义，因为并发事件实际上都是同时发生的。

【参考方案2】：

事先说明：

您的示例将按照您的预期工作，就好像在调用 sendRegularHeartbeats() 时上下文已经取消一样，case <-ctx.Done() 通信将是唯一准备好继续并因此被选中的通信。另一个case <-time.After(1 * time.Second) 只会在1 秒后 准备好继续，所以一开始不会被选中。但要在可能准备好多个案例时明确处理优先级，请继续阅读。

---

与switch statement 的case 分支不同（评估顺序是它们列出的顺序），select statement 的case 分支中没有优先级或任何顺序保证。

引用Spec: Select statements:

如果一个或多个通信可以继续，通过统一的伪随机选择选择一个可以继续的通信。否则，如果存在默认情况，则选择该情况。如果没有默认情况，“select”语句会阻塞，直到至少有一个通信可以继续。

如果可以进行更多通信，则随机选择一个。期间。

如果您想保持优先级，您必须自己（手动）执行此操作。您可以使用多个 select 语句（后续的，不是嵌套的）来执行此操作，在 earlier select 中列出具有更高优先级的语句，还请务必添加 default 分支以避免阻塞如果这些还没有准备好继续。您的示例需要 2 个 select 语句，第一个检查 <-ctx.Done() 因为这是您想要更高优先级的语句。

我还建议使用单个 time.Ticker 而不是在每次迭代中调用 time.After()（time.After() 在后台也使用 time.Ticker，但它不会重复使用它只是“扔掉它”并创建下次通话时换一个新的）。

这是一个示例实现：

func sendRegularHeartbeats(ctx context.Context) 
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for 
        select 
        case <-ctx.Done():
            return
        default:
        

        select 
        case <-ctx.Done():
            return
        case <-ticker.C:
            sendHeartbeat()
        
    

如果在调用sendRegularHeartbeats() 时上下文已经取消，这将不会发送任何心跳，因为您可以在Go Playground 上检查/验证它。

如果您将 cancel() 调用延迟 2.5 秒，则将发送恰好 2 个心跳：

ctx, cancel := context.WithCancel(context.Background())
go sendRegularHeartbeats(ctx)
time.Sleep(time.Millisecond * 2500)
cancel()
time.Sleep(time.Second * 2)

在Go Playground 上试试这个。

【参考方案3】：

接受的答案有一个错误的建议：

func sendRegularHeartbeats(ctx context.Context) 
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for 
        //first select 
        select 
        case <-ctx.Done():
            return
        default:
        

        //second select
        select 
        case <-ctx.Done():
            return
        case <-ticker.C:
            sendHeartbeat()
        
    

这无济于事，因为以下情况：

两个频道都是空的 首先选择运行 两个通道同时收到一条消息 你在同一个概率游戏中，就好像你在第一次选择中什么都没做一样

另一种但仍然不完美的方法是在使用代码事件后防止并发的 Done() 事件（“错误选择”），即

func sendRegularHeartbeats(ctx context.Context) 
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()

    for             
        //select as usual
        select 
        case <-ctx.Done():
            return
        case <-ticker.C:
            //give priority to a possible concurrent Done() event non-blocking way
            select 
              case <-ctx.Done():
              return
            default:
            
            sendHeartbeat()
        
    

警告：这个问题是它允许混淆“足够接近”的事件 - 例如即使一个ticker 事件更早到达，Done 事件也很快到来，抢占了心跳。目前还没有完美的解决方案。

【讨论】：

你是对的 - 如果调用 sendRegularHeartbeats 并且 - 稍后 - 两个通道同时关闭，则接受的答案将不起作用。但是，如果在调用 sendRegularHeartbeats 之前关闭两个通道，它将起作用 - 这正是我的具体问题。

在这种情况下（注意事项），我们可以使用标志 sendHeartbeat() 来停放 Done() 的到达，因为我们已经满足了代码，然后立即处理完成。