高消息速率下pebbe/zmq4代理CPU占用过高问题排查优化

• 验证目标：测试ZMQ v4作为IPC场景消息代理的可行性，测试使用封装ZMQ原生C库的Go语言实现pebbe ZMQ
• 测试架构：采用XPUB-XSUB代理模式，发布端、订阅端直连代理节点，分别在1500消息/秒、10000消息/秒发送速率下压测。已知该Go库为薄封装层，实际消息收发逻辑通过cgo调用C层代码实现
• 测试环境：运行在ARM架构设备上，1500消息/秒速率下代理进程CPU占用达40%-50%（测试环境单核心满负载为100%占用），运行时内存占用约900MB物理内存+100MB交换分区。因缺少对应性能基准，暂无法判断该占用水平是否合理
• 现有排查结论：CPU profile结果显示runtime.cgocall与runtime._ExternalCode占用绝大多数CPU资源，profile结果截图如下：
  
  因缺乏性能调优经验，暂未从现有profile结果找到优化切入点，需要可落地的cgo调用、外部代码块CPU占用优化方案。

ZMQ Broker（代理节点） #

import (
    "fmt"

    zmq "github.com/pebbe/zmq4"
    "github.com/pkg/profile"
)

func main() {
    defer profile.Start(profile.CPUProfile, profile.ProfilePath(".")).Stop()
    fmt.Println("Setting up XSUB socket")
    subscriberSocket, err := zmq.NewSocket(zmq.XSUB)
    if err != nil {
        fmt.Println("Error when creating XSUB socket -> ", err)
    }
    defer subscriberSocket.Close()

    err = subscriberSocket.Bind("tcp://127.0.0.1:8101")
    if err != nil {
        fmt.Println("Error when binding XSUB socket -> ", err)
    } else {
        fmt.Println("Succesfully accepting incoming connections on XSUB socket")
    }

    fmt.Println("Setting up XPUB socket")

    publisherSocket, err := zmq.NewSocket(zmq.XPUB)
    if err != nil {
        fmt.Println("Error when creating XPUB socket -> ", err)
    }
    defer publisherSocket.Close()

    err = publisherSocket.Bind("tcp://127.0.0.1:8100")
    if err != nil {
        fmt.Println("Error when binding XPUB socket -> ", err)
    } else {
        fmt.Println("Succesfully accepting incoming connections on XPUB socket")
    }

    err = zmq.Proxy(publisherSocket, subscriberSocket, nil)
    if err != nil {
        fmt.Println("Failed to start the XPUB XSUB broker -> ", err)
    }
}

Publisher（发布端） #

import (
    "time"

    zmq "github.com/pebbe/zmq4"

    "fmt"
)

func main() {
    publisher, err := zmq.NewSocket(zmq.PUB)
    if err != nil {
        fmt.Println("error when connecting to a pub socket -> ", err)
    }

    defer publisher.Close()

    err = publisher.Connect("tcp://127.0.0.1:8101")
    if err != nil {
        fmt.Println("error when connecting to a pub socket -> ", err)
    }

    for range time.Tick(time.Microsecond * 500) {
        sendToAll(publisher)
    }
}

func sendToAll(pub *zmq.Socket) {
    var message = "topicA test"
    _, err := pub.Send(message, zmq.DONTWAIT)
    if err != nil {
        println("error when sending message-> ", err)
    }
}

Subscriber（订阅端） #

import (
    "os"
    "strconv"

    zmq "github.com/pebbe/zmq4"

    "fmt"
)

func main() {
    //  Socket to talk to server
    fmt.Println("Collecting updates from broker...")
    subscriber, err := zmq.NewSocket(zmq.SUB)
    if err != nil {
        fmt.Println("error when opening new socket to SUB -> ", err)
    }
    defer subscriber.Close()
    err = subscriber.Connect("tcp://127.0.0.1:8100")
    if err != nil {
        fmt.Println("error when connecting to XSUB port -> ", err)
    }

    err = subscriber.SetSubscribe("topicA ")
    if err != nil {
        fmt.Println("error when setting subscription filter -> ", err)
    }
    i := 0
    for {
        msg, err := subscriber.Recv(0)
        if err != nil {
            fmt.Println("error when reciveing subscription info -> ", err)
            os.Exit(1)
        }
        i += 1
        fmt.Println(msg + "\n -> count is" + strconv.Itoa(i))
    }
}

交叉编译参数 #

使用如下参数交叉编译，暂不确定编译配置是否影响最终性能：

#!/bin/bash

ARM_PREFIX="arm-linux-androideabi-" 

TOOLCHAIN_PATH="/home/NDK/arm"

