概述

摘要：本文从源码层面解读了kubernetes源码中常用的workqueue.ParallelizeParallelizeUntil()框架的源码实现，并且本文也将举例说明了workqueue.ParallelizeUntil()方法的典型使用场景。

正文

说明：基于 kubernetes v1.18.0 源码分析

在Kubernetes源码中, 我们经常会读到workqueue.ParallelizeUntil()函数，它的作用是在并发多个worker来处理任务，直到接收到context发出的停止信号或任务完成。workqueue.ParallelizeUntil源码位于k8s.io/client-go/util/workqueue/parallelizer.go，该workqueue是client-go中的一个工作队列，队列包括三种：FIFO、延迟队列和限速队列。关于workqueue我已经在之前的文章中有详细的介绍，如需了解请阅览informer中的WorkQueue机制的实现分析与源码解读(1)

ParallelizeUntil()的源码解读

ParallelizeUntil()方法作用是并发多个worker来处理任务，直到接收到context发出的停止信号或任务完成

workers 表示启动多少个worker并发处理任务

pieces 表示要处理任务对应的index的数量

DoWorkPieceFunc 表示用于处理任务的工作函数

ctx 使用context控制并发任务的停止

// 定义worker函数
type DoWorkPieceFunc func(piece int)// ParallelizeUntil is a framework that allows for parallelizing N 
// independent pieces of work until done or the context is canceled.
// parallelelizeuntil是一个框架，它允许并行处理任务，直到完成或上下文被取消。
func ParallelizeUntil(ctx context.Context, workers, pieces int, doWorkPiece DoWorkPieceFunc) {// 定义一个stop信号，当手动ctx.Done()信号后，就让整个 ParallelizeUntil 任务停止var stop <-chan struct{}if ctx != nil {stop = ctx.Done()}// 将把需要多的工作对应的索引放到chan中，放入完成后就关闭chan，这里借助chan,是因为chan是线程安全的。toProcess := make(chan int, pieces)for i := 0; i < pieces; i++ {toProcess <- i}close(toProcess)// 对于要处理的对象数量少于worker数量时if pieces < workers {workers = pieces}wg := sync.WaitGroup{}wg.Add(workers)// 启动多个worker，同时从chan中取对象进行处理，知道收到停止信号或完成任务for i := 0; i < workers; i++ {go func(x int) {defer utilruntime.HandleCrash()defer wg.Done()for piece := range toProcess {fmt.Printf("work %d  ",x)select {case <-stop:returndefault:doWorkPiece(piece)}}}(i)}wg.Wait()
}

从源码分析知道，ParallelizeUntil()的逻辑比较简单，先将要处理的任务，放到chan中，放入完成后就关闭chan，这里借助chan,是因为chan是线程安全的。之后再启动多个worker协程，这些协程不断的从chan去job进行处理。逻辑示意图如下。

如果在多想象一下，ParallelizeUntil()的逻辑，是不是与工厂中多个机器人不断的从传输带上去东西来进行加工的场景啊。

接下来我们再看下，worker协程中捕获panic的HandleCrash的源码。源码逻辑比较简单，就是worker协程中如果出现panic将被recover捕获，捕获之后如果定义了额外的handler函数，会遍历执行。

var PanicHandlers = []func(interface{}){logPanic}// PanicHandlers is a list of functions which will be invoked when a panic happens.
var PanicHandlers = []func(interface{}){logPanic}// HandleCrash simply catches a crash and logs an error. Meant to be called via
// defer.  Additional context-specific handlers can be provided, and will be
// called in case of panic.  HandleCrash actually crashes, after calling the
// handlers and logging the panic message.
//
// E.g., you can provide one or more additional handlers for something like shutting down go routines gracefully.
func HandleCrash(additionalHandlers ...func(interface{})) {// 捕获panicif r := recover(); r != nil {for _, fn := range PanicHandlers {fn(r)}for _, fn := range additionalHandlers {fn(r)}if ReallyCrash {// Actually proceed to panic.panic(r)}}
}

代码测试

编写一个简单程序，测试workqueue.ParallelizeUntil()方法的使用.

需求： 通过workqueue.ParallelizeUntil()，并发的找出100以内的素数。

