one-one
先创建out.go目录与文件夹
// 定义了一个名为out的包，用于处理输出相关的功能。
package out
import "fmt"
// Out结构体定义了一个channel，用于存储需要输出的数据。
type Out struct {
data chan interface{} // data字段是一个interface{}类型的channel，用于存储任意类型的数据。
}
// out是一个全局变量，指向Out类型的指针，用于全局访问。
var out *Out
// Newout函数用于创建一个新的Out实例，如果全局变量out为nil，则创建一个新的Out实例并初始化其data字段。
// 如果out已经初始化，则直接返回现有的out实例。
func Newout() *Out {
if out == nil {
out = &Out{
make(chan interface{}, 65535), // 初始化data字段为一个容量为65535的channel。
}
}
return out
}
// Println函数用于向Out实例的data channel发送数据。
// 它接受一个interface{}类型的参数i，并将i发送到out的data channel中。
func Println(i interface{}) {
out.data <- i // 向channel发送数据。
}
// OutPut方法是Out结构体的方法，用于从data channel中接收数据并打印。
// 它是一个无限循环，使用select语句监听data channel。
// 当channel中有数据时，使用fmt.Println打印接收到的数据。
func (o *Out) OutPut() {
for {
select {
case i := <-o.data: // 从channel中接收数据。
fmt.Println(i) // 打印接收到的数据。
}
}
}
同时创建one-one.go文件，来实现单生产者单消费者模型
// 定义了一个名为one_one的包，用于实现生产者-消费者模式。
package one_one
import (
"producer-consumer/out" // 导入自定义的out包，用于输出功能。
"sync" // 导入sync包，用于同步goroutine。
)
// Task结构体定义了一个任务，包含一个ID字段。
type Task struct {
ID int64
}
// Task的run方法用于执行任务，这里只是简单地打印任务的ID。
func (t *Task) run() {
out.Println(t.ID) // 使用out包的Println函数打印任务ID。
}
// taskCh是一个缓冲channel，用于在生产者和消费者之间传递Task对象。
var taskCh = make(chan Task, 10)
// taskNum定义了要生成的任务数量。
const taskNum int64 = 1000
// producer函数是一个生产者goroutine，它生成taskNum个任务并发送到channel。
func producer(wo chan<- Task) {
var i int64
for i = 1; i < taskNum; i++ {
t := Task{i} // 创建一个新的Task对象。
wo <- t // 将Task对象发送到channel。
}
close(wo) // 生产结束后关闭channel。
}
// consumer函数是一个消费者goroutine，它从channel接收任务并执行。
func consumer(ro <-chan Task) {
for t := range ro {
if t.ID != 0 {
t.run() // 执行任务。
}
}
}
// Exec函数用于启动生产者和消费者goroutine，并等待它们完成。
func Exec() {
wg := &sync.WaitGroup{} // 创建一个WaitGroup对象用于同步。
wg.Add(2) // 增加两个计数，一个用于生产者，一个用于消费者。
// 启动生产者goroutine。
go func(wg *sync.WaitGroup) {
defer wg.Done() // 确保在goroutine结束时减少WaitGroup计数。
go producer(taskCh) // 调用producer函数启动生产者。
}(wg)
// 启动消费者goroutine。
go func(wg *sync.WaitGroup) {
defer wg.Done() // 确保在goroutine结束时减少WaitGroup计数。
consumer(taskCh) // 调用consumer函数启动消费者。
}(wg)
wg.Wait() // 等待所有goroutine完成。
out.Println("执行成功") // 打印执行成功的消息。
}
最后在main函数中测试
// 定义了main包，这是程序的入口点。
package main
import (
one_one "producer-consumer/one-one" // 导入one-one包，它实现了生产者-消费者模式。
"producer-consumer/out" // 导入out包，它提供了输出功能。
"time" // 导入time包，用于处理时间相关的功能。
)
// main函数是程序的入口点。
func main() {
// 创建out包的Out实例，用于后续的输出操作。
o := out.Newout()
// 启动一个goroutine来运行Out实例的OutPut方法，这个方法会不断地从channel中读取数据并打印。
go o.OutPut()
// 调用one_one包的Exec函数，启动生产者和消费者逻辑。
one_one.Exec()
// 让main函数暂停4秒钟，确保有足够的时间让生产者和消费者完成它们的任务。
// 这是为了在程序结束前给goroutine足够的时间来处理和打印所有的输出。
time.Sleep(time.Second * 4)
}
one-many
将原先one-one文件稍做改变
Chanel的线程安全的，所以可以直接执行并发操作
package one_many
import (
"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
func producer(wo chan<- Task) {
var i int64
for i = 1; i < taskNum; i++ {
t := Task{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()
go producer(taskCh)
}(wg)
var i int64
for 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("执行成功")
}
many-one
package many_one
import (
"producer-consumer/out"
"sync"
)
// Task 结构体表示一个任务，每个任务有一个唯一的 ID
type Task struct {
ID int64 // 任务的 ID，用于标识任务
}
// run 方法用来执行任务（在这里是打印任务 ID）
func (t *Task) run() {
out.Println(t.ID) // 输出任务 ID
}
// taskCh 是一个缓冲区为 10 的 channel，用于在生产者和消费者之间传递 Task
var taskCh = make(chan Task, 10)
// taskNum 代表总共有多少个任务需要处理
const taskNum int64 = 10000
// nums 代表每次生产者生产的任务数量（批量生产）
const nums int64 = 100
// producer 函数模拟任务的生产者。它会将任务（Task）发送到任务 channel 中
func producer(wo chan<- Task, startNum int64, nums int64) {
// 循环生成任务并将任务发送到任务通道
var i int64
for i = startNum; i < startNum+nums; i++ {
t := Task{ID: i} // 创建一个任务
wo <- t // 将任务发送到任务通道
}
}
// consumer 函数模拟任务的消费者。它从任务通道接收任务并处理
func consumer(ro <-chan Task) {
// 从任务通道读取任务并执行
for t := range ro {
if t.ID != 0 { // 如果任务 ID 不为 0，则执行任务
t.run()
}
}
}
// Exec 是主执行函数，负责启动生产者和消费者 goroutine，并进行协程同步
func Exec() {
// 创建两个 WaitGroup 用于同步生产者和消费者的执行
wg := &sync.WaitGroup{} // 用于等待消费者 goroutine 完成
pwg := &sync.WaitGroup{} // 用于等待所有生产者 goroutine 完成
// 启动消费者 goroutine，开始从 taskCh 中消费任务
wg.Add(1) // 增加一个计数，表示一个 goroutine（消费者）需要等待
go func() {
defer wg.Done() // 在消费者完成时调用 Done()，减少计数
consumer(taskCh) // 启动消费者并处理任务
}()
// 启动多个生产者 goroutine，每个生产者负责生成一定数量的任务
for i := int64(0); i < taskNum; i += nums {
if i >= taskNum {
break
}
pwg.Add(1) // 增加一个计数，表示一个 goroutine（生产者）需要等待
// 启动生产者 goroutine，批量生成任务并将其发送到 taskCh
go func(i int64) {
defer pwg.Done() // 在生产者完成时调用 Done()，减少计数
producer(taskCh, i, nums) // 启动生产者并生成任务
}(i)
}
// 输出执行成功的提示
out.Println("执行成功")
// 等待所有生产者完成（所有任务已发送完）
pwg.Wait()
// 在所有生产者完成后关闭任务通道
// 使用 go 是为了防止与消费者的执行顺序冲突
// 一旦任务通道关闭，消费者会停止接收任务
go close(taskCh)
// 等待消费者完成（所有任务已被处理）
wg.Wait()
}
many-many
多对多模式，消费者和生产者都不主动退出，我们通过一个第三方信号来控制退出
// many_many包实现了多个生产者和多个消费者的场景。
package many_many
import "producer-consumer/out" // 导入out包，用于输出任务ID。
// Task结构体定义了一个任务，包含一个ID字段。
type Task struct {
ID int64
}
// Task的run方法用于执行任务，这里只是简单地打印任务的ID。
func (t *Task) run() {
out.Println(t.ID)
}
// taskChan是一个缓冲channel，用于在生产者和消费者之间传递Task对象。
var taskChan = make(chan Task, 10)
// done是一个用于通知生产者和消费者退出的channel。
var done = make(chan struct{})
// taskNum定义了生产者要生成的任务数量。
const taskNum int64 = 10000
// producer函数是一个生产者goroutine，它生成任务并发送到taskChan。
func producer(wo chan<- Task, done chan struct{}) {
var i int64
for {
if i >= taskNum {
i = 0 // 如果达到任务数量上限，重置i。
}
i++
t := Task{
ID: i,
}
// 使用select语句来处理两个case：发送任务到channel或者接收done信号退出。
select {
case wo <- t:
case <-done:
out.Println("生产者退出")
return
}
}
}
// consumer函数是一个消费者goroutine，它从taskChan接收任务并执行。
func consumer(ro <-chan Task, done chan struct{}) {
for {
select {
case t := <-ro:
if t.ID != 0 {
t.run() // 执行任务。
}
case <-done:
// 如果接收到done信号，处理channel中剩余的所有任务然后退出。
for t := range ro {
if t.ID != 0 {
t.run()
}
}
return
}
}
}
// Exec函数用于启动多个生产者和消费者goroutine。
func Exec() {
// 启动多个生产者goroutine。
for i := 0; i < 8; i++ {
go producer(taskChan, done)
}
// 启动多个消费者goroutine。
for i := 0; i < 4; i++ {
go consumer(taskChan, done)
}
}
此时会无限的生产和消费
这时候我们关闭channel
如果先关闭数据channel，在关闭控制channel
time.Sleep(5 * time.Second)
close(taskChan)
close(done)
time.Sleep(5 * time.Second)
fmt.Println(len(taskChan))
报错，说我们往关闭了的channel里写数据
因为，如果我们没有先关闭控制channel，那么消费者和生产者就都还没有收到停止的消息，在两行语句的时间差中，会发生很多写入channel的操作。
所以我们要先关闭控制channel
正常退出！
