Golang - 在 N 单位时间内执行 X 任务

Question

我正在尝试模拟一个程序，它需要在 N 秒内执行任务中的 X 任务并丢弃其他工作请求当收到 on 的值时，我试图timer在无限循环中使用 with select <-timer.C，我正在继续模拟任务，其中包含切片中的所有数据，这些数据受到保护maxLimit并再次将其重置timer为初始持续时间这是代码     type Pool struct {            // The maximum number of items allowed in pool            maxSize int            //the queue to hold the data            queue []interface{}            // time window for this pool            time time.Duration            //timer for the batch            timer *time.Timer            //channel            ch chan interface{}        }                func NewPool(maxSize int, t int32) *Pool {            p := &Pool{                maxSize: maxSize,                size:    0,                queue:   make([]interface{}, maxSize),                time:    time.Duration(t * int32(time.Second)),                timer:   time.NewTimer(time.Duration(t) * time.Second),                ch:      make(chan interface{}),            }            go p.Schedule()            return p        }                func (p *Pool) Add(ele interface{}) {            p.ch <- ele        }                func (p *Pool) Schedule() {            for {                select {                case <-p.timer.C:                    fmt.Println("Time is over")                    p.queue = make([]interface{}, 0)                    p.timer.Reset(p.time)                    p.flush()                case data := <-p.ch:                    if len(p.queue) < p.maxSize {                        fmt.Println("Addding")                        p.queue = append(p.queue, data)                            }                    //p.flush()                    if !p.timer.Stop() {                        <-p.timer.C                    }                    p.queue = make([]interface{}, 0)                        }                    }                }        但这没有按预期工作我在这里遗漏了一些东西，你能指导我使用并发模式来满足这个要求吗，谢谢

Answer 1

添加一些额外的输出可以帮助您定位问题：

case data := <-p.ch:

    fmt.Println("Got Data", len(p.queue), p.maxSize)

    if len(p.queue) < p.maxSize {

        p.queue = append(p.queue, data)

    }

    if !p.timer.Stop() {

        fmt.Println("Draining")

        <-p.timer.C

    }

    p.queue = make([]interface{}, 0)

}

使用该更改运行输出为：

Got Data 5 5

Got Data 0 5

Addding

Draining

因此正在处理两条消息（第一条不会输出，Adding因为len(p.queue)它是 5；这是因为您将其初始化为大小为 5 - make([]interface{}, maxSize)）。考虑收到消息时代码在做什么：

1. 处理queue

2. 停止计时器

3. 翻拍queue

现在从文档中获取timer.Stop()：

Stop 阻止 Timer 触发。如果调用停止计时器，它返回 true ，如果计时器已经过期或停止，则返回 false 。

在第一次迭代中，它起作用了（计时器停止，如有必要，通道被耗尽）。但是，您在停止计时器后不会重置计时器，因此在第二次迭代时，当您调用时计时器已经停止p.timer.Stop()。这意味着Stop()它将返回false（计时器已经停止！）并且您尝试耗尽通道（但由于计时器已经停止，这将永远阻塞）。

如何解决这个问题取决于你的目标；我怀疑你是想重置定时器？如果没有，你可以这样做：

if p.timer.C != nil && !p.timer.Stop() {

    fmt.Println("Draining")

    <-p.timer.C

}

p.timer.C = nil // Timer has been stopped (nil channel will block forever)