複数チャネルのシリアルデータを持つ非同期キュー

Question

問題がありました...私は1つのインタフェースを実装する3つの構造体を持っています。私は構造データを処理する関数を持っていますが、いくつかの構造のデータの処理に数秒かかることがあります。データは順次処理する必要があります。

私はこのようにそれをやった

：

type dater struct { 
    value int 
} 

type waiter struct { 
    data chan *dater 
} 

func (dat *waiter) conf(wait bool) { 
    go func() { 
     for { 
      obj := <-dat.data 
      if wait { 
       time.Sleep(5 * time.Second) 
      } 
      fmt.Println("WAIT", wait, *obj) 
     } 
    }() 
} 

func (dat *waiter) gowrite(w *dater) { 
    dat.data <- w 
} 

func main() { 
    waiters := []*waiter{} 

    first := &waiter{data: make(chan *dater)} 
    first.conf(false) 
    waiters = append(waiters, first) 

    second := &waiter{data: make(chan *dater)} 
    second.conf(true) 
    waiters = append(waiters, second) 

    for i := 0; i < 30; i++ { 
     for _, wait := range waiters { 
      wait.gowrite(&dater{value: i}) 
     } 
    } 
} 

は出力：

WAIT false {0} 
WAIT false {1} 
WAIT true {0} (SLEEP 5 sec) 
WAIT false {2} (will appear after 5 seconds) 
WAIT true {1} 
WAIT false {3} 
WAIT true {2} 
WAIT false {4} 
WAIT true {3} 
WAIT false {5} 

私がしたい： "第二" 構造の

WAIT false {0} 
WAIT false {1} 
WAIT false {2} 
WAIT false {3} 
WAIT false {4} 
WAIT false {5} 
WAIT true {0} (5 seconds have passed and we show the first message) 
WAIT true {1} (This message will appear after 5 seconds) 
WAIT true {2} 
WAIT true {3} 

"gowrite" 機能は、まで待機「最初の」構造のチャネルがデータを受信する。しかし、私が重要なのは、データを非同期的に処理することです。この場合、すべてのチャネルが正しいデータを受信します。

Answer 1

二つの別々のバッファリングのチャンネル、like so必要と思われる：

package main 

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

func main() { 
    wg.Add(2) 
    go first() 
    go second() 
    for i := 1; i < 10; i++ { 
     a <- i // send to first 
     b <- i // send to second 
    } 
    close(a) 
    close(b) 
    wg.Wait() 
} 

func first() { 
    for { 
     d, ok := <-a 
     time.Sleep(100 * time.Millisecond) 
     if !ok { 
      break 
     } 
     fmt.Println("first job:", d) // do first job here. 
    } 
    wg.Done() 
} 

func second() { 
    for { 
     d, ok := <-b 
     time.Sleep(300 * time.Millisecond) 
     if !ok { 
      break 
     } 
     fmt.Println("second job:", d) // do second job here. 
    } 
    wg.Done() 
} 

var a = make(chan int, 1000) 
var b = make(chan int, 1000) 
var wg sync.WaitGroup 

出力：

first job: 1 
first job: 2 
second job: 1 
first job: 3 
first job: 4 
first job: 5 
second job: 2 
first job: 6 
first job: 7 
first job: 8 
second job: 3 
first job: 9 
second job: 4 
second job: 5 
second job: 6 
second job: 7 
second job: 8 
second job: 9