ZeroMQ Goroutine间通信：高效利用inproc://传输

package main

import (
    "fmt"
    zmq "github.com/alecthomas/gozmq" // 假设使用此ZeroMQ绑定库
    "sync"
    "time"
)

// startWorker 函数现在接收一个共享的ZeroMQ上下文
func startWorker(context *zmq.Context, workerID int) {
    // defer context.Close() // 不在这里关闭上下文，因为它是共享的

    worker, err := context.NewSocket(zmq.REP)
    if err != nil {
        fmt.Printf("Worker %d: 无法创建套接字: %v\n", workerID, err)
        return
    }
    defer worker.Close() // 确保在worker退出时关闭套接字

    // 使用 inproc:// 连接到后端，现在它会工作
    err = worker.Connect("inproc://backend")
    if err != nil {
        fmt.Printf("Worker %d: 无法连接到 inproc://backend: %v\n", workerID, err)
        return
    }
    fmt.Printf("Worker %d: 成功连接到 inproc://backend\n", workerID)

    for {
        data, err := worker.Recv(0)
        if err != nil {
            fmt.Printf("Worker %d: 接收数据失败: %v\n", workerID, err)
            break // 退出循环或处理错误
        }
        fmt.Printf("Worker %d 收到数据: %s\n", workerID, string(data))
        worker.Send([]byte(fmt.Sprintf("Worker %d 收到您的数据", workerID)), 0)
    }
}

func main() {
    // 创建一个 ZeroMQ 上下文，供所有Goroutine共享
    context, err := zmq.NewContext()
    if err != nil {
        fmt.Println("无法创建ZeroMQ上下文:", err)
        return
    }
    defer context.Close() // 确保在main函数退出时关闭上下文

    // 客户端前端套接字
    frontend, err := context.NewSocket(zmq.ROUTER)
    if err != nil {
        fmt.Println("无法创建前端套接字:", err)
        return
    }
    defer frontend.Close()
    frontend.Bind("tcp://*:5559")
    fmt.Println("前端绑定到 tcp://*:5559")

    // 服务后端套接字
    backend, err := context.NewSocket(zmq.DEALER)
    if err != nil {
        fmt.Println("无法创建后端套接字:", err)
        return
    }
    defer backend.Close()
    // 现在使用 inproc:// 绑定，因为Worker将共享同一个上下文
    err = backend.Bind("inproc://backend")
    if err != nil {
        fmt.Println("无法绑定到 inproc://backend:", err)
        return
    }
    fmt.Println("后端绑定到 inproc://backend")

    var wg sync.WaitGroup
    numWorkers := 4
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        // 将共享的上下文传递给每个Worker Goroutine
        go func(id int) {
            defer wg.Done()
            startWorker(context, id)
        }(i + 1)
    }

    // 启动内置设备（消息队列）
    // 注意：zmq.Device 是一个阻塞调用，它会接管当前Goroutine
    // 因此，如果要在Device之后执行其他逻辑，需要将其放入单独的Goroutine
    go func() {
        fmt.Println("启动ZeroMQ QUEUE设备...")
        zmq.Device(zmq.QUEUE, frontend, backend)
    }()

    // 为了演示，让main Goroutine运行一段时间，以便Worker可以处理请求
    fmt.Println("Broker正在运行，等待Worker和客户端连接...")
    time.Sleep(5 * time.Second) // 运行5秒钟，以便Worker有时间连接
    // 实际应用中，这里可能是select{}或其他阻塞机制来保持main Goroutine存活

    // 模拟发送一些请求到前端
    clientContext, _ := zmq.NewContext()
    defer clientContext.Close()
    client, _ := clientContext.NewSocket(zmq.REQ)
    defer client.Close()
    client.Connect("tcp://127.0.0.1:5559")

    for i := 0; i < 3; i++ {
        msg := fmt.Sprintf("你好，来自客户端 %d", i+1)
        client.Send([]byte(msg), 0)
        reply, _ := client.Recv(0)
        fmt.Printf("客户端收到回复: %s\n", string(reply))
        time.Sleep(500 * time.Millisecond)
    }

    // 优雅关闭：在实际应用中，需要一个机制来通知Worker停止并等待它们退出
    // 这里简单地等待一段时间，然后程序退出
    fmt.Println("等待Worker Goroutine完成...")
    // 无法直接等待zmq.Device的Goroutine，因为它是阻塞的
    // 实际应用中，需要一个信号量来优雅地停止Device
    time.Sleep(1 * time.Second) // 给Worker一点时间处理最后的请求
    // wg.Wait() // 如果Worker能正常退出，这里可以等待

    fmt.Println("程序退出。")
}