CC="${TOOLCHAIN_PATH}/bin/${ARM_PREFIX}gcc" \
CFLAGS="-march=armv7-a -mfpu=neon" \
GOOS=android \
GOARCH=arm \
GOARM=7 \
CGO_ENABLED=1 \
PKG_CONFIG_PATH="${TOOLCHAIN_PATH}/lib/pkgconfig" \
go build -o test

自定义代理实现 #

自行实现了一版代理逻辑（推测pebbe/zmq4内置的zmq.Proxy逻辑与该实现近似），测试后CPU占用水平与直接调用zmq.Proxy基本一致，代码如下：

for {
    // this will block forever till an event occurs
    sockets, err := poller.Poll(-1)
    if err != nil {
        fmt.Println("error when establishing xpubsub poller")
    }
    for _, socket := range sockets {
        switch s := socket.Socket; s {
        case publisherSocket:
            msg, err := s.Recv(0)
            if err != nil {
                fmt.Println("error when recieving on publisherSocket")
            }
            subscriberSocket.Send(msg, zmq.DONTWAIT)
        case subscriberSocket:
            msg, err := s.Recv(0)
            if err != nil {
                fmt.Println("error when recieving on subscribersocket")
            }
            publisherSocket.Send(msg, zmq.DONTWAIT)
        }
    }
}

占用合理性判断 #

当前CPU、内存占用完全不合理。同配置ARM设备上，原生C实现的ZMQ XPUB-XSUB代理处理1500消息/秒的流量时CPU占用应低于5%，10000消息/秒场景下也不会超过15%，空载内存占用应在10MB级别。
阻塞poller低占用的预期逻辑本身没错，当前高占用来自Go/C跨边界调用的冗余开销，而非poller本身：Go层实现的代理每转发1条消息，至少触发4次cgo上下文切换（poll唤醒、收消息、发消息、回到poll阻塞），ARM架构下单次cgo切换开销约100-200ns，叠加未开编译优化、非阻塞调用自旋、debug版libzmq等问题，开销会被放大数倍。900MB级别的内存占用基本可以确认是链接了未开优化的debug版本libzmq，或存在消息持续积压。

降低cgo与外部代码CPU开销的可行方案 #

• 不要在Go层实现代理转发逻辑，直接调用libzmq原生的zmq_proxy接口。原生C实现的代理逻辑全部运行在C层，初始化完成后不需要反复做Go/C上下文切换，可直接消除转发路径上的所有cgo开销。注意pebbe/zmq4的旧版本zmq.Proxy封装存在问题，会在Go层做消息循环而非直接调用C层原生接口，可升级到最新版本验证，或自行编写cgo封装直接调用libzmq的原生代理函数。
• 优化编译参数：给CFLAGS增加-O2 -flto开启C编译器优化与链接时优化，Go编译增加-ldflags="-s -w"去掉调试符号；同时确认链接的libzmq库是用相同优化参数编译的release版本，debug版本libzmq自带大量断言、运行时检查，CPU、内存开销会比release版本高2-3倍。
• 替换本地传输协议：同机IPC场景将传输层从tcp://127.0.0.1替换为ipc://本地套接字，省去TCP协议栈封包解包、校验计算的开销，C层处理性能可提升30%以上。
• 去掉收发逻辑中的DONTWAIT非阻塞标志：非阻塞模式下socket缓冲区状态不满足收发条件时会立刻返回EAGAIN错误，若无退避逻辑会触发大量无效的自旋cgo调用，平白消耗CPU。代理转发路径上配合阻塞poller使用阻塞式收发即可，不会出现卡死问题。
• 发布端开启消息批量发送：不要每生成一条小消息就立刻调用Send，可攒够10-20条消息或攒满1ms窗口再批量发送，摊薄单次cgo调用的固定开销。

cgo封装网络库性能调优参考方向 #

• 热路径逻辑尽量下沉到C层：高频调用的消息收发、转发、协议解析逻辑不要放在Go层实现，cgo固定切换开销在高频场景下占比极高，每秒10万次cgo调用即可占满单核心10%以上的CPU资源。
• 减少跨边界内存拷贝与指针传递：Go向C传递字符串、字节切片时会触发内存拷贝，大消息场景下开销明显；Go 1.14之后版本对cgo传递Go指针有运行时栈扫描检查，高频调用下该检查开销占比很高，尽量避免在热路径传递Go指针。
• 分层做性能分析：当pprof显示runtime._ExternalCode占比高时，说明CPU时间主要消耗在C层，Go侧pprof无法采集C层内部调用栈，需要配合perf、gprof等原生性能工具分析C代码的瓶颈点。
• 始终做原生基准对照：同逻辑先跑纯C实现的版本做性能基线，再对比Go封装版本的表现，可快速区分开销是来自底层库本身还是封装层。

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