Go Pond 怎样设置有限队列并拒绝超额任务

如果任务入口不能无限接收请求，Go Pond v2 可以把“正在运行多少个任务”和“最多允许多少个任务等待”拆开设置：用 pond.NewPool(2, pond.WithQueueSize(2)) 保留两个运行槽和两个排队槽，再用 TrySubmitErr 做非阻塞提交。返回值里的 ok=false 就表示本次任务没有进入池，调用方可以立即返回、降级或稍后重试。

官方地址：https://github.com/alitto/pond

这个组合适合批量抓取、限流调用和后台消费这类场景：系统愿意承受一个明确大小的短队列，但不愿意让突发流量把内存和下游连接一起拖垮。

先把四个容量位置分清楚

Pond 的并发参数和队列参数解决的是两件事。NewPool(2) 约束同时工作的 worker 数；WithQueueSize(2) 约束已经被池接收、但还没开始执行的任务数。两者都满时，才进入“超额提交”的处理分支。

用 TrySubmitErr 在队列满时立即拒绝

下面的例子模拟批量抓取任务。池只有两个运行槽和两个等待槽，第五个及之后的提交不会把当前 goroutine 卡住，而是通过 ok 把接收结果交给调用方。任务本身仍然返回 error，方便处理执行阶段的失败；“有没有被池接收”则由第二个返回值负责。

# 创建一个最小示例模块，并固定使用 Pond v2 的导入路径。
mkdir pond-bounded-demo
cd pond-bounded-demo
go mod init example.com/pond-bounded-demo
go get github.com/alitto/pond/v2

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/alitto/pond/v2"
)

func fetch(id int) error {
	// 用短暂延迟模拟一次外部请求，真实代码应在这里传递超时上下文。
	time.Sleep(80 * time.Millisecond)
	if id == 2 {
		return errors.New("upstream rejected task")
	}
	return nil
}

func main() {
	// 2 个任务运行时槽位 + 2 个等待槽位，超额提交不应阻塞入口。
	pool := pond.NewPool(2, pond.WithQueueSize(2))
	defer pool.StopAndWait() // 无论提交结果如何，都等待已接收任务收尾。
	accepted := make([]pond.Task, 0, 4)

	for id := 1; id 

这里的容量不是“最多只允许四次调用”。任务完成后，运行槽会再次可用，后续提交仍然可以进入池。ok=false 只代表提交瞬间没有空余容量，业务层应决定是丢弃、返回忙碌，还是把原任务放到外部消息系统。

三种提交策略应该怎样选

有界队列只定义了容量，满载时怎么处理还要看业务。Pond v2 同时提供按次尝试和全局非阻塞两种方向；默认提交则会在队列没有空间时等待。

// pool-wide non-blocking：整个池在队列满时都采用快速失败。
pool := pond.NewPool(2,
	pond.WithQueueSize(2),
	pond.WithNonBlocking(true),
)

task := pool.SubmitErr(func() error {
	// 任务进入池后才会执行这里的业务逻辑。
	return fetch(10)
})
if err := task.Wait(); err != nil {
	// 队列满时这里可能得到 pond.ErrQueueFull；执行失败也会走这里。
	fmt.Println("task unavailable or failed:", err)
}

如果只有一个入口需要快速失败，优先用 TrySubmitErr，因为接收结果更直接；如果所有调用方都希望统一处理队列满错误，可以配置 WithNonBlocking(true)，同时按 errors.Is 或项目约定把排队失败和任务执行失败分开统计。

零队列与小队列的取舍

WithQueueSize(0) 会关闭等待队列。此时任务要么直接被运行中的 worker 接收，要么在使用 TrySubmit、TrySubmitErr 或非阻塞选项时立即失败。这种模式的延迟更容易预测，但突发流量的拒绝比例也会更高。

设置为 1 或 2 的小队列，通常给短暂抖动留出一点缓冲；它并不会让下游处理能力变强，只是把一部分瞬时压力从入口转移到了受控的等待区。队列大小应和任务平均耗时、入口峰值、可接受延迟一起决定，不要只按机器核数拍一个数字。

用指标确认池确实按预期工作

容量配置完成后，建议把指标接到现有监控，而不是只看日志中的“拒绝”字样。RunningWorkers() 反映正在执行的 worker，WaitingTasks() 反映当前排队数，DroppedTasks() 反映因队列满而被丢弃的任务累计数。它们回答的是不同问题。

func report(pool pond.Pool) {
	// 这些数字用于区分并发占用、当前积压和历史丢弃量。
	fmt.Printf(
		"running=%d waiting=%d dropped=%d\\n",
		pool.RunningWorkers(),
		pool.WaitingTasks(),
		pool.DroppedTasks(),
	)
}

应用退出或配置热更新时，使用 StopAndWait 等待已经接收的任务完成；不要把“拒绝新任务”和“取消已经运行的任务”混成一个动作。需要取消长任务时，还要让任务函数主动监听上下文。

落地时保留这份容量清单

1. 先写清最多同时运行多少个任务，再写清最多允许等待多少个任务。
2. 需要快速失败时使用 TrySubmitErr，把 ok=false 转成明确的业务结果。
3. 统一非阻塞时使用 WithNonBlocking(true)，并单独统计 pond.ErrQueueFull。
4. 用 WaitingTasks 观察积压，用 DroppedTasks 观察拒绝趋势，不用单个日志推断容量。
5. 退出前调用 StopAndWait，确保已接收任务有清晰的收尾路径。

常见追问

WithQueueSize(2) 是总容量还是只计算等待任务？

它描述的是队列容量，不包含当前正在运行的 worker。示例中的总接收量可以理解为“运行槽 + 等待槽”，但任务完成后运行槽会循环复用。

任务被拒绝后应该自动重试吗？

不应在池内无条件自旋重试。可以把拒绝交给带退避的外部队列、返回可重试状态，或按业务优先级丢弃低价值任务。

为什么不直接使用无限队列？

无限队列能减少入口阻塞，却会把突发流量变成持续积压，最终表现为内存、延迟或下游连接压力。有限队列让系统在过载时更早给出可观测的拒绝信号。

总结一下：先用 NewPool 限制同时运行数，再用 WithQueueSize 限制等待数；需要“满了就返回”时，按次使用 TrySubmitErr，全局使用 WithNonBlocking(true)。真正的容量边界还要配合指标和退出收尾一起落地。