【Lars教程目錄】
Lars源代碼
https://github.com/aceld/Lars
【Lars系統(tǒng)概述】
第1章-概述
第2章-項目目錄構(gòu)建
【Lars系統(tǒng)之Reactor模型服務(wù)器框架模塊】
第1章-項目結(jié)構(gòu)與V0.1雛形
第2章-內(nèi)存管理與Buffer封裝
第3章-事件觸發(fā)EventLoop
第4章-鏈接與消息封裝
第5章-Client客戶端模型
第6章-連接管理及限制
第7章-消息業(yè)務(wù)路由分發(fā)機(jī)制
第8章-鏈接創(chuàng)建/銷毀Hook機(jī)制
第9章-消息任務(wù)隊列與線程池
第10章-配置文件讀寫功能
第11章-udp服務(wù)與客戶端
第12章-數(shù)據(jù)傳輸協(xié)議protocol buffer
第13章-QPS性能測試
第14章-異步消息任務(wù)機(jī)制
第15章-鏈接屬性設(shè)置功能
【Lars系統(tǒng)之DNSService模塊】
第1章-Lars-dns簡介
第2章-數(shù)據(jù)庫創(chuàng)建
第3章-項目目錄結(jié)構(gòu)及環(huán)境構(gòu)建
第4章-Route結(jié)構(gòu)的定義
第5章-獲取Route信息
第6章-Route訂閱模式
第7章-Backend Thread實時監(jiān)控
【Lars系統(tǒng)之Report Service模塊】
第1章-項目概述-數(shù)據(jù)表及proto3協(xié)議定義
第2章-獲取report上報數(shù)據(jù)
第3章-存儲線程池及消息隊列
【Lars系統(tǒng)之LoadBalance Agent模塊】
第1章-項目概述及構(gòu)建
第2章-主模塊業(yè)務(wù)結(jié)構(gòu)搭建
第3章-Report與Dns Client設(shè)計與實現(xiàn)
第4章-負(fù)載均衡模塊基礎(chǔ)設(shè)計
第5章-負(fù)載均衡獲取Host主機(jī)信息API
第6章-負(fù)載均衡上報Host主機(jī)信息API
第7章-過期窗口清理與過載超時(V0.5)
第8章-定期拉取最新路由信息(V0.6)
第9章-負(fù)載均衡獲取Route信息API(0.7)
第10章-API初始化接口(V0.8)
第11章-Lars Agent性能測試工具
第12章- Lars啟動工具腳本
15) 異步消息任務(wù)機(jī)制
? 我們之前在include/task_msg.h中, 其中task的消息類型我們只是實現(xiàn)了NEW_CONN,目的是thread_pool選擇一個線程，讓一個線程里的thread_queue去創(chuàng)建一個連接對象河绽。但是并沒有對NEW_TASK的任務(wù)類型進(jìn)行定義蠢莺。這種類型是允許服務(wù)端去執(zhí)行某項具體的業(yè)務(wù)。并不是根據(jù)客戶端來消息去被動回復(fù)的業(yè)務(wù)，而是服務(wù)端主動發(fā)送的業(yè)務(wù)給到客戶端圣猎。
15.1 任務(wù)函數(shù)類型
? 我們先定義task的回調(diào)函數(shù)類型
lars_reactor/include/event_loop.h
//...
//定義異步任務(wù)回調(diào)函數(shù)類型
typedef void (*task_func)(event_loop *loop, void *args);
//...
? 為了防止循環(huán)頭文件引用，我們把typedef定義在event_loop.h中。
lars_reactor/include/task_msg.h
#pragma once
#include "event_loop.h"
//定義異步任務(wù)回調(diào)函數(shù)類型
typedef void (*task_func)(event_loop *loop, void *args);
struct task_msg
{
enum TASK_TYPE
{
NEW_CONN, //新建鏈接的任務(wù)
NEW_TASK, //一般的任務(wù)
};
TASK_TYPE type; //任務(wù)類型
//任務(wù)的一些參數(shù)
union {
//針對 NEW_CONN新建鏈接任務(wù)营袜，需要傳遞connfd
int connfd;
//針對 NEW_TASK 新建任務(wù),
//可以給一個任務(wù)提供一個回調(diào)函數(shù)
struct {
task_func task_cb; //注冊的任務(wù)函數(shù)
void *args; //任務(wù)函數(shù)對應(yīng)的形參
};
};
};
? task_func是我們定義的一個任務(wù)的回調(diào)函數(shù)類型，第一個參數(shù)當(dāng)然就是讓哪個loop機(jī)制去執(zhí)行這個task任務(wù)丑罪。很明顯荚板，一個loop是對應(yīng)一個thread線程的。也就是讓哪個thread去執(zhí)行這個task任務(wù)吩屹。args是task_func的函數(shù)形參跪另。
?
15.2 event_loop模塊添加task任務(wù)機(jī)制
? 我們知道，task綁定一個loop煤搜，很明顯免绿，一個event_loop應(yīng)該擁有需要被執(zhí)行的task集合。
? 在這里宅楞，我們將event_loop加上已經(jīng)就緒的task任務(wù)的屬性
lars_reactor/include/event_loop.h
#pragma once
/*
*
* event_loop事件處理機(jī)制
*
* */
#include <sys/epoll.h>
#include <ext/hash_map>
#include <ext/hash_set>
#include <vector>
#include "event_base.h"
#include "task_msg.h"
#define MAXEVENTS 10
// map: fd->io_event
typedef __gnu_cxx::hash_map<int, io_event> io_event_map;
//定義指向上面map類型的迭代器
typedef __gnu_cxx::hash_map<int, io_event>::iterator io_event_map_it;
//全部正在監(jiān)聽的fd集合
typedef __gnu_cxx::hash_set<int> listen_fd_set;
//定義異步任務(wù)回調(diào)函數(shù)類型
typedef void (*task_func)(event_loop *loop, void *args);
class event_loop
{
public:
//構(gòu)造针姿，初始化epoll堆
event_loop();
//阻塞循環(huán)處理事件
void event_process();
//添加一個io事件到loop中
void add_io_event(int fd, io_callback *proc, int mask, void *args=NULL);
//刪除一個io事件從loop中
void del_io_event(int fd);
//刪除一個io事件的EPOLLIN/EPOLLOUT
void del_io_event(int fd, int mask);
// ===========================================
//獲取全部監(jiān)聽事件的fd集合
void get_listen_fds(listen_fd_set &fds) {
fds = listen_fds;
}
//=== 異步任務(wù)task模塊需要的方法 ===
//添加一個任務(wù)task到ready_tasks集合中
void add_task(task_func func, void *args);
//執(zhí)行全部的ready_tasks里面的任務(wù)
void execute_ready_tasks();
// ===========================================
private:
int _epfd; //epoll fd
//當(dāng)前event_loop 監(jiān)控的fd和對應(yīng)事件的關(guān)系
io_event_map _io_evs;
//當(dāng)前event_loop 一共哪些fd在監(jiān)聽
listen_fd_set listen_fds;
//一次性最大處理的事件
struct epoll_event _fired_evs[MAXEVENTS];
// ===========================================
//需要被執(zhí)行的task集合
typedef std::pair<task_func, void*> task_func_pair;
std::vector<task_func_pair> _ready_tasks;
// ===========================================
};
添加了兩個屬性:
task_func_pair: 回調(diào)函數(shù)和參數(shù)的鍵值對.
_ready_tasks: 所有已經(jīng)就緒的待執(zhí)行的任務(wù)集合。
同時添加了兩個主要方法：
void add_task(task_func func, void *args): 添加一個任務(wù)到_ready_tasks中.
void execute_ready_tasks():執(zhí)行全部的_ready_tasks任務(wù)厌衙。
將這兩個方法實現(xiàn)如下：
lars_reactor/src/event_loop.cpp
//...
//添加一個任務(wù)task到ready_tasks集合中
void event_loop::add_task(task_func func, void *args)
{
task_func_pair func_pair(func, args);
_ready_tasks.push_back(func_pair);
}
//執(zhí)行全部的ready_tasks里面的任務(wù)
void event_loop::execute_ready_tasks()
{
std::vector<task_func_pair>::iterator it;
for (it = _ready_tasks.begin(); it != _ready_tasks.end(); it++) {
task_func func = it->first;//任務(wù)回調(diào)函數(shù)
void *args = it->second;//回調(diào)函數(shù)形參
//執(zhí)行任務(wù)
func(this, args);
}
//全部執(zhí)行完畢距淫，清空當(dāng)前的_ready_tasks
_ready_tasks.clear();
}
//...
? 那么execute_ready_tasks()函數(shù)需要在一個恰當(dāng)?shù)臅r候被執(zhí)行，我們這里就放在每次event_loop一次epoll_wait()處理完一組fd事件之后婶希，觸發(fā)一次額外的task任務(wù)榕暇。
lars_reactor/src/event_loop.cpp
//阻塞循環(huán)處理事件
void event_loop::event_process()
{
while (true) {
io_event_map_it ev_it;
int nfds = epoll_wait(_epfd, _fired_evs, MAXEVENTS, 10);
for (int i = 0; i < nfds; i++) {
//...
//...
}
//每次處理完一組epoll_wait觸發(fā)的事件之后，處理異步任務(wù)
this->execute_ready_tasks();
}
}
? 這里補(bǔ)充一下喻杈，因為在task的回調(diào)函數(shù)中彤枢，有形參event_loop *loop，可能會使用當(dāng)前l(fā)oop中監(jiān)控的fd信息筒饰，所以我們應(yīng)該給event_loop補(bǔ)充一個獲取當(dāng)前l(fā)oop監(jiān)控的全部fd信息的方法
class event_loop{
//...
//獲取全部監(jiān)聽事件的fd集合
void get_listen_fds(listen_fd_set &fds) {
fds = listen_fds;
}
//...
};
15.3 thread_pool模塊添加task任務(wù)機(jī)制
? 接下來我們就要用thread_pool來想每個thread所綁定的event_pool中去發(fā)送task任務(wù)缴啡，很明顯thread_pool應(yīng)該具備能夠?qū)ask加入到event_pool中的_ready_task集合的功能。
lars_reactor/include/thread_pool.h
#pragma once
#include <pthread.h>
#include "task_msg.h"
#include "thread_queue.h"
class thread_pool
{
public:
//構(gòu)造瓷们，初始化線程池, 開辟thread_cnt個
thread_pool(int thread_cnt);
//獲取一個thead
thread_queue<task_msg>* get_thread();
//發(fā)送一個task任務(wù)給thread_pool里的全部thread
void send_task(task_func func, void *args = NULL);
private:
//_queues是當(dāng)前thread_pool全部的消息任務(wù)隊列頭指針
thread_queue<task_msg> ** _queues;
//當(dāng)前線程池中的線程個數(shù)
int _thread_cnt;
//已經(jīng)啟動的全部therad編號
pthread_t * _tids;
//當(dāng)前選中的線程隊列下標(biāo)
int _index;
};
? send_task()方法就是發(fā)送給線程池中全部的thread去執(zhí)行task任務(wù).
lars_reactor/src/thread_pool.cpp
void thread_pool::send_task(task_func func, void *args)
{
task_msg task;
//給當(dāng)前thread_pool中的每個thread里的pool添加一個task任務(wù)
for (int i = 0; i < _thread_cnt; i++) {
//封裝一個task消息
task.type = task_msg::NEW_TASK;
task.task_cb = func;
task.args = args;
//取出第i個thread的消息隊列
thread_queue<task_msg> *queue = _queues[i];
//發(fā)送task消息
queue->send(task);
}
}
? send_task()的實現(xiàn)實際上是告知全部的thread业栅，封裝一個NEW_TASK類型的消息，通過task_queue告知對應(yīng)的thread.很明顯當(dāng)我們進(jìn)行 queue->send(task)的時候谬晕，當(dāng)前的thread綁定的loop,就會觸發(fā)deal_task_message()回調(diào)了碘裕。
lars_reactor/src/thread_pool.cpp
/*
* 一旦有task消息過來，這個業(yè)務(wù)是處理task消息業(yè)務(wù)的主流程
*
* 只要有人調(diào)用 thread_queue:: send()方法就會觸發(fā)次函數(shù)
*/
void deal_task_message(event_loop *loop, int fd, void *args)
{
//得到是哪個消息隊列觸發(fā)的
thread_queue<task_msg>* queue = (thread_queue<task_msg>*)args;
//將queue中的全部任務(wù)取出來
std::queue<task_msg> tasks;
queue->recv(tasks);
while (tasks.empty() != true) {
task_msg task = tasks.front();
//彈出一個元素
tasks.pop();
if (task.type == task_msg::NEW_CONN) {
//是一個新建鏈接的任務(wù)
//并且將這個tcp_conn加入當(dāng)當(dāng)前線程的loop中去監(jiān)聽
tcp_conn *conn = new tcp_conn(task.connfd, loop);
if (conn == NULL) {
fprintf(stderr, "in thread new tcp_conn error\n");
exit(1);
}
printf("[thread]: get new connection succ!\n");
}
else if (task.type == task_msg::NEW_TASK) {
//===========是一個新的普通任務(wù)===============
//當(dāng)前的loop就是一個thread的事件監(jiān)控loop,讓當(dāng)前l(fā)oop觸發(fā)task任務(wù)的回調(diào)
loop->add_task(task.task_cb, task.args);
//==========================================
}
else {
//其他未識別任務(wù)
fprintf(stderr, "unknow task!\n");
}
}
}
? 我們判斷task.type如果是NEW_TASK就將該task加入到當(dāng)前l(fā)oop中去.
通過上面的設(shè)計攒钳，可以看出來帮孔，thread_pool的send_task()應(yīng)該是一個對外的開發(fā)者接口，所以我們要讓服務(wù)器的tcp_server能夠獲取到thread_pool屬性.
lars_reactor/include/tcp_server.h
class tcp_server {
//...
//獲取當(dāng)前server的線程池
thread_pool *thread_poll() {
return _thread_pool;
}
//...
};
? ok不撑，這樣我們基本上完成的task異步處理業(yè)務(wù)的機(jī)制. 下面我們來測試一下這個功能.
15.4 完成Lars Reactor V0.11開發(fā)
server.cpp
#include "tcp_server.h"
#include <string>
#include <string.h>
#include "config_file.h"
tcp_server *server;
void print_lars_task(event_loop *loop, void *args)
{
printf("======= Active Task Func! ========\n");
listen_fd_set fds;
loop->get_listen_fds(fds);//不同線程的loop文兢，返回的fds是不同的
//可以向所有fds觸發(fā)
listen_fd_set::iterator it;
//遍歷fds
for (it = fds.begin(); it != fds.end(); it++) {
int fd = *it;
tcp_conn *conn = tcp_server::conns[fd]; //取出fd
if (conn != NULL) {
int msgid = 101;
const char *msg = "Hello I am a Task!";
conn->send_message(msg, strlen(msg), msgid);
}
}
}
//回顯業(yè)務(wù)的回調(diào)函數(shù)
void callback_busi(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
printf("callback_busi ...\n");
//直接回顯
conn->send_message(data, len, msgid);
}
//打印信息回調(diào)函數(shù)
void print_busi(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
printf("recv client: [%s]\n", data);
printf("msgid: [%d]\n", msgid);
printf("len: [%d]\n", len);
}
//新客戶端創(chuàng)建的回調(diào)
void on_client_build(net_connection *conn, void *args)
{
int msgid = 101;
const char *msg = "welcome! you online..";
conn->send_message(msg, strlen(msg), msgid);
//創(chuàng)建鏈接成功之后觸發(fā)任務(wù)
server->thread_poll()->send_task(print_lars_task);
}
//客戶端銷毀的回調(diào)
void on_client_lost(net_connection *conn, void *args)
{
printf("connection is lost !\n");
}
int main()
{
event_loop loop;
//加載配置文件
config_file::setPath("./serv.conf");
std::string ip = config_file::instance()->GetString("reactor", "ip", "0.0.0.0");
short port = config_file::instance()->GetNumber("reactor", "port", 8888);
printf("ip = %s, port = %d\n", ip.c_str(), port);
server = new tcp_server(&loop, ip.c_str(), port);
//注冊消息業(yè)務(wù)路由
server->add_msg_router(1, callback_busi);
server->add_msg_router(2, print_busi);
//注冊鏈接hook回調(diào)
server->set_conn_start(on_client_build);
server->set_conn_close(on_client_lost);
loop.event_process();
return 0;
}
? 我們在每次建立連接成功之后晤斩，觸發(fā)任務(wù)機(jī)制。其中print_lars_task()方法就是我們的異步任務(wù)姆坚。由于是全部thead都出發(fā)尸昧，所以該方法會被每個thread執(zhí)行。但是不同的thread中的pool所返回的fd是不一樣的旷偿，這里在print_lars_task()中，我們給對應(yīng)的客戶端做了一個簡單的消息發(fā)送爆侣。
?
client.cpp
#include "tcp_client.h"
#include <stdio.h>
#include <string.h>
//客戶端業(yè)務(wù)
void busi(const char *data, uint32_t len, int msgid, net_connection *conn, void *user_data)
{
//得到服務(wù)端回執(zhí)的數(shù)據(jù)
char *str = NULL;
str = (char*)malloc(len+1);
memset(str, 0, len+1);
memcpy(str, data, len);
printf("recv server: [%s]\n", str);
printf("msgid: [%d]\n", msgid);
printf("len: [%d]\n", len);
}
//客戶端銷毀的回調(diào)
void on_client_build(net_connection *conn, void *args)
{
int msgid = 1;
const char *msg = "Hello Lars!";
conn->send_message(msg, strlen(msg), msgid);
}
//客戶端銷毀的回調(diào)
void on_client_lost(net_connection *conn, void *args)
{
printf("on_client_lost...\n");
printf("Client is lost!\n");
}
int main()
{
event_loop loop;
//創(chuàng)建tcp客戶端
tcp_client client(&loop, "127.0.0.1", 7777, "clientv0.6");
//注冊消息路由業(yè)務(wù)
client.add_msg_router(1, busi);
client.add_msg_router(101, busi);
//設(shè)置hook函數(shù)
client.set_conn_start(on_client_build);
client.set_conn_close(on_client_lost);
//開啟事件監(jiān)聽
loop.event_process();
return 0;
}
? 客戶端代碼無差別萍程。
編譯并運行
服務(wù)端:
$ ./server
msg_router init...
ip = 127.0.0.1, port = 7777
create 0 thread
create 1 thread
create 2 thread
create 3 thread
create 4 thread
add msg cb msgid = 1
add msg cb msgid = 2
begin accept
begin accept
[thread]: get new connection succ!
callback_busi ...
======= Active Task Func! ========
======= Active Task Func! ========
======= Active Task Func! ========
======= Active Task Func! ========
======= Active Task Func! ========
客戶端：
$ ./client
msg_router init...
do_connect EINPROGRESS
add msg cb msgid = 1
add msg cb msgid = 101
connect 127.0.0.1:7777 succ!
recv server: [welcome! you online..]
msgid: [101]
len: [21]
recv server: [Hello Lars!]
msgid: [1]
len: [11]
recv server: [Hello I am a Task!]
msgid: [101]
len: [18]
? task機(jī)制已經(jīng)集成完畢,lars_reactor功能更加強(qiáng)大了。
關(guān)于作者：
作者：Aceld(劉丹冰)
mail: danbing.at@gmail.com
github: https://github.com/aceld
原創(chuàng)書籍gitbook: http://legacy.gitbook.com/@aceld
原創(chuàng)聲明:未經(jīng)作者允許請勿轉(zhuǎn)載, 如果轉(zhuǎn)載請注明出處
