TeamTalk db_proxy_server CProxyConn詳解

1. 簡介

網絡連接配接代理，用于處理網絡連接配接，收發資料包。

2. 類與接口

CProxyConn使用需要配合兩個輔助類CHandlerMap和CProxyTask。

CHandlerMap用于将資料包commandId和相應處理函數形成映射，友善通過commandId找到相應的執行函數。

CProxyTask任務代理，可以将工作任務封裝成該類扔進工作隊列，等待工作線程處理。

CProxyConn

// 處理網絡連接配接
void OnConnect(net_handle_t handle)

// 由于資料包是在另一個線程處理的，是以不能在主線程delete資料包，是以需要Override這個方法
// 接收網絡資料，接收完資料後将有效資料包裡對應操作封裝成任務交給其他工作線程處理
void OnRead()

// 定時器處理函數，用于定時發送資料包
void OnTimer(uint64_t curr_tick)

// 根據資料包中對應的業務操作封裝成相應任務交給其他工作線程處理
void HandlePduBuf(uchar_t* pdu_buf, uint32_t pdu_len)

// 将待回複的資料包加傳入連結表，等待其他工作線程發送
void AddResponsePdu(uint32_t conn_uuid, CImPdu* pPdu)

// 将主線程待回複資料包連結清單裡内容發走，若消息包裡pPdu為空則關閉連接配接
void SendResponsePduList()

// 關閉連接配接并釋放資源
void Close()
           

// 定時器回調函數，處理每個連接配接的定時任務
void proxy_timer_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)

// loop回調函數，将主線程待回複資料包連結清單裡内容發走
void proxy_loop_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)

/*
 * 用于優雅的關閉連接配接：
 * 伺服器收到SIGTERM信号後，發送CImPduStopReceivePacket資料包給每個連接配接，
 * 通知消息伺服器不要往自己發送資料包請求，
 * 然後注冊4s後調用的回調函數，回調時再退出程序
 */
void exit_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)

// 自定義SIGTERM處理事件
static void sig_handler(int sig_no)

// 建立線程池，添加loog任務，自定義SIGTERM事件，添加定時器任務
int init_proxy_conn(uint32_t thread_num)

           

3. 作為伺服器一般流程

首先建立工作線程池，工作線程會不斷從m_task_list裡去除任務執行；建立工作線程池同時主線程會netlib_add_loop添加loop任務，不斷将待回複資料包連結清單裡内容發走；建立工作線程池同時主線程也會注冊定時任務，定時任務主要用于伺服器向用戶端發送心跳包，同時關閉掉長時間未向伺服器發送資料的用戶端連接配接；之後開時監聽用戶端連接配接，新連接配接調用_AcceptNewSocket，之後該連接配接收到資料時CBaseSocket調用m_callback，這個m_callback為netlib_listen時設定的回調函數，CProxyConn::OnConnect會重置m_callback為imconn_callback，之後都是通過imconn_callback來管理；伺服器會将用戶端請求的業務操作封裝成CProxyTask添加到m_worker_list，線程池裡工作線程會處理m_worker_list裡業務，這裡業務操作和請求資料包裡commandId通過CHandlerMap是一一映射的，可以快速的通過commandId執行相應的業務操作；業務操作會調用AddResponsePdu将伺服器待回複資料包加入消息連結清單s_response_pdu_list，主線程loop函數proxy_loop_callback會将該連結清單中待回複消息全部發走。

4. 測試

//
//  test_proxyconn.cpp
//  test_proxyconn
//
//  Created by blueBling on 22-04-29.
//  Copyright (c) 2022年blueBling. All rights reserved.
//

#include "ProxyConn.h"

#include <iostream>


using std::cout;
using std::endl;

// this callback will be replaced by imconn_callback() in OnConnect()
void proxy_serv_callback(void* callback_data, uint8_t msg, uint32_t handle, void* pParam)
{
	if (msg == NETLIB_MSG_CONNECT)
	{
		CProxyConn* pConn = new CProxyConn();
		pConn->OnConnect(handle);
	}
	else
	{
		log("!!!error msg: %d", msg);
	}
}

int test_proxyconn() {

	init_proxy_conn(3);

	netlib_listen("192.168.49.128", 7777, proxy_serv_callback, NULL);

	netlib_eventloop(10);

	return 0;
}


int main(){

	test_proxyconn();

	//這裡mysql和redis連接配接池未釋放存在記憶體洩漏問題，解決方法參考test_dbpool

	return 0;
}
           

測試結果：

5. 源碼

Github