无Panic的持续运行Goroutine异常终止问题排查

问题描述 #

• 为每个传感器创建独立Goroutine，定期采集状态并记录日志；Goroutine通过channel接收终止信号，传感器状态存储在sync.Map中（存储指针）
• 运行数天后部分监控Goroutine停止记录，但处理Goroutine仍能正常更新传感器状态
• 已通过recover()捕获panic并尝试重建线程，但无效；Goroutine终止时无错误日志，怀疑内存泄漏或其他静默终止原因

异常原因分析 #

从提供的代码来看，存在多个可能导致Goroutine静默终止的问题：

1. 锁状态跟踪变量名不一致：代码中混用isStateLocked和isSessionStateLocked，导致panic发生时无法正确判断锁状态，可能引发逻辑混乱，甚至无法解锁导致其他阻塞，但更直接的是破坏了Goroutine的错误恢复逻辑。
2. time.Sleep放在select default分支：当前循环结构下，Goroutine大部分时间处于sleep状态，仅在每次sleep结束后才检查终止channel。若sleep期间channel被关闭或发送信号，Goroutine无法及时响应；同时这种写法会导致select的终止分支失去实时性。
3. 长期持有旧的状态指针：Goroutine启动时仅从sync.Map加载一次状态指针，若后续该传感器的状态指针被替换，当前Goroutine会持有旧指针继续运行。如果旧指针的channel未被触发，且新指针的监控Goroutine被重复创建，旧Goroutine可能因insertLog阻塞等原因静默停止。
4. 未处理channel关闭场景：若state.ch被意外关闭，case <-state.ch会立即返回零值，导致Goroutine直接终止且无日志记录。
5. 无阻塞操作超时控制：insertLog若因数据库连接耗尽、日志队列满等原因永久阻塞，Goroutine会卡在该步骤，不再执行后续循环，表现为“停止记录”，且这种情况不会触发recover()。

解决方案 #

针对上述问题，可通过以下步骤修复：

1. 统一锁状态变量名：确保锁的加锁/解锁状态跟踪准确，避免panic时无法正确解锁。
2. 用time.After替代单独的time.Sleep：将定期采集逻辑放在select的超时分支中，既保证定期执行，又能实时响应终止信号。
3. 每次循环重新加载状态：避免持有旧的状态指针，确保Goroutine始终操作最新的传感器状态。
4. 添加阻塞操作超时控制：对insertLog这类可能阻塞的操作添加超时，防止Goroutine被永久卡住。
5. 处理channel关闭事件：接收channel信号时检查channel是否关闭，记录日志后再退出。
6. 增强日志与监控：添加关键节点日志（如Goroutine启动、状态采集、日志插入结果），便于排查问题；同时可定期统计活跃监控Goroutine数量，及时发现异常。

修复后的示例代码 #

import (
    "sync"
    "time"
    "fmt"
    "runtime/debug"
)

// sensor state struct
type sensorState struct {
    a float64
    b int
    lock sync.Mutex
    ch   chan int
}

type curStatus struct {
    a float64
    b int
    // ... 其他字段
}

// log struct
type stateLog struct {
    state     curStatus
    timestamp int64
    method    string
}

// stateMap maintains state of different sensors 
var stateMap sync.Map

func init() {
    stateMap.Store("Sensor-A", &sensorState{
        a:    0,
        b:    0,
        lock: sync.Mutex{},
        ch:   make(chan int, 1), // 带缓冲避免发送终止信号时阻塞
    })
}

func ChargerMonitoringThreadV2(sensorId string) {
    isStateLocked := false
    sensorStatus := curStatus{}

    defer func() {
        if er := recover(); er != nil {
            fmt.Printf("ERROR: [sensorId:%s][ChargerMonitoringThreadV2] Recovered from - %v\n", sensorId, er)
            fmt.Printf("ERROR: [sensorId:%s][ChargerMonitoringThreadV2] stacktrace from panic:\n%s\n", sensorId, string(debug.Stack()))

            if isStateLocked {
                fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Released lock\n", sensorId)
                // 尝试加载最新状态并解锁，避免旧指针无效
                if sessionState, ok := stateMap.Load(sensorId); ok {
                    if state, ok := sessionState.(*sensorState); ok {
                        state.lock.Unlock()
                    }
                }
                isStateLocked = false
            }

            fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Restarting monitoring thread...\n", sensorId)
            go ChargerMonitoringThreadV2(sensorId)
            // logging.ReportException(fmt.Errorf("%v\n%s", er, string(debug.Stack()))) // 保留原有日志上报逻辑
            return
        }
    }()

    fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Started monitoring thread...\n", sensorId)

    for {
        // 每次循环重新加载状态，确保操作最新指针
        sessionState, ok := stateMap.Load(sensorId)
        if !ok {
            fmt.Printf("WARN: [sensorId:%s][ChargerMonitoringThreadV2] Sensor state not found, exiting...\n", sensorId)
            return
        }
        state, ok := sessionState.(*sensorState)
        if !ok {
            fmt.Printf("WARN: [sensorId:%s][ChargerMonitoringThreadV2] Invalid sensor state type, exiting...\n", sensorId)
            return
        }

        select {
        case sig, ok := <-state.ch:
            if !ok {
                fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Termination channel closed, exiting...\n", sensorId)
            } else {
                fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Received termination signal %d, exiting...\n", sensorId, sig)
            }
            return
        case <-time.After(60 * time.Second):
            curTS := time.Now().UnixMilli()

            state.lock.Lock()
            isStateLocked = true

            sensorStatus.a = state.a
            sensorStatus.b = state.b // 补充采集b字段

            state.lock.Unlock()
            isStateLocked = false

            fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Collected state: a=%.2f, b=%d\n", sensorId, sensorStatus.a, sensorStatus.b)

            // 对insertLog添加超时控制，避免永久阻塞
            logChan := make(chan error, 1)
            go func() {
                logChan <- insertLog(&stateLog{
                    state:     sensorStatus,
                    timestamp: curTS,
                    method:    "monitoringThread",
                })
            }()

            select {
            case err := <-logChan:
                if err != nil {
                    fmt.Printf("ERROR: [sensorId:%s][ChargerMonitoringThreadV2] Failed to insert log: %v\n", sensorId, err)
                } else {
                    fmt.Printf("INFO: [sensorId:%s][ChargerMonitoringThreadV2] Log inserted successfully\n", sensorId)
                }
            case <-time.After(5 * time.Second):
                fmt.Printf("ERROR: [sensorId:%s][ChargerMonitoringThreadV2] Insert log timed out\n", sensorId)
            }
        }
    }
}

// 模拟insertLog函数，实际需替换为业务逻辑
func insertLog(log *stateLog) error {
    // 写入数据库/文件等操作
    return nil
}

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