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

Question

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

然后我在 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

这似乎至少部分是一个错误(或难以理解的行为)，因此我建议您在 repo 协议(protocol)上创建一个问题。可能与: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 代码之前打开套接字，它也使得异步线程在一次写入后阻塞。希望您可以调整它以适合您的用例。