具有多个通道的串行数据的异步队列

Question

有一个问题...我有三个实现一个接口的结构。我有一个处理结构数据的函数，但碰巧处理某些结构的数据可能需要几秒钟。数据必须按顺序处理。

我是这样做的：

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}

“second”结构的“gowrite”函数一直等待，直到结构“first”的通道接收到数据。但对我来说重要的是它们异步处理数据而不会相互阻塞。在这种情况下，所有通道都会收到正确的数据。

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