1

基于之前关于 SO 的问题，建议我使用 DEALER/ROUTER 模型来最大化性能（而不是 REQ/REP 模型），我设置了以下客户端和服务器代码。

客户端 asynccli.c 源触发 8 个线程，每个线程在 zmq TCP 套接字上发送和接收。服务器 asyncsrv.c 触发 4 个工作线程并使用代理将传入请求分发给工作线程。

对于持续 10 秒的测试，我体验到的性能范围从 40000 条消息到 120000 条，最多 12000 条消息/秒，这是相当低的。我在具有 8GB 内存的 i7（8HT 内核）笔记本电脑上运行 Ubuntu。我使用 czmq 库。

我认为我可以使用 ZeroMQ 实现 > 200,000 msgs/s。我想我没有正确地抓住“异步”的东西。周围有任何 C 示例代码吗？基本上我看不到如何获得异步的东西，因为我目前在这里使用 zmq_poll()。

asynccli.c：

// results : 4000/s
#include "czmq.h"
int id = 0;

static void *
client_task (void *args)
{
    zctx_t *ctx = zctx_new ();
    void *client = zsocket_new (ctx, ZMQ_DEALER);

    char identity [10];

    sprintf (identity, "%d", id);
    zsockopt_set_identity (client, identity);
    zsocket_connect (client, "tcp://localhost:5570");

    zmq_pollitem_t items [] = { { client, 0, ZMQ_POLLIN, 0 } };
    int request_nbr = 0;
    while (true) {
        //  Tick once per second, pulling in arriving messages
        int centitick;
        for (centitick = 0; centitick < 100; centitick++) {
            zmq_poll (items, 1, 1);
            if (items [0].revents & ZMQ_POLLIN) {
                zmsg_t *msg = zmsg_recv (client);
                //zframe_print (zmsg_last (msg), identity);
                zmsg_destroy (&msg);
                break;
            }
        }

        id+=1;
        zstr_send (client, "request #%d", ++request_nbr);
    }
    zctx_destroy (&ctx);
    return NULL;
}

//  The main thread simply starts several clients and a server, and then
//  waits for the server to finish.

int main (void)
{

    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);
    zthread_new (client_task, NULL);

    zclock_sleep (10 * 1000);    //  Run for 10 seconds then quit
    printf ("\\ntotal iterations = %d\n" , id );
    return 0;
}

asyncsrv.c：

#include "czmq.h"

static void server_worker (void *args, zctx_t *ctx, void *pipe);

void *server_task (void *args)
{
    //  Frontend socket talks to clients over TCP
    zctx_t *ctx = zctx_new ();
    void *frontend = zsocket_new (ctx, ZMQ_ROUTER);
    zsocket_bind (frontend, "tcp://*:5570");

    //  Backend socket talks to workers over inproc
    void *backend = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_bind (backend, "inproc://backend");

    //  Launch pool of worker threads, precise number is not critical
    int thread_nbr;
    for (thread_nbr = 0; thread_nbr < 3; thread_nbr++)
        zthread_fork (ctx, server_worker, NULL);

    //  Connect backend to frontend via a proxy
    zmq_proxy (frontend, backend, NULL);

    zctx_destroy (&ctx);
    return NULL;
}

static void
server_worker (void *args, zctx_t *ctx, void *pipe)
{
    void *worker = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_connect (worker, "inproc://backend");

    while (true) {
        //  The DEALER socket gives us the reply envelope and message
        zmsg_t *msg = zmsg_recv (worker);
        zframe_t *identity = zmsg_pop (msg);
        zframe_t *content = zmsg_pop (msg);
        assert (content);
        zmsg_destroy (&msg);

        //  Sleep for some fraction of a second
        zframe_send (&identity, worker, ZFRAME_REUSE + ZFRAME_MORE);
        zframe_send (&content, worker, ZFRAME_REUSE);

        zframe_destroy (&identity);
        zframe_destroy (&content);
    }
}

int main (void)
{
    zthread_new (server_task, NULL);
    zclock_sleep (15 * 1000);    //  Run for 15 seconds then quit
    return 0;
}
4

2 回答 2

3

您必须在工人中使用与经销商相同的逻辑：

while(1)

zmq_pollitem_t items [] = { { worker, 0, ZMQ_POLLIN, 0 } };

zmq_poll (items, 1, 1);

if (items [0].revents & ZMQ_POLLIN)

如果没有 if 语句， zmsg_recv() 会阻塞。此外，zmq_pollitem_t 应该在每次迭代时在 while 内重新创建。我认为这是低层次的原因，看看使用 FD_SET 和 select 的原始套接字，它可能会给你一个线索......。

此外，这也是不正确的：

for (centitick = 0; centitick < 100; centitick++) {

如果您只想测量 100 次迭代，请使用一些计数器。

于 2013-07-23T13:30:29.470 回答
3

问题是：通过在代码中的“发送”之前进行“读取”，客户端的发送能力受到“限制”。

现在，客户端中的代码是：

while(true)
{
  pull_any_income_messages()
  send()
}

如果有任何待处理的传入消息，这将严重限制客户端发送任何内容。所以，这本质上变成了一个请求-回复模式。

要扩展这一点，您必须将“拉入消息”和“发送”部分解耦。一种方法可能是，不是让通用客户端线程同时处理发送和接收，而是为客户端创建两个单独的线程类型，一个专门发送，另一个专门读取。

另一种方法是进行“基于信用的流量控制”。ZMQ 指南中的第 7 章有关于它的信息 ( http://zguide.zeromq.org/page:all#Chapter-Advanced-Architecture-using-MQ )。

-GK

http://gk.palem.in/

于 2014-04-23T09:43:00.413 回答