server_common/f8
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
aozhiwei 2018-11-26 21:26:02 +08:00
parent ca5ccaf83c
commit 6f668cf5c5
4 changed files with 400 additions and 387 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

506
cpp/dbpool.cc
View File

@ -15,301 +15,304 @@
#include "framework/cpp/msgqueue.h" #include "framework/cpp/msgqueue.h"
#include "framework/cpp/utils.h" #include "framework/cpp/utils.h"
enum AsyncQueryError namespace f8
{ {
AQE_NO_ERROR = 0,
AQE_EXEC_TYPE_ERROR = 1,
AQE_QUERY_TYPE_ERROR = 2,
AQE_SYNTAX_ERROR = 3,
AQE_CONN_ERROR = 4
};
struct AsyncQueryRequest enum AsyncQueryError
{ {
long long context_id = 0; AQE_NO_ERROR = 0,
std::string sql; AQE_EXEC_TYPE_ERROR = 1,
AQE_QUERY_TYPE_ERROR = 2,
AQE_SYNTAX_ERROR = 3,
AQE_CONN_ERROR = 4
};
struct AsyncQueryRequest
{
long long context_id = 0;
std::string sql;
#if 1 #if 1
std::string _sql_fmt; std::string _sql_fmt;
std::vector<a8::XValue> _sql_params; std::vector<a8::XValue> _sql_params;
a8::XObject conn_info; a8::XObject conn_info;
#endif #endif
a8::XParams param; a8::XParams param;
time_t add_time = 0; time_t add_time = 0;
AsyncDBOnOkFunc on_ok = nullptr; AsyncDBOnOkFunc on_ok = nullptr;
AsyncDBOnErrorFunc on_error = nullptr; AsyncDBOnErrorFunc on_error = nullptr;
a8::TimerAttacher timer_attacher; a8::TimerAttacher timer_attacher;
}; };
struct AsyncQueryNode struct AsyncQueryNode
{ {
int socket_handle = 0; int socket_handle = 0;
int query_type = 0; int query_type = 0;
long long context_id = 0; long long context_id = 0;
std::string sql; std::string sql;
#if 1 #if 1
std::string _sql_fmt; std::string _sql_fmt;
std::vector<a8::XValue> _sql_params; std::vector<a8::XValue> _sql_params;
a8::XObject conn_info; a8::XObject conn_info;
#endif #endif
AsyncQueryNode* nextnode = nullptr; AsyncQueryNode* nextnode = nullptr;
}; };
class DBThread class DBThread
{
public:
void Init()
{ {
loop_mutex_ = new std::mutex(); public:
loop_cond_ = new std::condition_variable();
last_checkdb_tick_ = a8::XGetTickCount(); void Init()
{
loop_mutex_ = new std::mutex();
loop_cond_ = new std::condition_variable();
top_node_ = nullptr; last_checkdb_tick_ = a8::XGetTickCount();
bot_node_ = nullptr;
work_node_ = nullptr;
msg_mutex_ = new std::mutex();
work_thread_ = new std::thread(&DBThread::WorkThreadProc, this);
}
#if 0
void AddAsyncQuery(int sockhandle, int query_type, long long context_id, const std::string& sql)
{
AsyncQueryNode *p = new AsyncQueryNode();
p->query_type = query_type;
p->socket_handle = sockhandle;
p->context_id = context_id;
p->sql = sql;
std::unique_lock<std::mutex> lk(*loop_mutex_);
msg_mutex_->lock();
if (bot_node_) {
bot_node_->nextnode = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
msg_mutex_->unlock();
loop_cond_->notify_all();
}
#endif
void AddAsyncQuery(AsyncQueryNode* p)
{
std::unique_lock<std::mutex> lk(*loop_mutex_);
msg_mutex_->lock();
if (bot_node_) {
bot_node_->nextnode = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
msg_mutex_->unlock();
loop_cond_->notify_all();
}
private:
void WorkThreadProc()
{
while (true) {
CheckDB();
ProcessMsg();
WaitLoopCond();
}
}
void CheckDB()
{
if (a8::XGetTickCount() - last_checkdb_tick_ < 1000 * 60 * 5) {
return;
}
last_checkdb_tick_ = a8::XGetTickCount();
if (last_conn_ && last_query_) {
if (last_query_->ExecQuery("SELECT 1;", {}) <= 0) {
#if 0
a8::UdpLog::Instance()->Warning("mysql disconnect", {});
if (conn.Connect(dbhost_, 3306, dbuser_, dbpasswd_, gamedb_)) {
InitMysqlConnection(&query);
a8::UdpLog::Instance()->Info("mysql reconnect successed", {});
} else {
a8::UdpLog::Instance()->Info("mysql reconnect failed", {});
}
#endif
}
}
}
void ProcessMsg()
{
if (!work_node_ && top_node_) {
msg_mutex_->lock();
work_node_ = top_node_;
top_node_ = nullptr; top_node_ = nullptr;
bot_node_ = nullptr; bot_node_ = nullptr;
msg_mutex_->unlock(); work_node_ = nullptr;
msg_mutex_ = new std::mutex();
work_thread_ = new std::thread(&DBThread::WorkThreadProc, this);
} }
while (work_node_) {
AsyncQueryNode *pdelnode = work_node_;
work_node_ = work_node_->nextnode;
ProcAsyncQuery(pdelnode);
delete pdelnode;
}
}
void WaitLoopCond() #if 0
{ void AddAsyncQuery(int sockhandle, int query_type, long long context_id, const std::string& sql)
std::unique_lock<std::mutex> lk(*loop_mutex_);
{ {
AsyncQueryNode *p = new AsyncQueryNode();
p->query_type = query_type;
p->socket_handle = sockhandle;
p->context_id = context_id;
p->sql = sql;
std::unique_lock<std::mutex> lk(*loop_mutex_);
msg_mutex_->lock(); msg_mutex_->lock();
if (bot_node_) {
bot_node_->nextnode = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
msg_mutex_->unlock();
loop_cond_->notify_all();
}
#endif
void AddAsyncQuery(AsyncQueryNode* p)
{
std::unique_lock<std::mutex> lk(*loop_mutex_);
msg_mutex_->lock();
if (bot_node_) {
bot_node_->nextnode = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
msg_mutex_->unlock();
loop_cond_->notify_all();
}
private:
void WorkThreadProc()
{
while (true) {
CheckDB();
ProcessMsg();
WaitLoopCond();
}
}
void CheckDB()
{
if (a8::XGetTickCount() - last_checkdb_tick_ < 1000 * 60 * 5) {
return;
}
last_checkdb_tick_ = a8::XGetTickCount();
if (last_conn_ && last_query_) {
if (last_query_->ExecQuery("SELECT 1;", {}) <= 0) {
#if 0
a8::UdpLog::Instance()->Warning("mysql disconnect", {});
if (conn.Connect(dbhost_, 3306, dbuser_, dbpasswd_, gamedb_)) {
InitMysqlConnection(&query);
a8::UdpLog::Instance()->Info("mysql reconnect successed", {});
} else {
a8::UdpLog::Instance()->Info("mysql reconnect failed", {});
}
#endif
}
}
}
void ProcessMsg()
{
if (!work_node_ && top_node_) { if (!work_node_ && top_node_) {
msg_mutex_->lock();
work_node_ = top_node_; work_node_ = top_node_;
top_node_ = nullptr; top_node_ = nullptr;
bot_node_ = nullptr; bot_node_ = nullptr;
msg_mutex_->unlock();
}
while (work_node_) {
AsyncQueryNode *pdelnode = work_node_;
work_node_ = work_node_->nextnode;
ProcAsyncQuery(pdelnode);
delete pdelnode;
} }
msg_mutex_->unlock();
} }
if (!work_node_) {
loop_cond_->wait_for(lk, std::chrono::seconds(10));
}
}
bool ReCreateConn(a8::XObject& conn_info) void WaitLoopCond()
{ {
if (last_query_) { std::unique_lock<std::mutex> lk(*loop_mutex_);
delete last_query_; {
last_query_ = nullptr; msg_mutex_->lock();
if (!work_node_ && top_node_) {
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
}
msg_mutex_->unlock();
}
if (!work_node_) {
loop_cond_->wait_for(lk, std::chrono::seconds(10));
}
} }
if (last_conn_) {
delete last_conn_;
last_conn_ = nullptr;
}
last_conn_ = new a8::mysql::Connection();
last_query_ = last_conn_->CreateQuery();
if (last_conn_->Connect(
conn_info.Get("host"),
conn_info.Get("port"),
conn_info.Get("user"),
conn_info.Get("passwd"),
conn_info.Get("database")
)) {
f8::InitMysqlConnection(last_query_);
}
return true;
}
bool NeedReCreateConn(a8::XObject& conn_info) bool ReCreateConn(a8::XObject& conn_info)
{ {
if (!last_conn_) { if (last_query_) {
delete last_query_;
last_query_ = nullptr;
}
if (last_conn_) {
delete last_conn_;
last_conn_ = nullptr;
}
last_conn_ = new a8::mysql::Connection();
last_query_ = last_conn_->CreateQuery();
if (last_conn_->Connect(
conn_info.Get("host"),
conn_info.Get("port"),
conn_info.Get("user"),
conn_info.Get("passwd"),
conn_info.Get("database")
)) {
f8::InitMysqlConnection(last_query_);
}
return true; return true;
} }
if (last_conn_->GetHost() == conn_info.Get("host").GetString() &&
last_conn_->GetPort() == conn_info.Get("port").GetInt() &&
last_conn_->GetUser() == conn_info.Get("user").GetString() &&
last_conn_->GetPasswd() == conn_info.Get("passwd").GetString() &&
last_conn_->GetDataBase() == conn_info.Get("database").GetString()
) {
return false;
}
return true;
}
void ProcAsyncQuery(AsyncQueryNode* node) bool NeedReCreateConn(a8::XObject& conn_info)
{ {
if (NeedReCreateConn(node->conn_info)) { if (!last_conn_) {
ReCreateConn(node->conn_info); return true;
}
if (last_conn_->GetHost() == conn_info.Get("host").GetString() &&
last_conn_->GetPort() == conn_info.Get("port").GetInt() &&
last_conn_->GetUser() == conn_info.Get("user").GetString() &&
last_conn_->GetPasswd() == conn_info.Get("passwd").GetString() &&
last_conn_->GetDataBase() == conn_info.Get("database").GetString()
) {
return false;
}
return true;
} }
switch (node->query_type) {
case 0: void ProcAsyncQuery(AsyncQueryNode* node)
{ {
int ret = last_query_->ExecQuery(node->_sql_fmt.c_str(), node->_sql_params); if (NeedReCreateConn(node->conn_info)) {
if (ret < 0) { ReCreateConn(node->conn_info);
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid, }
a8::XParams() switch (node->query_type) {
.SetSender(node->context_id) case 0:
.SetParam1(AQE_SYNTAX_ERROR) {
.SetParam2(last_query_->GetError())); int ret = last_query_->ExecQuery(node->_sql_fmt.c_str(), node->_sql_params);
} else { if (ret < 0) {
DataSet* data_set = new DataSet(); f8::MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
data_set->reserve(last_query_->RowsNum()); a8::XParams()
while (!last_query_->Eof()) { .SetSender(node->context_id)
auto& row = a8::FastAppend(*data_set); .SetParam1(AQE_SYNTAX_ERROR)
int field_num = last_query_->FieldsNum(); .SetParam2(last_query_->GetError()));
row.reserve(field_num); } else {
for (int i = 0; i < field_num; i++) { DataSet* data_set = new DataSet();
row.push_back(last_query_->GetValue(i).GetString()); data_set->reserve(last_query_->RowsNum());
while (!last_query_->Eof()) {
auto& row = a8::FastAppend(*data_set);
int field_num = last_query_->FieldsNum();
row.reserve(field_num);
for (int i = 0; i < field_num; i++) {
row.push_back(last_query_->GetValue(i).GetString());
}
last_query_->Next();
} }
last_query_->Next(); f8::MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
a8::XParams()
.SetSender(node->context_id)
.SetParam1(AQE_NO_ERROR)
.SetParam2((void*)data_set));
} }
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
a8::XParams()
.SetSender(node->context_id)
.SetParam1(AQE_NO_ERROR)
.SetParam2((void*)data_set));
} }
} break;
break; case 1:
case 1: {
{ bool ret = last_query_->ExecScript(node->_sql_fmt.c_str(), node->_sql_params);
bool ret = last_query_->ExecScript(node->_sql_fmt.c_str(), node->_sql_params); if (!ret) {
if (!ret) { MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid, a8::XParams()
a8::XParams() .SetSender(node->context_id)
.SetSender(node->context_id) .SetParam1(AQE_SYNTAX_ERROR)
.SetParam1(AQE_SYNTAX_ERROR) .SetParam2(last_query_->GetError()));
.SetParam2(last_query_->GetError())); } else {
} else { DataSet* data_set = new DataSet();
DataSet* data_set = new DataSet(); MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid, a8::XParams()
a8::XParams() .SetSender(node->context_id)
.SetSender(node->context_id) .SetParam1(AQE_NO_ERROR)
.SetParam1(AQE_NO_ERROR) .SetParam2((void*)data_set));
.SetParam2((void*)data_set)); }
} }
break;
default:
{
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
a8::XParams()
.SetSender(node->context_id)
.SetParam1(AQE_QUERY_TYPE_ERROR)
.SetParam2("不可识别的query类型"));
}
break;
} }
break;
default:
{
MsgQueue::Instance()->PostMsg_r(exec_async_query_msgid,
a8::XParams()
.SetSender(node->context_id)
.SetParam1(AQE_QUERY_TYPE_ERROR)
.SetParam2("不可识别的query类型"));
}
break;
} }
}
public: public:
int exec_async_query_msgid = 0; int exec_async_query_msgid = 0;
private: private:
std::mutex *loop_mutex_ = nullptr; std::mutex *loop_mutex_ = nullptr;
std::condition_variable *loop_cond_ = nullptr; std::condition_variable *loop_cond_ = nullptr;
a8::mysql::Connection* last_conn_ = nullptr; a8::mysql::Connection* last_conn_ = nullptr;
a8::mysql::Query* last_query_ = nullptr; a8::mysql::Query* last_query_ = nullptr;
long long last_checkdb_tick_ = 0; long long last_checkdb_tick_ = 0;
std::thread *work_thread_ = nullptr; std::thread *work_thread_ = nullptr;
AsyncQueryNode *top_node_ = nullptr; AsyncQueryNode *top_node_ = nullptr;
AsyncQueryNode *bot_node_ = nullptr; AsyncQueryNode *bot_node_ = nullptr;
AsyncQueryNode *work_node_ = nullptr; AsyncQueryNode *work_node_ = nullptr;
std::mutex *msg_mutex_ = nullptr; std::mutex *msg_mutex_ = nullptr;
}; };
class DBPoolImpl class DBPoolImpl
{
public:
void Init()
{ {
public:
void Init()
{
#if 1 #if 1
/*mysql_init()不是完全线程安全的，但是只要成功调用一次就后就线程安全了， /*mysql_init()不是完全线程安全的，但是只要成功调用一次就后就线程安全了，
如果有多线程并发使用mysql_init()，建议在程序初始化时空调一次mysql_init()，他的这点特性很像qsort() 如果有多线程并发使用mysql_init()，建议在程序初始化时空调一次mysql_init()，他的这点特性很像qsort()
*/ */
a8::mysql::Connection conn; a8::mysql::Connection conn;
#endif #endif
curr_seqid = 1000001; curr_seqid = 1000001;
@ -464,3 +467,4 @@ void DBPool::ExecAsyncScript(a8::XObject conn_info, const char* querystr, std::v
{ {
impl_->InternalExecAsyncSql(1, conn_info, querystr, args, param, on_ok, on_error, hash_code); impl_->InternalExecAsyncSql(1, conn_info, querystr, args, param, on_ok, on_error, hash_code);
} }
}

47
cpp/dbpool.h
View File

@ -1,28 +1,31 @@
#pragma once #pragma once
typedef std::vector<std::vector<std::string>> DataSet; namespace f8
typedef void (*AsyncDBOnOkFunc)(a8::XParams& param, const DataSet* data_set);
typedef void (*AsyncDBOnErrorFunc)(a8::XParams& param, int error_code, const std::string& error_msg);
class DBPoolImpl;
class DBPool : public a8::Singleton<DBPool>
{ {
private: typedef std::vector<std::vector<std::string>> DataSet;
DBPool() {}; typedef void (*AsyncDBOnOkFunc)(a8::XParams& param, const DataSet* data_set);
friend class a8::Singleton<DBPool>; typedef void (*AsyncDBOnErrorFunc)(a8::XParams& param, int error_code, const std::string& error_msg);
public: class DBPoolImpl;
void Init(); class DBPool : public a8::Singleton<DBPool>
void UnInit(); {
void SetThreadNum(int thread_num); private:
DBPool() {};
friend class a8::Singleton<DBPool>;
//执行异步并行查询 public:
void ExecAsyncQuery(a8::XObject conn_info, const char* querystr, std::vector<a8::XValue> args, void Init();
a8::XParams param, AsyncDBOnOkFunc on_ok, AsyncDBOnErrorFunc on_error, long long hash_code); void UnInit();
//执行异步并行sql void SetThreadNum(int thread_num);
void ExecAsyncScript(a8::XObject conn_info, const char* querystr, std::vector<a8::XValue> args,
a8::XParams param, AsyncDBOnOkFunc on_ok, AsyncDBOnErrorFunc on_error, long long hash_code);
private: //执行异步并行查询
DBPoolImpl* impl_ = nullptr; void ExecAsyncQuery(a8::XObject conn_info, const char* querystr, std::vector<a8::XValue> args,
}; a8::XParams param, AsyncDBOnOkFunc on_ok, AsyncDBOnErrorFunc on_error, long long hash_code);
//执行异步并行sql
void ExecAsyncScript(a8::XObject conn_info, const char* querystr, std::vector<a8::XValue> args,
a8::XParams param, AsyncDBOnOkFunc on_ok, AsyncDBOnErrorFunc on_error, long long hash_code);
private:
DBPoolImpl* impl_ = nullptr;
};
}

185
cpp/msgqueue.cc
View File

@ -5,112 +5,115 @@
#include "framework/cpp/msgqueue.h" #include "framework/cpp/msgqueue.h"
#include "app.h" #include "app.h"
struct MsgQueueNode namespace f8
{ {
struct list_head entry; struct MsgQueueNode
MsgHandleFunc func;
};
class MsgQueueImp
{
public:
int curr_im_msgid = 10000;
std::map<int, list_head> msg_handlers;
void ProcessMsg(int msgid, const a8::XParams& param)
{ {
auto itr = msg_handlers.find(msgid); struct list_head entry;
if (itr != msg_handlers.end()) { MsgHandleFunc func;
list_head* head = &itr->second; };
struct MsgQueueNode *node = nullptr;
struct MsgQueueNode *tmp = nullptr; class MsgQueueImp
list_for_each_entry_safe(node, tmp, head, entry) { {
node->func(param); public:
int curr_im_msgid = 10000;
std::map<int, list_head> msg_handlers;
void ProcessMsg(int msgid, const a8::XParams& param)
{
auto itr = msg_handlers.find(msgid);
if (itr != msg_handlers.end()) {
list_head* head = &itr->second;
struct MsgQueueNode *node = nullptr;
struct MsgQueueNode *tmp = nullptr;
list_for_each_entry_safe(node, tmp, head, entry) {
node->func(param);
}
} }
} }
}
CallBackHandle RegisterCallBack(int msgid, MsgHandleFunc handle_func) CallBackHandle RegisterCallBack(int msgid, MsgHandleFunc handle_func)
{ {
MsgQueueNode* node = new MsgQueueNode(); MsgQueueNode* node = new MsgQueueNode();
INIT_LIST_HEAD(&node->entry); INIT_LIST_HEAD(&node->entry);
node->func = handle_func; node->func = handle_func;
auto itr = msg_handlers.find(msgid); auto itr = msg_handlers.find(msgid);
if (itr == msg_handlers.end()) { if (itr == msg_handlers.end()) {
msg_handlers[msgid] = list_head(); msg_handlers[msgid] = list_head();
itr = msg_handlers.find(msgid); itr = msg_handlers.find(msgid);
assert(itr != msg_handlers.end()); assert(itr != msg_handlers.end());
INIT_LIST_HEAD(&itr->second); INIT_LIST_HEAD(&itr->second);
}
list_add_tail(&node->entry, &itr->second);
return &node->entry;
} }
list_add_tail(&node->entry, &itr->second);
return &node->entry; };
void MsgQueue::Init()
{
imp_ = new MsgQueueImp();
} }
}; void MsgQueue::UnInit()
{
delete imp_;
imp_ = nullptr;
}
void MsgQueue::Init() void MsgQueue::SendMsg(int msgid, a8::XParams param)
{ {
imp_ = new MsgQueueImp(); imp_->ProcessMsg(msgid, param);
} }
void MsgQueue::UnInit() void MsgQueue::PostMsg(int msgid, a8::XParams param)
{ {
delete imp_; param._sys_field = msgid;
imp_ = nullptr; a8::Timer::Instance()->AddDeadLineTimer(0, param,
} [] (const a8::XParams& param)
{
MsgQueue::Instance()->imp_->ProcessMsg(param._sys_field, param);
});
}
void MsgQueue::SendMsg(int msgid, a8::XParams param) void MsgQueue::AddDelayMsg(int msgid, a8::XParams param, int milli_seconds)
{ {
imp_->ProcessMsg(msgid, param); param._sys_field = msgid;
} a8::Timer::Instance()->AddDeadLineTimer(milli_seconds, param,
[] (const a8::XParams& param)
{
MsgQueue::Instance()->imp_->ProcessMsg(param._sys_field, param);
});
}
void MsgQueue::PostMsg(int msgid, a8::XParams param) void MsgQueue::RemoveCallBack(CallBackHandle handle)
{ {
param._sys_field = msgid; list_head* head = handle;
a8::Timer::Instance()->AddDeadLineTimer(0, param, MsgQueueNode* node = list_entry(head, struct MsgQueueNode, entry);
[] (const a8::XParams& param) list_del_init(&node->entry);
{ delete node;
MsgQueue::Instance()->imp_->ProcessMsg(param._sys_field, param); }
});
}
void MsgQueue::AddDelayMsg(int msgid, a8::XParams param, int milli_seconds) CallBackHandle MsgQueue::RegisterCallBack(int msgid, MsgHandleFunc handle_func)
{ {
param._sys_field = msgid; return imp_->RegisterCallBack(msgid, handle_func);
a8::Timer::Instance()->AddDeadLineTimer(milli_seconds, param, }
[] (const a8::XParams& param)
{
MsgQueue::Instance()->imp_->ProcessMsg(param._sys_field, param);
});
}
void MsgQueue::RemoveCallBack(CallBackHandle handle) int MsgQueue::AllocIMMsgId()
{ {
list_head* head = handle; return ++imp_->curr_im_msgid;
MsgQueueNode* node = list_entry(head, struct MsgQueueNode, entry); }
list_del_init(&node->entry);
delete node;
}
CallBackHandle MsgQueue::RegisterCallBack(int msgid, MsgHandleFunc handle_func) void MsgQueue::ProcessMsg(int msgid, const a8::XParams& param)
{ {
return imp_->RegisterCallBack(msgid, handle_func); imp_->ProcessMsg(msgid, param);
} }
int MsgQueue::AllocIMMsgId() void MsgQueue::PostMsg_r(int msgid, a8::XParams param)
{ {
return ++imp_->curr_im_msgid; a8::XParams* p = new a8::XParams();
} param.DeepCopy(*p);
App::Instance()->AddIMMsg(f8::IM_SysMsgQueue, a8::XParams().SetSender(msgid).SetParam1((void*)p));
void MsgQueue::ProcessMsg(int msgid, const a8::XParams& param) }
{
imp_->ProcessMsg(msgid, param);
}
void MsgQueue::PostMsg_r(int msgid, a8::XParams param)
{
a8::XParams* p = new a8::XParams();
param.DeepCopy(*p);
App::Instance()->AddIMMsg(f8::IM_SysMsgQueue, a8::XParams().SetSender(msgid).SetParam1((void*)p));
} }

49
cpp/msgqueue.h
View File

@ -1,30 +1,33 @@
#pragma once #pragma once
typedef std::function<void (const a8::XParams& param)> MsgHandleFunc; namespace f8
typedef list_head* CallBackHandle;
class MsgQueueImp;
class MsgQueue : public a8::Singleton<MsgQueue>
{ {
private: typedef std::function<void (const a8::XParams& param)> MsgHandleFunc;
MsgQueue() {}; typedef list_head* CallBackHandle;
friend class a8::Singleton<MsgQueue>;
public: class MsgQueueImp;
void Init(); class MsgQueue : public a8::Singleton<MsgQueue>
void UnInit(); {
private:
MsgQueue() {};
friend class a8::Singleton<MsgQueue>;
void SendMsg(int msgid, a8::XParams param); public:
void PostMsg(int msgid, a8::XParams param); void Init();
void AddDelayMsg(int msgid, a8::XParams param, int milli_seconds); void UnInit();
void RemoveCallBack(CallBackHandle handle);
CallBackHandle RegisterCallBack(int msgid, MsgHandleFunc handle_func);
int AllocIMMsgId();
void ProcessMsg(int msgid, const a8::XParams& param);
//线程安全版本 void SendMsg(int msgid, a8::XParams param);
void PostMsg_r(int msgid, a8::XParams param); void PostMsg(int msgid, a8::XParams param);
void AddDelayMsg(int msgid, a8::XParams param, int milli_seconds);
void RemoveCallBack(CallBackHandle handle);
CallBackHandle RegisterCallBack(int msgid, MsgHandleFunc handle_func);
int AllocIMMsgId();
void ProcessMsg(int msgid, const a8::XParams& param);
private: //线程安全版本
MsgQueueImp* imp_ = nullptr; void PostMsg_r(int msgid, a8::XParams param);
};
private:
MsgQueueImp* imp_ = nullptr;
};
}