我如何等待频道活动的间歇触发某些事情？答案

【问题标题】：How do I wait for lulls in channel activity to trigger something?我如何等待频道活动的间歇触发某些事情？
【发布时间】：2016-06-29 00:55:09
【问题描述】：

我有一个可以接收到大量写入的通道。我想等到通道上的大量发送完成后再触发操作。

我看过这个gist，但是，如果缓冲区中有数据，它将每隔interval发送一次输出：

func debounceChannel(interval time.Duration, output chan int) chan int {
  input := make(chan int)

  go func() {
    var buffer int
    var ok bool

    // We do not start waiting for interval until called at least once
    buffer, ok = <-input 
    // If channel closed exit, we could also close output
    if !ok {
      return
    }

    // We start waiting for an interval
    for {
      select {
      case buffer, ok = <-input:
        // If channel closed exit, we could also close output
        if !ok {
          return
        }

      case <-time.After(interval):
        // Interval has passed and we have data, so send it
        output <- buffer
        // Wait for data again before starting waiting for an interval
        buffer, ok = <-input
        if !ok {
          return
        }
        // If channel is not closed we have more data and start waiting for interval
      }
    }
  }()

  return input
}

在我的情况下，我想等到输入通道上不再有任何数据发送到此突发，然后再触发或发送到输出。

我如何做到这一点？

【问题讨论】：

你的意思是把所有数据都输入到输入通道，然后一一读取到输出通道吗？
没有。我有一个输入通道，可以突发接收数据。我想从输入中读取数据，并且每次读取大量数据时只写入输出通道一次。
您能否扩展评论“等到所有写入完成”？是否打算使用延迟作为信号？
@Mark，我更新了我的问题并将其更改为“通道上的突发发送”。我找到了解决问题的方法，并发布了一些实现该问题的代码作为我的答案。

标签： go debouncing

【解决方案1】：

这就是我最终实现的去抖动器：

func Debounce(lull time.Duration, in chan struct{}, out chan struct{}) {

    go func() {

        var last int64 = 0

        for {
            select {
            case <-in:
                last = time.Now().Unix()

            case <-time.Tick(lull):
                if last != 0 && time.Now().Unix() >= last+int64(lull.Seconds()) {
                    last = 0
                    out <- struct{}{}
                }
            }
        }
    }()
}

这需要一段平静时间，即如果我们没有收到输入的持续时间，那么我们假设数据突发中存在中断。有2个通道，1个输入和1个输出。数据突发到达输入端，对于每个突发，我们在突发结束时写入输出通道。

实现非常简单。每次从输入通道接收时，我都会存储当前的 unix 时间戳。然后，我有一个停顿时间的行情。所有这些都是检查我们是否超过了最后一次爆发的等待时间。如果是这样，它会将last 重置为 0，并在输出通道上发出一个事件。

下面是一些使用 debounce 函数的代码，该函数具有 2 秒的停顿时间，它在输入通道上发送随机突发：

func main() {

    out := make(chan struct{})
    in := make(chan struct{})

    Debounce(2*time.Second, in, out)

    // Generating bursts of input data
    go func(in chan struct{}) {

        for {
            select {
            case <-time.Tick(1 * time.Second):
                in <- struct{}{}

                fmt.Println("Sending!")

                shouldSleep := rand.Intn(2)
                if shouldSleep == 1 {
                    time.Sleep(5 * time.Second)
                }
            }
        }
    }(in)

    // Listening for output events
    go func(out chan struct{}) {

        for _ = range out {
            fmt.Println("Got an event!")
        }
    }(out)

    // Do not let main terminate.
    done := make(chan struct{})
    <-done
}

【讨论】：

【解决方案2】：

听起来你需要在 goroutine 之间进行同步，也许沿着这条线。

func main() {

        // Create a channel for our input
        input := make(chan int, 1)
        // Create another for synchronization between main and forked goroutines
        done := make(chan bool)

        go func() {
                // block-wait for received value
                <-input

                // do some more things here

                // when done, send signal to the main goroutine
                done <- true
        }()

        // Do something while wait for the forked goroutine

        // this block until `<-done`
        <-done
        close(mychan)
}

post 这个post 非常清楚地解释了使用通道和同步组进行同步。

【讨论】：