使用Golang Gin+MQTT3.1时关闭Channel触发Panic问题排查

问题根源 #

你遇到的Panic核心原因是**mqttChannels这个普通Go map不具备并发安全性**：

• Gin的每个HTTP请求都会在独立goroutine中执行NodeRequestSync，涉及对mqttChannels的写入和删除操作；
• MQTT客户端的响应处理HandleDeviceResponse运行在客户端自身的goroutine中，会对mqttChannels执行查询和写入channel的操作；
• 普通map在并发读写时会触发竞态条件，破坏map内部结构，最终导致运行时panic。UUID保证requestId唯一只能避免key冲突，但解决不了并发操作map的本质问题，加延迟只是降低了冲突概率，无法彻底修复。

解决方案 #

改用Go标准库的sync.Map替代普通map，它内置了并发安全的锁机制，同时优化channel的发送逻辑避免goroutine泄漏：

修改后的完整代码 #

import (
    "errors"
    "strings"
    "time"
    "sync"
    "github.com/google/uuid"
    "github.com/gin-gonic/gin"
    mqtt "github.com/eclipse/paho.mqtt.golang"
)

var mqttChannels sync.Map

func NodeRequestSync(c *gin.Context, mqttPath string, deviceId string, payload interface{}, timeout time.Duration) (string, error) {
    requestId := uuid.New().String() 
    requestTopic := myRequestTopic(nodeId, mqttPath, requestId)
    // 创建带缓冲的channel，避免发送阻塞
    respChan := make(chan []byte, 1)
    // 并发安全地存储channel
    mqttChannels.Store(requestId, respChan)
    
    // 发送MQTT请求
    NodeClient.Publish(requestTopic, 0, false, payload)
    
    // 处理响应或超时
    select {
    case responseData := <-respChan:
        mqttChannels.Delete(requestId)
        close(respChan) // 关闭channel释放资源
        return string(responseData), nil
    case <-time.After(timeout):
        mqttChannels.Delete(requestId)
        close(respChan)
        return "", errors.New("timeout or node un-exist")
    }
}

func HandleDeviceResponse(client mqtt.Client, msg mqtt.Message) {
    requestId := msg.Topic()[strings.LastIndex(msg.Topic(), "/")+1:]
    // 并发安全地查询channel
    ch, ok := mqttChannels.Load(requestId)
    if !ok {
        return
    }
    respChan, okCast := ch.(chan []byte)
    if !okCast {
        return
    }
    
    // 非阻塞发送，避免因请求已被处理而阻塞当前goroutine
    select {
    case respChan <- msg.Payload():
    default:
        // 发送失败，说明请求已超时或被处理，直接忽略
    }
}

关键优化点 #

1. sync.Map替代普通map：所有对mqttChannels的读写操作都通过Store/Load/Delete完成，保证并发安全；
2. 带缓冲的channel：设置容量为1，避免发送端因接收端未就绪而阻塞；
3. 非阻塞发送：在HandleDeviceResponse中用select+default实现非阻塞发送，防止MQTT响应goroutine被阻塞；
4. 及时关闭channel：请求处理完成（成功/超时）后关闭channel，避免资源泄漏。

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