限制来自通道的已处理消息的数量

Question

我每秒通过通道收到大约 200 000 条消息到我的工作人员，我需要将发送给客户端的消息数量限制为每秒 20 条。 这使它每 50 毫秒 1 条消息

在 LOOP 的帮助下，worker 将在整个程序生命周期内仍然活着（而不是为每条消息打开一个通道）。

我的目标： - 由于消息的顺序很重要，我想跳过在阻塞的 50 毫秒内收到的所有消息，只保存最新的消息 - 如果最新的消息在阻塞的 50 毫秒内出现，我希望在循环内的阻塞时间结束并且没有新消息到来时处理保存的消息！

我的策略 - 不断向同一频道发送尚未处理的最新消息

但它的问题是，如果该消息是在（来自应用程序的）新消息之后发送的呢？

下面的代码更像是一个作为工作代码的算法，只是想要一个关于如何做的提示/方法。

func example (new_message_from_channel <-chan *message) {
    default = message
    time = now_milliseconds
    diff_accepted = 50milli
    for this_message := range new_message_from_channel {
        if now_millisecond -  time >= diff_accepted {
            send_it_to_the_client
            time = now_milliseconds
        } else {
            //save the latest message
            default = this_message

            //My problem is how to process this latest message when the blocked 50ms is over and no new message coming ?!

            //My strategy - keep sending it to the same channel
            theChannel <- default
        }

    }
}

如果你有优雅的方法，欢迎与我分享:)

Answer 1

使用速率限制器，您可以创建一个throttle 函数，该函数将采用：速率和通道作为输入；并返回两个频道 - 一个包含所有原始频道项目，另一个仅以固定速率中继项目：

func throttle(r time.Duration, in <-chan event) (C, tC <-chan event) {

    // "writeable" channels
    var (
        wC  = make(chan event)
        wtC = make(chan event)
    )

    // read-only channels - returned to caller
    C = wC
    tC = wtC

    go func() {
        defer close(wC)
        defer close(wtC)

        rl := rate.NewLimiter(
            rate.Every(r),
            1,
        )

        // relays input channel's items to two channels:
        // (1) gets all writes from original channel
        // (2) only writes at a fixed frequency
        for ev := range in {
            wC <- ev
            if rl.Allow() {
                wtC <- ev
            }
        }
    }()
    return
}

工作示例：https://play.golang.org/p/upei0TiyzNr

---

编辑：

为了避免使用速率限制器，而是使用简单的time.Ticker：

tick := time.NewTicker(r)

for ev := range in {
    select {
    case wC <- ev: // write to main
    case <-tick.C:
        wC <- ev  // write to main ...
        wtC <- ev // ... plus throttle channel
    }
}

工作示例：https://play.golang.org/p/UTRXh72BvRl