Go语言中ZeroMQ中断信号的惯用处理方法

import (
    "os"
    "os/signal"
    "syscall" // For specific signal types like syscall.SIGINT
)

// listenForInterrupt 负责监听中断信号，并在收到信号时通知退出通道
func listenForInterrupt(exitChan chan<- bool) {
    sigChan := make(chan os.Signal, 1) // 使用带缓冲的通道，防止信号丢失
    // 监听 SIGINT (Ctrl+C) 和 SIGTERM (终止信号)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    // 阻塞等待信号
    <-sigChan
    fmt.Println("\nW: 收到中断信号，准备退出...")
    exitChan <- true // 通知主goroutine退出
}

import (
    zmq "github.com/pebbe/zmq4" // 推荐使用更活跃的zmq库，如pebbe/zmq4
    "fmt"
)

// zmqReceiver 负责在一个单独的goroutine中接收ZeroMQ消息
func zmqReceiver(socket *zmq.Socket, dataChan chan<- []byte, errChan chan<- error) {
    for {
        // Recv是阻塞的，但在独立的goroutine中不会影响主循环
        msgbytes, err := socket.RecvBytes(0) // 0 表示阻塞模式
        if err != nil {
            // 检查是否是预期的上下文关闭错误
            if zmq.Error(err) == zmq.ErrContextClosed {
                fmt.Println("ZeroMQ上下文已关闭，接收器退出。")
                return // 上下文关闭，接收器可以安全退出
            }
            errChan <- err // 将错误发送到错误通道
        } else {
            dataChan <- msgbytes // 将接收到的数据发送到数据通道
        }
    }
}

package main

import (
    "fmt"
    "os"
    "os/signal"
    "syscall"
    zmq "github.com/pebbe/zmq4" // 推荐使用pebbe/zmq4
)

func listenForInterrupt(exitChan chan<- bool) {
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    <-sigChan
    fmt.Println("\nW: 收到中断信号，准备退出...")
    exitChan <- true
}

func zmqReceiver(socket *zmq.Socket, dataChan chan<- []byte, errChan chan<- error) {
    for {
        msgbytes, err := socket.RecvBytes(0)
        if err != nil {
            if zmq.Error(err) == zmq.ErrContextClosed {
                fmt.Println("ZeroMQ上下文已关闭，接收器退出。")
                return
            }
            errChan <- err
        } else {
            dataChan <- msgbytes
        }
    }
}

func main() {
    exit := make(chan bool)
    data := make(chan []byte)
    errors := make(chan error)

    // 启动信号监听goroutine
    go listenForInterrupt(exit)

    // 初始化ZeroMQ上下文和套接字
    context, err := zmq.NewContext()
    if err != nil {
        fmt.Printf("创建ZeroMQ上下文失败: %v\n", err)
        return
    }
    defer context.Term() // 使用Term()代替Close()来关闭上下文

    socket, err := context.NewSocket(zmq.REP)
    if err != nil {
        fmt.Printf("创建ZeroMQ套接字失败: %v\n", err)
        return
    }
    defer socket.Close()

    err = socket.Bind("tcp://*:5555")
    if err != nil {
        fmt.Printf("绑定ZeroMQ地址失败: %v\n", err)
        return
    }
    fmt.Println("ZeroMQ服务器已启动，监听 tcp://*:5555")

    // 启动ZeroMQ接收goroutine
    go zmqReceiver(socket, data, errors)

    // 主循环，使用select监听多个通道
    for {
        select {
        case <-exit: // 收到退出信号
            fmt.Println("W: 收到中断信号，正在关闭服务器...")
            // 在这里可以进行清理工作，例如保存状态等
            return // 退出主循环，程序结束
        case err := <-errors: // 收到ZeroMQ接收错误
            fmt.Printf("ZeroMQ接收错误: %v\n", err)
            // 根据错误类型决定是否继续或退出
            // 例如，如果是网络断开等严重错误，可能需要退出
        case msgbytes := <-data: // 收到ZeroMQ数据
            fmt.Printf("收到消息: %s\n", string(msgbytes))
            // 假设这是一个REP套接字，需要回复消息
            reply := fmt.Sprintf("Hello %s", string(msgbytes))
            _, err := socket.Send(reply, 0)
            if err != nil {
                fmt.Printf("发送回复失败: %v\n", err)
            }
        }
    }
}