Select实现优先级并发技巧

序言

Go语言中使用 Select 语句可以同时对多个case进行监听，它通常与多个 channel 一起使用去进行协程的资源监听。当 Select 的其中一个 case 满足条件时，Select 就会执行对应的 case 分支。

这里我们写一个简单的程序去运行看看他的特性：

func main() {
    numList := make(chan int, 5)
    strList := make(chan string, 1)
    go func() {
        for i := 0; i < 5; i++ {
            numList <- i
        }
        strList <- fmt.Sprintf("曼波曼波")
    }()
    for {
        select {
        case num := <-numList:
            fmt.Println(num)
        case str := <-strList:
            fmt.Println(str)
            return
        }
    }
}

运行结果：

# 第一次运行结果：
0
1
曼波

# 第二次运行结果：
0
曼波

# 第三次运行结果：
0
1
2
曼波

# 第四次运行结果：
0
1
2
3
4
曼波

我们会发现上面的运行结果基本每次都不一样，这是因为 Select 的特性导致的，Select 和 Switch 不同，它并不按照代码书写的顺序去进行判断分支条件是否满足，而是会（伪）随机的选择一个满足条件的 case 分支进行执行。

因此上面的代码并不是先打印完 numList 再打印 strList。上面代码开启了两个协程，因为子协程将两个 channel 需要的值都赋予了给它们，但是当主协程去进行 Select 读取的时候，因为两个 case 都满足条件，会触发 Select 的（伪）随机特性，使得每次运行的结果都可能不一样。

因此，想要在多个满足条件的 case 上先运行我们想要的 case，我们就需要给这部分加个优先级。

优先级

例如我们期望先打印完 numList 再打印 strList 后结束，我们就可以这样写：

func main() {
    numList := make(chan int, 5)
    strList := make(chan string, 1)
    go func() {
        for i := 0; i < 5; i++ {
            numList <- i
        }
        strList <- fmt.Sprintf("曼波曼波")
    }()
    for {
        select {
        case num := <-numList:
            fmt.Println(num)
        case str := <-strList:
            for {
                select {
                case num := <-numList:
                    fmt.Println(num)
                default:
                    fmt.Println(str)
                    return
                }
            }
        }
    }
}

输出结果：

# 输出结果：
0
1
2
3
4
曼波曼波

这样，使用双层 for 循环嵌套去判断就行了，里层 for 循环的 Select 选择，当 num 的 case 满足条件时，就运行它，不然就运行 str 的默认 case 。因为之前一个 Select 里面两个 case 会存在随机，当里层用了 default 之后，有其他的 case ，就会优先选择其他 case 而不是 default 的内容啦。

额外

具体K8s中也有利用 Select 优先级去处理 node 与 pod 的优先级更新，优先更新 node 再更新 pod。

for {
    select {
    case <-stopCh:
        return
    case nodeUpdate := tc.nodeUpdateChannels[workers]:
        tc.handleNodeUpdate(ctx, nodeUpdate)
        tc.nodeUpdateQueue.Done(nodeUpdate)
    case podUpdate := <- tc.podUpdateChannels[works]:
        //If we found a Pod update we need to empty Node queue first.
    pirority:
        for {
            select {
            case nodeUpdate := tc.nodeUpdateChannels[workers]:
                tc.handleNodeUpdate(ctx, nodeUpdate)
                tc.nodeUpdateQueue.Done(nodeUpdate)
            default:
                // 跳出pirority外，跳出到for循环
                // 如果不这样做，只能break出下一次的Select
                break pirority 
            }
        }
        // After Node queue is emptied we process podUpdate.
        tc.handlePodUpdate(ctx, podUpdate)
        tc.PodUpdateQueue.Done(podUpdate)
    }
}

这里因为要一直轮询优先去查 node 是否更新，高优先级的 node 正常更新，当没有高优先级的 node 准备就绪了，K8s就需要保证清空完 node 的升级，再去更新 pod。（意思就是始终优先更新node）

所以里侧 for 循环如果没有 node 就绪更新，要跳出再去更新对应的 pod ，之后触发下一次外层 for 循环一直轮询优先查找 node 是否更新。

总结

至此，这就是Go语言中 Select 的一个优先级的并发技巧。我也只是听别人的讲解学的，感觉有用就记录了下来。