Julia ZMQ对接外部WebSocket时触发StateError错误问题求解

1. WebSocket逻辑结构错误：注释的@async死循环每次都会新建一个WebSocket连接，执行完单次写入后就退出do块自动关闭连接，高频创建销毁套接字会占用大量系统套接字资源，和ZMQ底层依赖的libzmq的套接字管理逻辑产生冲突，libzmq误将不属于自己管理的WebSocket套接字当作自身资源操作，就会抛出Socket operation on non-socket错误，Windows平台套接字实现差异导致报错为未知错误。
2. 资源初始化顺序错误：WebSocket异步协程放在ZMQ资源初始化之前执行，无sleep的死循环会占满当前线程的协程调度资源，导致ZMQ的上下文、套接字初始化和连接操作被抢占，资源初始化不完整时调用发送接口就会触发状态异常。
3. 异步逻辑无等待机制：注释的异步块内部没有添加任何休眠逻辑，死循环占满CPU调度，导致ZMQ的IO操作无法正常获得执行时间片，进一步加剧资源冲突问题。

按要求调整代码逻辑即可解决：

• 调整WebSocket逻辑结构：将接收/保活逻辑放到WebSockets.open的do块内部，复用同一个连接，仅在连接断开时才触发重连，避免高频创建销毁套接字。
• 调整资源初始化顺序：先完成ZMQ的上下文、套接字初始化和连接操作，再启动WebSocket异步协程，避免ZMQ资源初始化被抢占。
• 给异步逻辑增加合理的休眠和错误捕获，避免空转占满调度资源，同时处理连接异常自动重连。

修正后的示例代码如下：

using ZMQ
using WebSockets
using JSON3

const uri = "wss://ws.okex.com:8443/ws/v5/public"

function produce_string()
    return "hi"
end

function main()
    payload = Dict(
            :op => "subscribe",
            :args => [
                Dict(
                    "channel" => "books50-l2-tbt",
                    "instType" => "Futures",
                    "instId" => "FIL-USD-220325",
                ),
            ],
        )
    
    # 先完成ZMQ资源初始化，再启动异步协程
    ctx = Context()
    zmq_socket = Socket(ctx, PUB)
    addr = "tcp://localhost:8093"
    ZMQ.connect(zmq_socket, addr)
    sleep(3)
    ZMQ.send(zmq_socket, "hi")

    # 调整后的WebSocket异步逻辑
    @async while true
        try
            WebSockets.open(uri) do ws
                # 连接建立后仅发送一次订阅请求
                if isopen(ws)
                    write(ws, JSON3.write(payload))
                end
                # 连接保活/接收逻辑放在块内部，复用连接
                while isopen(ws)
                    data, read_success = readguarded(ws)
                    if !read_success
                        break
                    end
                    # 此处可添加收到WebSocket消息后的处理逻辑
                    sleep(0.01)
                end
            end
        catch e
            @warn "WebSocket连接异常，1秒后重连" exception=e
            sleep(1)
        end
    end

    # 原有ZMQ发送逻辑
    while true
        my_string = produce_string()
        ZMQ.send(zmq_socket, my_string)
        println("sent")
        sleep(1)
    end

end

main()

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