使用Go语言实现ZeroMQ通信（zmq4库指南）

ZeroMQ（ØMQ）是一个高性能异步消息库，而zmq4是Go语言中最流行的ZeroMQ绑定库。下面我将详细介绍如何使用zmq4实现各种消息模式。

安装zmq4

首先需要安装ZeroMQ系统库，然后在Go项目中引入zmq4：

# Ubuntu/Debian安装ZeroMQ
sudo apt-get install libzmq3-dev

# 安装Go绑定
go get github.com/pebbe/zmq4

基本消息模式实现

1. 请求-应答模式（REQ-REP）

服务端（应答方）：

package main

import (
	"fmt"
	"github.com/pebbe/zmq4"
)

func main() {
	// 创建REP套接字
	responder, _ := zmq4.NewSocket(zmq4.REP)
	defer responder.Close()
	responder.Bind("tcp://*:5555")

	for {
		// 等待请求
		msg, _ := responder.Recv(0)
		fmt.Printf("Received request: [%s]\n", msg)

		// 发送响应
		responder.Send("World", 0)
	}
}

客户端（请求方）：

package main

import (
	"fmt"
	"github.com/pebbe/zmq4"
	"time"
)

func main() {
	// 创建REQ套接字
	requester, _ := zmq4.NewSocket(zmq4.REQ)
	defer requester.Close()
	requester.Connect("tcp://localhost:5555")

	for i := 0; i < 10; i++ {
		// 发送请求
		requester.Send("Hello", 0)

		// 接收响应
		reply, _ := requester.Recv(0)
		fmt.Printf("Received reply %d [%s]\n", i, reply)
		
		time.Sleep(time.Second)
	}
}

2. 发布-订阅模式（PUB-SUB）

发布者：

package main

import (
	"github.com/pebbe/zmq4"
	"time"
)

func main() {
	publisher, _ := zmq4.NewSocket(zmq4.PUB)
	defer publisher.Close()
	publisher.Bind("tcp://*:5556")

	for {
		// 发布两条消息，分别带有A和B前缀
		publisher.Send("A Hello", 0)
		publisher.Send("B World", 0)
		time.Sleep(time.Second)
	}
}

订阅者：

package main

import (
	"fmt"
	"github.com/pebbe/zmq4"
)

func main() {
	subscriber, _ := zmq4.NewSocket(zmq4.SUB)
	defer subscriber.Close()
	subscriber.Connect("tcp://localhost:5556")
	
	// 只订阅以"A"开头的消息
	subscriber.SetSubscribe("A")

	for {
		msg, _ := subscriber.Recv(0)
		fmt.Printf("Received: %s\n", msg)
	}
}

3. 管道模式（PUSH-PULL）

任务分发者：

package main

import (
	"github.com/pebbe/zmq4"
	"time"
)

func main() {
	sender, _ := zmq4.NewSocket(zmq4.PUSH)
	defer sender.Close()
	sender.Bind("tcp://*:5557")

	// 让所有worker连接
	time.Sleep(time.Second)

	// 发送100个任务
	for i := 0; i < 100; i++ {
		sender.Send("task", 0)
	}
}

工作者：

package main

import (
	"fmt"
	"github.com/pebbe/zmq4"
)

func main() {
	receiver, _ := zmq4.NewSocket(zmq4.PULL)
	defer receiver.Close()
	receiver.Connect("tcp://localhost:5557")

	for {
		task, _ := receiver.Recv(0)
		fmt.Printf("Processing task: %s\n", task)
	}
}

高级特性

多部分消息

// 发送多部分消息
socket.Send("header", zmq4.SNDMORE)
socket.Send("body", 0)

// 接收多部分消息
part1, _ := socket.Recv(0)
more, _ := socket.GetRcvmore()
if more {
    part2, _ := socket.Recv(0)
}

设置选项

// 设置超时
socket.SetRcvtimeo(time.Second * 5)

// 设置身份标识
socket.SetIdentity("worker1")

// 设置高水位线
socket.SetRcvhwm(1000)

错误处理最佳实践

func createSocket(socketType zmq4.Type, endpoint string) (*zmq4.Socket, error) {
    socket, err := zmq4.NewSocket(socketType)
    if err != nil {
        return nil, fmt.Errorf("failed to create socket: %v", err)
    }
    
    err = socket.Bind(endpoint)
    if err != nil {
        socket.Close()
        return nil, fmt.Errorf("failed to bind to %s: %v", endpoint, err)
    }
    
    return socket, nil
}

性能优化建议

1. 复用Socket对象而不是频繁创建销毁
2. 合理设置高水位线(HWM)防止内存溢出
3. 考虑使用goroutine处理消息提高并发能力
4. 对于高吞吐场景，使用PUSH/PULL模式比REQ/REP更高效

zmq4库提供了ZeroMQ的全部功能，以上示例展示了最常见的几种模式。根据实际需求，你可以组合这些模式构建更复杂的分布式系统架构。