Go实现TCP中继PubSub时客户端与服务端连接出现数据竞态的原因是什么

你遇到的数据竞态和字节损坏问题本质是切片共享导致的并发读写冲突，和服务端客户端并发运行的设计认知无关，你的认知没有错误，问题出在缓冲区复用的逻辑上：

• 你在publisher函数中只初始化了一次buf := make([]byte, 2048)，所有读取操作都复用这同一块内存空间
• Go的切片是引用类型，你调用ps.Publish(topic, buf[:n])时传递的是切片引用，没有复制实际的字节内容
• 当publishergoroutine下一次调用remote.Read(buf)时，会直接覆盖这块内存的内容，而此时subscribergoroutine可能正在读取同一块内存往客户端连接写入，就触发了竞态检测，同时也会导致字节内容被覆盖、出现损坏丢失的问题。

除了核心的竞态问题，你的实现还有两处明显缺陷：

1. 完全忽略IO错误：remote.Read、conn.Write的错误都没有处理，当对端连接断开时，对应goroutine会进入死循环空转，还会导致Pubsub中的订阅记录无法释放，出现内存泄漏
2. 发布逻辑存在阻塞风险：Publish方法遍历订阅通道写入时，如果某一个订阅者处理速度慢、通道缓冲区满，会直接阻塞整个发布流程，导致其他所有订阅者都无法收到新消息。

1. 修复竞态问题 #

每次读取到字节后，拷贝一份新的独立切片再发布，避免共享内存：

publisher := func(topic string) {
    remote, err := net.Dial("tcp", *raddr)
    if err != nil {
        return
    }
    defer remote.Close() // 退出时主动关闭连接
    buf := make([]byte, 2048)
    for {
        n, err := remote.Read(buf)
        if err != nil {
            break // 读取错误直接退出循环
        }
        // 复制新的切片再发布
        newBuf := make([]byte, n)
        copy(newBuf, buf[:n])
        ps.Publish(topic, newBuf)
    }
}

2. 修复订阅端错误处理和资源泄漏 #

订阅者退出时主动调用Unsubscribe释放资源：

subscriber := func(conn net.Conn, ch <-chan []byte, sub Sub) {
    defer func() {
        conn.Close()
        ps.Unsubscribe(sub)
    }()
    for i := range ch {
        _, err := conn.Write(i)
        if err != nil {
            break // 写入错误直接退出
        }
    }
}
// 启动goroutine时传入sub参数
ch, sub := ps.Subscribe("relay")
go subscriber(conn, ch, sub)

3. 优化发布逻辑避免阻塞 #

可以将发布改为非阻塞模式，避免单个慢客户端拖垮整个服务：

func (ps *Pubsub) Publish(topic string, msg []byte) {
    ps.mu.RLock()
    defer ps.mu.RUnlock()

    for sub, ch := range ps.subs {
        if sub.topic == topic {
            select {
            case ch <- msg:
            default:
                // 可选逻辑：通道满时丢弃消息、或者打印日志告警
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者hankivstmb