我无法使用普通的零MQ推/拉模式在拉取实例上获取/打印数据

Question

我在获取和打印成功推送到数据的数据时遇到困难。我已经建立了一个简单的推挽式架构，其中Metatrader 4充当producer，Python后端充当consumer。

关于获取和打印数据，我只是无法实现。

这是我的push实例，可以正常工作：

Metatrader 4 zeroMQ推送实例：

#include <Zmq/Zmq.mqh>

// EA plot settings

extern string PROJECT_NAME = "Dashex.Feeder";
extern string ZEROMQ_PROTOCOL = "tcp";
extern string HOSTNAME = "localhost";
extern int PUSH_PORT = 32225;

extern string t0 = "--- Feeder Parameters ---";
input string DID = "insert your DID here";
extern string t1 = "--- ZeroMQ Configuration ---";
extern bool Publish_MarketData = false;

// ZeroMQ environment //

// CREATE ZeroMQ Context
Context context(PROJECT_NAME);

// CREATE ZMQ_PUSH SOCKET
Socket pushSocket(context, ZMQ_PUSH);

string Publish_Symbols[7] = {
   "EURUSD","GBPUSD","USDJPY","USDCAD","AUDUSD","NZDUSD","USDCHF"
};

//+------------------------------------------------------------------+
//| Expert initialization function                                   |
//+------------------------------------------------------------------+

int OnInit()
  {
//---

   EventSetTimer(1);     // Set Millisecond Timer to get client socket input

   context.setBlocky(false);

   // Send responses to PULL_PORT that client is listening on.
   Print("[PUSH] Connecting MT4 Server to Socket on Port " + IntegerToString(PUSH_PORT) + "..");
   pushSocket.connect(StringFormat("%s://%s:%d", ZEROMQ_PROTOCOL, HOSTNAME, PUSH_PORT));

   pushSocket.setSendHighWaterMark(1);
   pushSocket.setLinger(0);   

//---
   return(INIT_SUCCEEDED);
  }

//+------------------------------------------------------------------+
//| Expert deinitialization function                                 |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
//---

   Print("[PUSH] Disconnecting MT4 Server from Socket on Port " + IntegerToString(PUSH_PORT) + "..");
   pushSocket.disconnect(StringFormat("%s://%s:%d", ZEROMQ_PROTOCOL, HOSTNAME, PUSH_PORT));

   // Shutdown ZeroMQ Context
   context.shutdown();
   context.destroy(0);

   EventKillTimer();
}
//+------------------------------------------------------------------+
//| Expert tick function                                             |
//+------------------------------------------------------------------+
void OnTimer()
{
   /*
      Use this OnTimer() function to send market data to consumer.
   */
   if(!IsStopped() && Publish_MarketData == true)
   {
      for(int s = 0; s < ArraySize(Publish_Symbols); s++)
      {

         string _tick = GetBidAsk(Publish_Symbols[s]);
         Print("Sending " + Publish_Symbols[s] + " " + _tick + " to PUSH Socket");
         ZmqMsg reply(StringFormat("%s %s", Publish_Symbols[s], _tick));
         pushSocket.send(reply, true);
      }
   }
}
//+------------------------------------------------------------------+

string GetBidAsk(string symbol) {

   MqlTick last_tick;

   if(SymbolInfoTick(symbol,last_tick))
   {
       return(StringFormat("%f;%f", last_tick.bid, last_tick.ask));
   }

   // Default
   return "";
}

如何将数据推送到线路（上述代码中的字符串数组，代码段）：

         string _tick = GetBidAsk(Publish_Symbols[s]);
         Print("Sending " + Publish_Symbols[s] + " " + _tick + " to PUSH Socket");
         ZmqMsg reply(StringFormat("%s %s", Publish_Symbols[s], _tick));
         pushSocket.send(reply, true);

Python侧拉客户端，它不会从网络中获取/打印任何数据：

import zmq
import time
from time import sleep

context = zmq.Context()
zmq_socket = context.socket(zmq.PULL)
zmq_socket.bind("tcp://*:32220")
time.sleep(1)

while True:

    def pull_MT4():

        try:
            msg = zmq_socket.recv_string()
            print(msg)
            return msg
        except zmq.error.Again:
            print("\nResource timeout.. please try again.")
            sleep(0.000001)

        return None

没有打印数据的控制台输出：-（

Answer 1

在while True内部，您一次又一次定义函数pull_MT4，但从未执行此函数-从未使用pull_MT4()运行它。

您应直接在循环中使用此代码（不带定义功能且不带return）

while True:
    try:
        msg = zmq_socket.recv_string()
        print(msg)
    except zmq.error.Again:
        print("\nResource timeout.. please try again.")
        sleep(0.000001)

或者您应该在循环之前定义函数并在循环内部执行

def pull_MT4(): # define function
    try:
        msg = zmq_socket.recv_string()
        print(msg)
        return msg
    except zmq.error.Again:
        print("\nResource timeout.. please try again.")
        sleep(0.000001)
    return None

while True:
    pull_MT4() # execute function