import ("context""fmt""k8s.io/client-go/util/workqueue""testing""time"
)// 判断一个数是否是素数
func isPrieme(num int) bool {for i := 2; i < num; i++ {if num%i == 0 {return false}}return true
}// 定义用于保存结果的chan
var ResultChan = make(chan int, 100)func WorkFunc(num int) {fmt.Println(" check num: ", num)if isPrieme(num) {ResultChan <- num}
}// go test -mod=vendor -run="^TestParallelizeUntil" -v
func TestParallelizeUntil(t *testing.T) {// 定义超时信号对应的ctxctx, _ := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))// 判断一个数是否是素数myFunc := WorkFunc// 执行查找workqueue.ParallelizeUntil(ctx, 3, 100, myFunc)// 将结果放入结果chanfmt.Println("result:")close(ResultChan)for i := range ResultChan {fmt.Printf("%d ", i)}fmt.Println()}---------------执行与输出如下-----------------
(base) @PM80280051 ➜ utils git:(dev-xw) ✗ go test  -run="^TestParallelizeUntil" -v
=== RUN   TestParallelizeUntilcheck num:  0check num:  3check num:  4check num:  5check num:  6check num:  7check num:  8check num:  9check num:  10check num:  11check num:  12check num:  13check num:  14check num:  15check num:  16check num:  17check num:  1check num:  19check num:  20check num:  21check num:  22check num:  23check num:  24check num:  25check num:  26check num:  27check num:  28check num:  29check num:  30check num:  31check num:  32check num:  2check num:  34check num:  33check num:  36check num:  37check num:  38check num:  39check num:  40check num:  41check num:  42check num:  43check num:  44check num:  45check num:  46check num:  47check num:  48check num:  49check num:  35check num:  51check num:  52check num:  53check num:  54check num:  55check num:  56check num:  57check num:  58check num:  59check num:  60check num:  61check num:  62check num:  63check num:  64check num:  65check num:  66check num:  67check num:  68check num:  69check num:  70check num:  18check num:  72check num:  71check num:  74check num:  75check num:  76check num:  77check num:  78check num:  79check num:  80check num:  81check num:  82check num:  50check num:  84check num:  85check num:  86check num:  87check num:  88check num:  89check num:  90check num:  91check num:  92check num:  93check num:  94check num:  95check num:  96check num:  97check num:  83check num:  99check num:  98check num:  73result: 0 3 5 7 11 13 17 1 19 23 29 31 2 37 41 43 47 53 59 61 67 71 79 89 97 83 73 
--- PASS: TestParallelizeUntil (0.00s)
PASS
ok      kubecmdb/utils  0.552s

在这个例子中，ParallelizeUntil 函数，启动了3个worker，并发的处理10个任务，直到任务处理完，或者接收到ctx定义的超时信号。 注意，ParallelizeUntil 函数不会返回任何值，如果需要记录worker协程中的结果，可以自己定义。`

使用场景

在kubernetes源码中，Kube-scheduler在进行预选算法时，使用了workqueue.ParallelizeUntil()，并发16个worker同时执行预选算法。在集群规模较大时，并发处理能提升kube-scheduler调度任务的整体效能。

预选算法执行预选的源码findNodesThatPassFilters

// k8s.io/kubernetes/pkg/scheduler/core/generic_scheduler.go// findNodesThatPassFilters finds the nodes that fit the filter plugins.
// findNodesThatPassFilters使用过滤插件来查找适合的节点。
func (g *genericScheduler) findNodesThatPassFilters(ctx context.Context, prof *profile.Profile, state *framework.CycleState, pod *v1.Pod, statuses framework.NodeToStatusMap) ([]*v1.Node, error) {allNodes, err := g.nodeInfoSnapshot.NodeInfos().List()if err != nil {return nil, err}numNodesToFind := g.numFeasibleNodesToFind(int32(len(allNodes)))// Create filtered list with enough space to avoid growing it// and allow assigning.filtered := make([]*v1.Node, numNodesToFind)// 代码略// Stops searching for more nodes once the configured number of feasible nodes// are found.// ！！！重点，启动了16个work，并发的执行 checkNode 函数，对nodes节点进行预选操作workqueue.ParallelizeUntil(ctx, 16, len(allNodes), checkNode)processedNodes := int(filteredLen) + len(statuses)g.nextStartNodeIndex = (g.nextStartNodeIndex + processedNodes) % len(allNodes)// filter 列表预选算法，过滤出满足预选算法的node节点filtered = filtered[:filteredLen]if err := errCh.ReceiveError(); err != nil {statusCode = framework.Errorreturn nil, err}return filtered, nil
}

结论

workqueue.ParallelizeUntil框架广泛的适用于kubernetes源码，通过对其源码的解读，我们了解到了其如何实现与使用场景。我们可以在平时日常开发中，也可以多多尝试使用这个成熟的并发任务框架。

我们可以通过阅读kubernetes源码，学习Kubernetes内部机制，同时Kubernets项目中有一些好的成熟框架，我们可以学以致用，多在用在日常开发中才能真正掌握。

参考资料

Kubernete-v1.18源码