【从零开始一步步学习VSOA开发】并行RPC服务端

并行RPC服务端

概念

之前 RPC 使用方式中的所有回调函数都是串行执行的，VSOA 为 C 语言开发版本提供了并行处理 RPC 请求的功能，开发者可以通过并发 RPC 相关接口创建线程池并行处理 RPC 请求。并发服务模式通过多线程来实现，能有效提高响应速度，降低排队等待时间。

程序源码

并行 RPC 服务通过vsoa_server_plistener_xxxx 系列接口来实现。
将 RPC 服务端程序修改为并行模式后，源码如下：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "vsoa_platform.h"
#include "vsoa_server.h"
#include "vsoa_plistener.h"

#define MY_SERVER_ADDR                      "0.0.0.0"
#define MY_SERVER_PORT                      (4001)
#define MY_SERVER_NAME                      "{\"name\":\"echo_server\"}"
#define MY_SERVER_PASSWD                    "123456"

#define PLISTEN_THREAD_COUNT                4

static void command_echo (void *arg, vsoa_server_t *server, vsoa_cli_id_t cid,
                          vsoa_header_t *vsoa_hdr, vsoa_url_t *url, vsoa_payload_t *payload)
{
    struct sockaddr_in addr_in;
    socklen_t          namelen = sizeof(addr_in);
    LW_OBJECT_HANDLE  tid = API_ThreadIdSelf();

    vsoa_server_cli_address(server, cid, (struct sockaddr *)&addr_in, &namelen);
    printf("[%x]client %d addr is %s:%d\n", (UINT)tid, cid, inet_ntoa(addr_in.sin_addr), ntohs(addr_in.sin_port));
    printf("message, url:%.*s, param:%.*s, data:%.*s\n",
           (int)url->url_len, url->url,
           (int)payload->param_len, payload->param,
           (int)payload->data_len, (char *)payload->data);

    vsoa_server_cli_reply(server, cid, 0, vsoa_parser_get_seqno(vsoa_hdr), 0, payload);
}

int main (int argc, char **argv)
{
    vsoa_server_t *server;

    /*
    * 创建服务端
    */
    server = vsoa_server_create(MY_SERVER_NAME);
    if (!server) {
        fprintf(stderr, "Can not create VSOA server!\n");
        return  (-1);
    }

    /*
    * 设置密码，设置为NULL，表示密码为空，客户端可以不输入密码
    */
    vsoa_server_passwd(server, MY_SERVER_PASSWD);

    /*
    * 创建并行RPC服务
    * 先创建并行RPC监听器，再将RPC回调绑定到多线程，最后添加RPC服务
    */
    vsoa_url_t url;
    vsoa_plistener_t *pl;
    vsoa_plistener_handler_t handler;
    url.url     = "/echo";
    url.url_len = strlen(url.url);
    pl = vsoa_server_plistener_create(PLISTEN_THREAD_COUNT);
    handler = vsoa_server_plistener_handler(pl, true, PLISTEN_THREAD_COUNT, command_echo, NULL);
    vsoa_server_add_listener(server, &url, handler.callback, handler.arg);

    /*
    * 启动微服务
    */
    struct sockaddr_in addr;
    bzero(&addr, sizeof(struct sockaddr_in));
    addr.sin_family      = AF_INET;
    addr.sin_port        = htons(MY_SERVER_PORT);
    addr.sin_addr.s_addr = inet_addr(MY_SERVER_ADDR);
    addr.sin_len         = sizeof(struct sockaddr_in);

    if (!vsoa_server_start(server, (struct sockaddr *)&addr, sizeof(struct sockaddr_in))) {
        vsoa_server_close(server);
        fprintf(stderr, "Can not start VSOA server!\n");
        return  (-1);
    }

    /*
    * 进入监听事件循环
    */
    while (1) {
        int     cnt;
        int     max_fd;
        fd_set  fds;
        struct timespec timeout = {1, 0 };

        FD_ZERO(&fds);
        max_fd = vsoa_server_fds(server, &fds);

        cnt = pselect(max_fd + 1, &fds, NULL, NULL, &timeout, NULL);
        if (cnt > 0) {
            vsoa_server_input_fds(server, &fds);
        }
    }

    return (0);
}

运行效果

将新服务端程序替换原服务端程序继续 echo 例程测试，执行效果如下：
服务端执行效果：

客户端执行效果：