Golang多线程多个消费者处理

最近就一个情景用golang写了几个实现。

情景：

一个生产者，两个工作者，一个检查者。

工作者和检查者的代码

func (p *Processor) LoopChannel(input, output chan *Data, done <-chan struct{}) {
	for {
		select {
		case r := <-input:
			if r.Flag != p.No {
				r.ArrayDataHandle(p.No)
			}
			output <- r
		case <-done:
			return
		}
	}
}

func (p *Processor) Checking(input, output chan *Data, done <-chan struct{}) {
	for {
		select {
		case r := <-input:
			if len(r.Array) < 2 {
				output <- r
			} else {
				fmt.Println("show data ", r, r.Array)
			}
		case <-done:
			return
		}
	}
}

两个工作者同时处理生产者生产的数据。检查者完成检查后放行数据。

方案一：

遍历工作者，对任意工作者从工作队列中拿一个产品，如果自己没处理过过这个产品则处理，否则跳过，将处理完的数据交由检查者检查。

检查者如果发现数据缺少某个生产者的处理，则将数据退回队伍中。

	originChan := make(chan *lib.Data, 10)
	checkChan := make(chan *lib.Data, 10)

	go processor1.LoopChannel(originChan, checkChan, done)
	go processor2.LoopChannel(originChan, checkChan, done)
	go processor1.Checking(checkChan, originChan, done)
	go processor2.Checking(checkChan, originChan, done)

方案二：

遍历产品，将产品指针交由每个生产者队列。每个工作者处理完之后，交由检查者检查。

检查者检查产品，若产品没处理完则跳过。

	go processor1.LoopChannel(channel1, checkChan, done)
	go processor2.LoopChannel(channel2, checkChan, done)
	go processor1.Checking(checkChan, channel1, done)
	//go processor2.Checking(checkChan, channel2, done)

	d := &lib.Data{}
	d.Init()
	channel1 <- d
	channel2 <- d

代码地址：

Golang多线程 多个消费者处理

Golang多线程多个消费者处理