深入解析 client-go 中的 WorkQueue

后端

2023-09-03 00:11:09

在 Kubernetes 等复杂系统中，高效处理并行任务至关重要。client-go 库中的 WorkQueue 是一种强大的工具，可帮助开发者管理并行任务，确保有序和可靠的处理。

<#section>WorkQueue 简介</#section>

WorkQueue 是一种队列结构，用于存储需要处理的任务。它提供了一种机制来添加、获取和处理任务，同时确保每个任务只被一个工作进程处理一次。

<#subsection>Queue 接口</#subsection>

WorkQueue 实现了 Queue 接口，该接口定义了三个主要方法：

add(item)： 将一个新项添加到队列中。
get()： 从队列中获取一个项进行处理。
done(item)： 在项处理完成后将其从队列中移除。

<#subsection>DelayingQueue</#subsection>

WorkQueue 还提供了一个 DelayingQueue 接口，它允许将项添加到队列中并指定延迟时间。在延迟时间过后，该项才会可用于处理。

<#section>使用 WorkQueue</#section>

使用 WorkQueue 管理并行任务通常涉及以下步骤：

创建 WorkQueue 实例： 使用 NewRateLimitingQueue() 或 NewDelayingQueue() 创建一个 WorkQueue 实例。
添加任务： 使用 Add() 方法将任务添加到队列中。任务可以是任何类型的值，但通常是包含要执行的工作的结构。
处理任务： 每个工作进程应从队列中调用 Get() 方法获取任务。在任务处理完成后，应调用 Done() 方法将其从队列中移除。
关闭队列： 在处理完所有任务后，应调用 ShutDown() 方法关闭队列。

<#section>示例</#section>

以下示例展示了如何使用 WorkQueue 管理并行任务：

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	clientgosharedinformers "k8s.io/client-go/informers"
	clientgopkgapi "k8s.io/client-go/pkg/api"
	clientgopkgapiscore "k8s.io/client-go/pkg/apis/core"
	clientgoqueue "k8s.io/client-go/util/workqueue"
)

func main() {
	// 创建一个新的 WorkQueue。
	workQueue := clientgoqueue.NewRateLimitingQueue(clientgoqueue.NewMaxOfRateLimiter(time.Second, 10))

	// 为 Pod 事件创建 Informer。
	informerFactory := clientgosharedinformers.NewSharedInformerFactory(nil, time.Minute)
	podInformer := informerFactory.Core().V1().Pods()

	// 注册一个处理 Pod 事件的处理函数。
	podInformer.Informer().AddEventHandler(clientgoqueue.FilteringResourceEventHandler{
		Filter: clientgoqueue.EventFilterFromWatch,
		Handler: clientgoqueue.EnqueueWork(workQueue, &clientgoqueue.RateLimitingInterface{
			MaxConcurrency:  10,
			MaxPending:      2000,
			DelayBetween:    time.Second,
			Handle:         processPodEvent,
		}),
	})

	// 启动 Informer。
	informerFactory.Start(context.Background().Done())

	// 运行工作队列。
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		defer wg.Done()
		workQueue.Run(context.Background())
	}()

	// 等待 Informer 和工作队列退出。
	informerFactory.WaitForCacheSync(context.Background().Done())
	workQueue.ShutDown()
	wg.Wait()
}

func processPodEvent(key string) error {
	fmt.Printf("处理 Pod 事件：%s\n", key)
	return nil
}