最佳进程间消息传递模式,用于定期侦听同一本地主机上的多个进程

我有一个奇怪的情况,我需要从我的主进程创建许多进程。

我创建的这些进程将对来自web套接字的一些消息进行排队。

我的主要流程的最小示例是:

Socket*[] socketList;

string sRecv( Socket* socket)
{
    ubyte[256] buffer;
    immutable size = socket.receive(buffer);
    import std.algorithm: min;
    return buffer[0 .. min(size,256)].idup.asString();
}

void startServer( string servername )
{
    auto pid = spawnProcess(["/home/erdem/eclipse-workspace/WebSocketDenemesi/websocketdenemesi",
                              servername, "\n"]);
    auto requester = new Socket(SocketType.req);
    auto allName = "ipc:///tmp/" ~  servername;
    requester.connect(allName);
    socketList ~= requester;

}

void main() {

    import std.array : split;
    import std.algorithm : each;

    startServer("iotabtc@depth");
    startServer("iotabtc@aggTrade");
    startServer("ethbtc@depth");

    int counter = 30;
    while(counter--) {
        foreach ( requester; socketList)
        {
            requester.send("send"); 
        }

        foreach ( requester; socketList)
        {
            auto strList = sRecv(requester).split("\n");
            strList.each!( str => writefln("Received [%d]reply [%s]", strList.length,  str) );

        }
        sleep(1000.msecs);
    }
    foreach ( requester; socketList)
    {
        requester.send("done"); 
    }
}

对于我的小流程,我有一个最小的例子:

WebSocket startSocket( string temp )
{
    auto ws_url = URL(temp);
    auto ws = connectWebSocket(ws_url);
    if ( !ws.connected )
        return null;    
    return  ws;
}

void close( WebSocket ws )
{
    int timeOut = 5;
    while ( ws && ws.connected && timeOut-- )
    {
        vibe.core.concurrency.async( { ws.close(); return true;} ); 
        sleep(5.msecs);
    }   
}

string sRecv(ref Socket socket)
{
    ubyte[256] buffer;
    immutable size = socket.tryReceive(buffer)[0];
    import std.algorithm: min;
    return size ? buffer[0 .. min(size,256)].idup.asString() : "";
}

void main( string[] args ) {

    auto responder = Socket(SocketType.rep);
    string seperatorChar = args[2];
    string temp = "ipc:///tmp/" ~ args[1];
    responder.bind(temp);

    string socketName =  "wss://stream.binance.com:9443/ws/" ~ args[1];
    auto curSocket = startSocket(socketName);
    string curString;
    while (true) {
        auto result = responder.sRecv();
        if ( result == "send")
        {
            responder.send(curString);      
            curString = "";
        }
        else if ( result == "done" )
        {
            break;
        }
        else 
        {
            if ( curSocket.dataAvailableForRead )
            {
                auto text = curSocket.receiveText();
                if ( !curString.empty )
                   curString ~= seperatorChar;
                curString ~= text;
            }
        }
        sleep(100.msecs);
    }
    writeln( "Shutting down: ", args[1]);
    curSocket.close();

}

REQ/REP 插座。有没有更好的方法来达到我的要求。例如,是否有更好的消息传递模式?例如,是否有一种模式,其中我的小进程不被 responder.receive( buffer ); .

如果有的话,我就不需要从另一个线程监听websocket了。