golang实现生产者消费者模式

在Go语言中，生产者消费者模式可以通过使用Goroutines和Channels来实现。Goroutines允许并发执行，而Channels则用于在生产者和消费者之间安全地传递数据。
生产者消费者模式的基本思路
生产者：负责生成数据并将其放入一个共享的缓冲区（Channel）。
消费者：从共享的缓冲区中取出数据并进行处理。
同步：使用Channel来同步生产者和消费者之间的操作，确保数据的安全传递。

建立一个channel

package outimport "fmt"type Out struct {data chan interface{}
}var out *Outfunc NewOut() *Out {if out == nil {out = &Out{data: make(chan interface{}, 65535),}}return out
}
func Println(i interface{}) {out.data <- i
}
func (o *Out) OutPut() {//for i := range o.data {//	fmt.Println(i)//	fmt.Println("out put")//}//fmt.Println("结束")for {select {case i := <-o.data:fmt.Println(i)}}
}

一对一

package one_oneimport ("producer-consumer/out""sync"
)type Task struct {ID int64
}func (t *Task) run() {out.Println(t.ID)
}var taskCh = make(chan Task, 10)const taskNum int64 = 10000func producer(wo chan<- Task) {var i int64for i = 1; i <= taskNum; i++ {t := Task{ID: i,}wo <- t}close(wo)
}
func consumer(ro <-chan Task) {for t := range ro {if t.ID != 0 {t.run()}}
}
func Exec() {wg := &sync.WaitGroup{}wg.Add(2)go func(wg *sync.WaitGroup) {defer wg.Done()producer(taskCh)}(wg)go func(wg *sync.WaitGroup) {defer wg.Done()consumer(taskCh)}(wg)wg.Wait()out.Println("执行成功")
}

一对多

package one_manyimport ("producer-consumer/out""sync"
)type Task struct {ID int64
}func (t *Task) run() {out.Println(t.ID)
}var taskCh = make(chan Task, 10)const taskNum int64 = 10000func producer(wo chan<- Task) {var i int64for i = 1; i <= taskNum; i++ {t := Task{ID: i,}wo <- t}close(wo)
}
func consumer(ro <-chan Task) {for t := range ro {if t.ID != 0 {t.run()}}
}
func Exec() {wg := &sync.WaitGroup{}wg.Add(1)go func(wg *sync.WaitGroup) {defer wg.Done()producer(taskCh)}(wg)var i int64for i = 0; i < taskNum; i++ {if i%100 == 0 {wg.Add(1)go func(wg *sync.WaitGroup) {defer wg.Done()consumer(taskCh)}(wg)}}wg.Wait()out.Println("执行成功")
}

多对一

package many_oneimport ("producer-consumer/out""sync"
)type Task struct {ID int64
}func (t *Task) run() {out.Println(t.ID)
}var taskCh = make(chan Task, 10)const taskNum int64 = 10000
const nums int64 = 100func producer(wo chan<- Task, startNum int64, nums int64) {var i int64for i = startNum; i < taskNum+nums; i++ {t := Task{ID: i,}wo <- t}
}
func consumer(ro <-chan Task) {for t := range ro {if t.ID != 0 {t.run()}}
}
func Exec() {wg := &sync.WaitGroup{}pwg := &sync.WaitGroup{}var i int64for i = 0; i < taskNum; i += nums {if i >= taskNum {break}wg.Add(1)pwg.Add(1)go func(i int64) {defer wg.Done()defer pwg.Done()producer(taskCh, i, nums)}(i)}wg.Add(1)go func() {defer wg.Done()consumer(taskCh)}()pwg.Wait()go close(taskCh)wg.Wait()out.Println("执行成功")
}

多对多

package many_manyimport ("fmt""producer-consumer/out""time"
)type Task struct {ID int64
}func (t *Task) run() {out.Println(t.ID)
}var taskCh = make(chan Task, 10)
var done = make(chan struct{})const taskNum int64 = 10000func producer(wo chan<- Task, done chan struct{}) {var i int64for {if i >= taskNum {i = 0}i++t := Task{ID: i,}select {case wo <- t:case <-done:out.Println("生产者退出")return}}
}
func consumer(ro <-chan Task, done chan struct{}) {for {select {case t := <-ro:if t.ID != 0 {t.run()}case <-done:for t := range ro {if t.ID != 0 {t.run()}}out.Println("消费者退出")return}}
}
func Exec() {go producer(taskCh, done)go producer(taskCh, done)go producer(taskCh, done)go producer(taskCh, done)go producer(taskCh, done)go producer(taskCh, done)go consumer(taskCh, done)go consumer(taskCh, done)time.Sleep(time.Second * 5)close(done)close(taskCh)time.Sleep(time.Second * 5)fmt.Println(len(taskCh))
}

主程序入口

package mainfunc main() {//o := out.NewOut()//go o.OutPut()//one_one.Exec()//one_many.Exec()//many_one.Exec()//many_many.Exec()//sig := make(chan os.Signal)//signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)//<-sig
}