如何在使用渠道的golang流水線階段批量處理項目？

Question

我讀管道在線教程，並試圖構建經營這樣一個階段 - 將它們發送到了陳

如果我們還沒有前

1. 批了傳入事件中的10個批次的每個在5秒內沒有看到10個事件，將我們收到的數量合併併發送給他們，關閉出局並返回。

但是，我不知道第一個選擇案例會是什麼樣子。嘗試了很多事情，但無法通過這個。 任何指針非常感謝！

func BatchEvents(inChan <- chan *Event) <- chan *Event { 
    batchSize := 10 
    comboEvent := Event{} 
    go func() { 
     defer close(out) 
     i = 0 
     for event := range inChan { 
      select { 
      case -WHAT GOES HERE?-: 
       if i < batchSize { 
        comboEvent.data = append(comboEvent.data, event.data) 
        i++; 
       } else { 
        out <- &comboEvent 
        // reset for next batch 
        comboEvent = Event{} 
        i=0; 
       } 
      case <-time.After(5 * time.Second): 
       // process whatever we have seen so far if the batch size isn't filled in 5 secs 
       out <- &comboEvent 
       // stop after 
       return 
      } 
     } 
    }() 
    return out 
} 

Answer 1

而不是做一個範圍在通道中，你的第一選擇的情況下應該是從該通道，具有無限循環中的整個事情。

func BatchEvents(inChan <-chan *Event) <-chan *Event { 
    batchSize := 10 
    comboEvent := Event{} 
    go func() { 
     defer close(out) 
     i = 0 
     for { 
      select { 
      case event, ok := <-inChan: 
       if !ok { 
        return 
       } 
       comboEvent.data = append(comboEvent.data, event.data) 
       i++ 
       if i == batchSize { 
        out <- &comboEvent 
        // reset for next batch 
        comboEvent = Event{} 
        i = 0 
       } 
      case <-time.After(5 * time.Second): 
       // process whatever we have seen so far if the batch size isn't filled in 5 secs 
       if i > 0 { 
        out <- &comboEvent 
       } 
       // stop after 
       return 
      } 
     } 
    }() 
    return out 
}