Go语言实现生产者-消费者模式的方法总结

package main
import (
        "fmt"
        "time"
)
func producer(ch chan<- int) {
        for i := 1; i <= 5; i++ {
                ch <- i // 将数据发送到通道
                fmt.Println("生产者生产：", i)
                time.Sleep(time.Second) // 模拟生产过程
        }
        close(ch) // 关闭通道
}
func consumer(ch <-chan int, done chan<- bool) {
        for num := range ch {
                fmt.Println("消费者消费：", num)
                time.Sleep(2 * time.Second) // 模拟消费过程
        }
        done <- true // 通知主线程消费者已完成
}
func main() {
        ch := make(chan int, 3)  // 创建带缓冲的通道
        done := make(chan bool) // 用于通知主线程消费者已完成
        go producer(ch) // 启动生产者goroutine
        go consumer(ch, done) // 启动消费者goroutine
        // 主线程等待消费者完成
        <-done
        fmt.Println("消费者已完成")
        // 主线程结束，程序退出
}

package main
import (
        "fmt"
        "sync"
        "time"
)
type Data struct {
        Value int
}
type Queue struct {
        mutex      sync.Mutex
        cond       *sync.Cond
        buffer     []Data
        terminated bool
}
func NewQueue() *Queue {
        q := &Queue{}
        q.cond = sync.NewCond(&q.mutex)
        return q
}
func (q *Queue) Produce(data Data) {
        q.mutex.Lock()
        defer q.mutex.Unlock()
        q.buffer = append(q.buffer, data)
        fmt.Printf("Produced: %d\n", data.Value)
        // 唤醒等待的消费者
        q.cond.Signal()
}
func (q *Queue) Consume() Data {
        q.mutex.Lock()
        defer q.mutex.Unlock()
        // 等待数据可用
        for len(q.buffer) == 0 && !q.terminated {
                q.cond.Wait()
        }
        if len(q.buffer) > 0 {
                data := q.buffer[0]
                q.buffer = q.buffer[1:]
                fmt.Printf("Consumed: %d\n", data.Value)
                return data
        }
        return Data{}
}
func (q *Queue) Terminate() {
        q.mutex.Lock()
        defer q.mutex.Unlock()
        q.terminated = true
        // 唤醒所有等待的消费者
        q.cond.Broadcast()
}
func main() {
        queue := NewQueue()
        // 启动生产者
        for i := 1; i <= 3; i++ {
                go func(id int) {
                        for j := 1; j <= 5; j++ {
                                data := Data{Value: id*10 + j}
                                queue.Produce(data)
                                time.Sleep(time.Millisecond * 500) // 模拟生产时间
                        }
                }(i)
        }
        // 启动消费者
        for i := 1; i <= 2; i++ {
                go func(id int) {
                        for {
                                data := queue.Consume()
                                if data.Value == 0 {
                                        break
                                }
                                // 处理消费的数据
                                time.Sleep(time.Millisecond * 1000) // 模拟处理时间
                        }
                }(i)
        }
        // 等待一定时间后终止消费者
        time.Sleep(time.Second * 6)
        queue.Terminate()
        // 等待生产者和消费者完成
        time.Sleep(time.Second * 1)
}

import (
        "sync"
)
type Task struct {
        Priority int
        // 其他任务相关的字段...
}
type TaskQueue struct {
        cond      *sync.Cond
        tasks     []Task
}
func (q *TaskQueue) Enqueue(task Task) {
        q.cond.L.Lock()
        q.tasks = append(q.tasks, task)
        q.cond.Signal() // 通知等待的消费者
        q.cond.L.Unlock()
}
func (q *TaskQueue) Dequeue() Task {
        q.cond.L.Lock()
        for len(q.tasks) == 0 {
                q.cond.Wait() // 等待条件满足
        }
        task := q.findHighestPriorityTask()
        q.tasks = removeTask(q.tasks, task)
        q.cond.L.Unlock()
        return task
}
func (q *TaskQueue) findHighestPriorityTask() Task {
        // 实现根据优先级查找最高优先级任务的逻辑
        // ...
}
func removeTask(tasks []Task, task Task) []Task {
        // 实现移除指定任务的逻辑
        // ...
}