Golang实现N秒内处理X任务并丢弃超额请求的并发方案咨询

原代码核心问题 #

1. 结构体定义错误：NewPool中引用了未定义的size字段；
2. 队列初始化错误：queue被初始化为长度等于maxSize的切片，导致len(p.queue)一开始就达到上限，无法添加新任务；
3. 逻辑完全偏离需求：每次收到任务就清空队列并强制停止定时器，完全违背了「N秒内攒满X个任务再处理」的设计目标；
4. 无缓冲通道阻塞：主程循环调用Add时，因ch无缓冲会被持续阻塞，无法实现「超出请求直接丢弃」；
5. Timer处理不当：重置定时器前未正确排空已触发的通道，易引发goroutine泄漏。

修正后的实现代码 #

package main

import (
	"fmt"
	"sync"
	"time"
)

type Pool struct {
	maxSize int           // 批处理最大任务数
	queue   []interface{} // 任务队列
	window  time.Duration // 时间窗口
	timer   *time.Timer   // 定时器
	ch      chan interface{} // 任务接收通道
	mu      sync.Mutex    // 队列互斥锁，保证并发安全
}

func NewPool(maxSize int, t int32) *Pool {
	window := time.Duration(t) * time.Second
	p := &Pool{
		maxSize: maxSize,
		queue:   make([]interface{}, 0, maxSize), // 初始化空切片，容量设为maxSize
		window:  window,
		timer:   time.NewTimer(window),
		ch:      make(chan interface{}, maxSize), // 带缓冲通道，缓冲等于maxSize，超出直接丢弃
	}
	go p.schedule()
	return p
}

// Add 非阻塞添加任务，超出缓冲或队列容量的直接丢弃
func (p *Pool) Add(ele interface{}) {
	select {
	case p.ch <- ele:
	default:
		fmt.Println("任务超出处理上限，已丢弃:", ele)
	}
}

func (p *Pool) schedule() {
	for {
		select {
		case <-p.timer.C:
			// 时间窗口到期，处理当前队列任务
			p.flush()
			p.timer.Reset(p.window)
		case data := <-p.ch:
			p.mu.Lock()
			if len(p.queue) < p.maxSize {
				p.queue = append(p.queue, data)
			} else {
				fmt.Println("队列已满，已丢弃任务:", data)
				p.mu.Unlock()
				continue
			}
			// 队列满时立即触发批量处理
			if len(p.queue) == p.maxSize {
				p.mu.Unlock()
				p.flush()
				// 重置定时器前确保通道已排空
				if !p.timer.Stop() {
					select {
					case <-p.timer.C:
					default:
					}
				}
				p.timer.Reset(p.window)
			} else {
				p.mu.Unlock()
			}
		}
	}
}

func (p *Pool) flush() {
	p.mu.Lock()
	if len(p.queue) == 0 {
		p.mu.Unlock()
		return
	}
	// 复制任务队列并清空原队列，避免处理期间阻塞新任务添加
	tasks := make([]interface{}, len(p.queue))
	copy(tasks, p.queue)
	p.queue = p.queue[:0]
	p.mu.Unlock()

	// 模拟任务处理逻辑
	fmt.Println("=== 开始处理批量任务，共", len(tasks), "个 ===")
	for _, task := range tasks {
		fmt.Println("处理任务:", task)
		time.Sleep(500 * time.Millisecond)
	}
	fmt.Println("=== 批量任务处理完成 ===")
}

func main() {
	p := NewPool(5, 2) // 2秒时间窗口，最多攒5个任务
	for i := 0; i < 20; i++ {
		p.Add("xyz " + fmt.Sprint(i))
		time.Sleep(300 * time.Millisecond) // 模拟任务间隔发送
	}
	// 等待剩余任务处理完成
	time.Sleep(3 * time.Second)
}

关键实现要点 #

• 并发安全：用sync.Mutex保护任务队列，避免多goroutine修改时的竞态问题；
• 非阻塞丢弃：Add方法通过select+default实现非阻塞发送，超出通道缓冲的任务直接丢弃；
• 双触发机制：满足「队列满」或「时间窗口到期」任一条件时，立即触发批量处理；
• Timer安全重置：重置定时器前先尝试停止，并用select排空可能已触发的通道，避免goroutine泄漏；
• 队列处理优化：flush时先复制任务再清空原队列，保证处理任务期间新任务可正常添加。

内容的提问来源于stack exchange，提问作者Arpit