Julia ZMQ - 连接到其他 WebSockets 产生 StateError

Question

我正在尝试使用 ZMQ 将许多发布者连接到一个订阅者 (python)。这是一个这样的发布者（我使用连接而不是绑定，因为订阅者绑定）。在我取消阻止下面的注释代码之前，代码工作正常。

然后我在 Windows 收到此错误：

LoadError: StateError("Unknown error")

在 Ubuntu:

StateError("Socket operation on non-socket")

源代码：

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",
                ),
            ],
        )
    # Unblock this code to produce error
    # @async while true
    #     WebSockets.open(uri) do ws
    #         confirmation = true
    #         if isopen(ws)
    #             write(ws, JSON3.write(payload))
    #         end
    #     end
    # end

    ctx = Context()
    zmq_socket = Socket(ctx, PUB)
    addr = "tcp://localhost:" * string(8093)
    ZMQ.connect(zmq_socket, addr)
    sleep(3)
    ZMQ.send(zmq_socket, "hi")

    while true
        my_string = produce_string()
        ZMQ.send(zmq_socket, my_string)
        println("sent")
        sleep(1)
    end

end

main()

Answer 1

这似乎至少部分是一个错误（或难以理解的行为）所以我建议你在回购上创建一个问题。可能与：Test Error: Assertion failed: Socket operation on non-socket #147.

有关

但是，我们可以尽最大努力了解问题出在哪里，或许可以找到解决方法。由于 ZMQ.jl 使用 libzmq 在低级别处理套接字，它可能会干扰 Julia 对文件描述符的处理，我们可能有一个 race condition。让我们通过稍微修改您的代码来验证该假设：

    @async WebSockets.open(uri) do ws
        while true
            if isopen(ws)
                msg = JSON3.write(payload)
                write(ws, msg)
                display(ws.socket.bio)
                break
            end
        end
    end

    sleep(0.1)
    ctx = Context()
    zmq_socket = Socket(ctx, PUB)
    dump(zmq_socket)
    addr = "tcp://localhost:" * string(8093)
    ZMQ.connect(zmq_socket, addr)
    sleep(3)
    ZMQ.send(zmq_socket, "hi")

我只是改变了一些东西来让代码打印出必要的信息。我们看到：

Socket
  data: Ptr{Nothing} @0x0000000001d5e590
  pollfd: FileWatching._FDWatcher
    handle: Ptr{Nothing} @0x00000000018b7970
    fdnum: Int64 31
    refcount: Tuple{Int64, Int64}
      1: Int64 1
      2: Int64 0
    notify: Base.GenericCondition{Base.Threads.SpinLock}
      waitq: Base.InvasiveLinkedList{Task}
        head: Nothing nothing
        tail: Nothing nothing
      lock: Base.Threads.SpinLock
        owned: Int64 0
    events: Int32 0
    active: Tuple{Bool, Bool}
      1: Bool false
      2: Bool false

和

TCPSocket(RawFD(31) paused, 0 bytes waiting)

pollfd.fdnum 字段为 31，与 TCPSocket 文件描述符相同，所以这可能就是发生的情况。

我们能做什么？

在上面的代码中我已经对你的原始代码做了一个更改，我将 while 循环移动到 WebSockets.open 的调用中，你真的想在每个循环中打开一个新的套接字吗？其次，我们可以尝试稍微同步我们的线程，以确保在调用 ZMQ 之前我们已经完成打开套接字：

function main()
    payload = Dict(
            :op => "subscribe",
            :args => [
                Dict(
                    "channel" => "books50-l2-tbt",
                    "instType" => "Futures",
                    "instId" => "FIL-USD-220325",
                ),
            ],
        )
    msg_channel = Channel(1)
    @async WebSockets.open(uri) do ws
        while true
            if isopen(ws)
                msg = JSON3.write(payload)
                put!(msg_channel, msg)
                write(ws, msg)
            end
        end
    end

    println(take!(msg_channel))
    ctx = Context()
    zmq_socket = Socket(ctx, PUB)
    addr = "tcp://localhost:" * string(8093)
    ZMQ.connect(zmq_socket, addr)
    sleep(3)
    ZMQ.send(zmq_socket, "hi")

    while true
        my_string = produce_string()
        ZMQ.send(zmq_socket, my_string)
        println("sent")
        sleep(1)
    end
end

这里我使用了Channel在线程之间进行通信，这确保了套接字在我们继续ZMQ代码之前完成打开，它也使得异步线程在一次写入后阻塞。希望您可以调整它以适合您的用例。

Julia ZMQ - 连接到其他 WebSockets 产生 StateError

Julia ZMQ - connecting to other WebSockets produces StateError

tcp

ipc

zeromq

pyzmq

julia