高效能非同步網路訊息處理框架架構設計與實作

高效能非同步網路訊息處理框架架構設計與實作

近期重構了 kendynet 網路框架核心代碼，大幅精簡介面設計，提升開發者體驗。框架採三層分層架構，支援多執行緒 Poller 模型與靈活的訊息分發機制，適合高併發伺服器開發。

三層架構設計

1. 原始資料流層：單執行緒、無封包處理、無定時器，提供最基礎的 Socket 讀寫介面，供上層二次封裝。
2. 連線與封包層：單執行緒，內建 Connection 抽象、封包編解碼、收發超時管理。
3. 非同步網路邏輯層：核心層，抽象三大型別：
   • asynnet_t：網路引擎，建構時指定 pollercount，每個 Poller 獨立執行緒運行 epoll/kqueue/IOCP 事件迴圈。
   • sock_ident：邏輯層操作的 Socket 句柄封裝，執行緒安全，可跨執行緒傳遞。
   • msgdisp_t：訊息分發器，內建訊息佇列，接收網路執行緒推送的封包事件。

訊息分發器兩種運作模式

模式一：執行緒池模式

單一 msgdisp_t，多個邏輯執行緒共同呼叫 msg_loop(disp, timeout) 搶佔式處理訊息，適合無狀態業務邏輯。

模式二：共用網路層多服務模式

進程內啟動 N 個執行緒，各跑不同服務（如閒聊、遊戲、支付），共用底層 asynnet_t。每個執行緒建立專屬 msgdisp_t，網路層依據協定 ID 將訊息路由至對應分發器，實現「網路層複用、邏輯層隔離」。

非同步 Echo Server 完整範例

#include <stdio.h>
#include <stdlib.h>
#include "core/msgdisp.h"
#include "testcommon.h"

uint32_t recvsize = 0, recvcount = 0;

void asynconnect(msgdisp_t disp, sock_ident sock, const char *ip, int32_t port) {
    printf("asynconnect\n");
    disp->bind(disp, sock, 1, 0, 30*1000); // raw=1, send_timeout=30s
}
void asynconnected(msgdisp_t disp, sock_ident sock, const char *ip, int32_t port) {
    printf("asynconnected\n");
    ++client_count;
}
void asyndisconnected(msgdisp_t disp, sock_ident sock, const char *ip, int32_t port, uint32_t err) {
    --client_count;
}
int32_t asynprocesspacket(msgdisp_t disp, sock_ident sock, rpacket_t rpk) {
    recvsize += rpk_len(rpk);
    recvcount++;
    asyn_send(sock, wpk_create_by_other((struct packet*)rpk)); // 原封回傳
    return 1;
}
void asynconnectfailed(msgdisp_t disp, const char *ip, int32_t port, uint32_t reason) {}

int main(int argc, char **argv) {
    setup_signal_handler();
    InitNetSystem();

    asynnet_t asynet = asynnet_new(1); // 1 個 Poller 執行緒
    msgdisp_t disp = new_msgdisp(asynet,
        asynconnect, asynconnected, asyndisconnected,
        asynprocesspacket, asynconnectfailed);

    int32_t err = 0;
    disp->listen(disp, argv[1], atoi(argv[2]), &err); // IP, Port

    uint32_t tick = GetSystemMs(), now;
    while (!stop) {
        msg_loop(disp, 50); // 50ms 超時輪詢
        now = GetSystemMs();
        if (now - tick > 1000) {
            uint32_t elapse = now - tick;
            recvsize = (recvsize / elapse) / 1000; // MB/s
            printf("clients:%d, throughput:%u MB/s, pkts:%u\n",
                   client_count, recvsize, recvcount);
            tick = now;
            recvcount = 0; recvsize = 0;
        }
    }
    CleanNetSystem();
    return 0;
}

專案現況與後續規劃

• 目前僅支援 Linux TCP，代碼開源於 GitHub。
• 近期重點：完善 TCP 穩定性、支援 UDP、Windows IOCP 移植。
• 中長期：在框架之上實作 User-level Thread（協程）與 RPC 框架，進一步簡化業務邏輯開發。