Golang中一个生产者通道连接多个消费者会发生什么

Golang中一个生产者通道连接多个消费者会发生什么 我对Go语言还很陌生，尤其是对goroutine和channel不太熟悉。我想了解当单个生产者向一个channel写入数据，而多个并发的goroutine使用同一个channel时会发生什么。我脑海中的模型类似于消息平台中主题的发布/订阅模式。这样理解对吗？

我使用的是无缓冲channel，并且假设当生产者在channel上发送消息时，所有消费者都应该收到该消息的一个副本。这样对吗？

我遇到了死锁的情况，所以我怀疑这个假设可能是错误的。

3 回复

非常感谢！这非常有帮助。

更多关于Golang中一个生产者通道连接多个消费者会发生什么的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html


我推荐你两个视频：

在Go语言中，channel的语义是点对点的，而不是发布/订阅模式。当单个生产者向一个无缓冲channel发送数据时，该数据只会被一个消费者接收。多个消费者从同一个channel接收数据时，它们会竞争接收数据，每个数据项只会被其中一个消费者获取。

你的理解有误：无缓冲channel不会将消息复制给所有消费者。相反，每个消息只会被一个goroutine接收。这类似于工作队列模式，而不是发布/订阅模式。

以下是示例代码，演示了多个消费者从同一个channel接收数据的情况：

package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan int) // 无缓冲channel

    // 启动三个消费者
    for i := 1; i <= 3; i++ {
        go func(id int) {
            for msg := range ch {
                fmt.Printf("消费者 %d 收到: %d\n", id, msg)
            }
        }(i)
    }

    // 生产者发送数据
    for i := 1; i <= 5; i++ {
        ch <- i
        time.Sleep(time.Millisecond * 100) // 稍作延迟以便观察
    }

    close(ch)
    time.Sleep(time.Second) // 等待消费者处理完成
}

输出可能类似于：

消费者 1 收到: 1
消费者 2 收到: 2
消费者 3 收到: 3
消费者 1 收到: 4
消费者 2 收到: 5

每个数字只被一个消费者接收。如果你需要发布/订阅模式，可以考虑以下方案：

  1. 使用多个channel：为每个消费者创建独立的channel。
  2. 使用sync包：结合slice和互斥锁手动实现广播机制。
  3. 使用第三方库：如github.com/eapache/channels中的Broadcast channel。

以下是简单的广播模式实现示例：

package main

import (
    "fmt"
    "sync"
    "time"
)

type Broadcast struct {
    mu   sync.RWMutex
    subs []chan int
}

func (b *Broadcast) Subscribe() chan int {
    b.mu.Lock()
    defer b.mu.Unlock()
    ch := make(chan int, 1)
    b.subs = append(b.subs, ch)
    return ch
}

func (b *Broadcast) Publish(msg int) {
    b.mu.RLock()
    defer b.mu.RUnlock()
    for _, ch := range b.subs {
        ch <- msg
    }
}

func main() {
    b := &Broadcast{}

    // 启动三个消费者
    for i := 1; i <= 3; i++ {
        go func(id int) {
            ch := b.Subscribe()
            for msg := range ch {
                fmt.Printf("消费者 %d 收到: %d\n", id, msg)
            }
        }(i)
    }

    // 等待消费者订阅
    time.Sleep(time.Millisecond * 100)

    // 生产者发送数据
    for i := 1; i <= 3; i++ {
        b.Publish(i)
        time.Sleep(time.Millisecond * 100)
    }
}

关于你遇到的死锁问题，通常是因为无缓冲channel的同步特性导致的。例如，如果消费者数量少于发送的数据量，且没有足够的goroutine接收数据，生产者会在发送时阻塞，导致死锁。确保生产者和消费者的生命周期匹配，并使用close(ch)正确关闭channel。

回到顶部