Сокет ZeroMQ не получает данные, когда находится в программе - PullRequest
Новая
       19

Сокет ZeroMQ не получает данные, когда находится в программе

1 голос
/ 01 июля 2019

Я настраиваю сокет в Go для отправки и получения данных в API-интерфейс, который использует ZeroMQ для связи. Нам нужно использовать горутины. Я понимаю, что сокеты ZMQ крайне не поточнобезопасны, и для обеспечения их правильной работы все отправка и получение должны выполняться в одном потоке.

Я могу получить все данные из всех запросов API, за исключением одного фрагмента данных из одного конкретного вызова API - пульса, на который мы должны ответить, чтобы поддерживать соединение.

В этом API можно сделать несколько запросов, которые могут привести к отправке данных назад, а также к запросу сердцебиения. Каждый другой запрос API будет отправлять данные обратно, подождать около 5 секунд, а затем отправить запрос сердцебиения, необходимый для поддержания связи. Когда код запускается с использованием одного потока, мы можем получать пульс от каждого вызова API. Однако, когда я запускаю ту же функцию через процедуру, я получаю запрос сердцебиения от каждой функции API, КРОМЕ для одной из них.

Настройка сокета 

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

func testCode() {
    context, err := zmq.NewContext()

    if err != nil {
        fmt.Println("context err: ", err)
        return
    }

    //create Request Socket to talk to Request Server
    fmt.Fprint(os.Stdout,"Attempting to connect to Request Server...")
    sock, err := context.NewSocket(zmq.DEALER)
    if err != nil {
        fmt.Println("socket err: ", err)
        return
    }

    connection.RequestSocket = sock

    err = connection.RequestSocket.Connect(REQUEST_URL)
    if err != nil {
        fmt.Println("connect err: ", err)
        return
    }
}


//build the data to be sent up here

    ...

    _, err := socket.SendBytes(bufToSend, 0)
    if err != nil {
        fmt.Println("send err: ", err)
        return
    }

//receive all necessary data from the server
    for mask & info == 0 { 
        reply, err := socket.RecvBytes(0)
        if err != nil {
            fmt.Println("err: ", err)
            return
        }
        //process the data
        ...
}

Итак, когда я запускаю эту основную функцию в другом файле, мы получаем все правильные запросы данных и пульс от каждого запроса API 

func main() {
    testCode()
    for {
    }
}

Однако, когда я делаю testCode () программой (как показано ниже), я не получаю запрос сердцебиения только для одного из вызовов API 

func main() {
    go testCode()
    for {
    }
}

Интересно также то, что я отслеживал все входящие и исходящие TCP-порты, через которые мы общаемся, и подтвердил, что запросы пульса отправляются API. Итак, запрос сердцебиения отправляется, данные распознаются Wireshark как сообщение, возвращаемое от API, но я не получаю его через сокет для одного конкретного вызова API. Каждый другой вызов API отправляет запрос сердцебиения, и я вижу в своей программе, что я их получил.

Я убедился, что все создание, отправка и получение для сокета были выполнены в одном потоке. Меня смущает, почему эта проблема возникает и почему она возникает только для одного из вызовов API, а не для всех из них.

ОБНОВЛЕНИЕ: Очевидно, что для вызова API, который не получает пульс, я не получаю все данные от API. Я получаю около 18910 сообщений из 36872 сообщений, которые необходимо отправить. Я не знаю, почему через некоторое время он просто перестает посылать сообщения, когда в горутине

...