libjingle源码分析之：Thread和SocketServer
2013-07-13 20:12
411 查看
摘要
本文主要分析了libjingle源码中的Thread和SocketServer模块，以及它们是如何协同工作的。首先，介绍了Thread和SocketServer的模型，给出了如何使用Thread的示例。然后，分析了Thread中的默认消息循环的处理流程和如何自己处理消息。
概述
libjingle源码中，Thread和SocketServer模块的原理如下图所示。整个模型实际上是一个消息模型，Thread主要负责处理消息，MessageQue表示的是当前的消息队列，MessageHandler由用户用来定义处理消息的动作。而ThreadManager为单实例，可以获取当前的Thread，这样用户可以往当前的Thread中投递消息。SocketServer代表的是用来侦听Socket的服务，它是一个独立的模块。
消息的处理流程主要由Thread负责。上图中有两条处理流程，分别用两根带箭头的线表示。左边箭头的处理流程为：当消息队列中没有消息时，Thread将控制权转交给SocketServer，直到有消息时会通知SocketServer返回到Thread。也就是说Thread优先的是处理消息，在空闲时，会让SocketServer侦听socket。右边的箭头是正常的消息处理流程，获取消息并处理用户定义的对应的OnMessage函数。
SocketServer模块只是用在libjingle内部，用户并不需要直接使用它。P2P中使用了PhysicalSocketServer作为SocketServer，它的原理如下图所示：
PhysicalSocketServer主要是侦听基于本地网卡的socket（libjingle中还有一些伪socket），然后分发socket事件到Dispatcher中。Dispatcher是PhysicalSocketServer的分发体，功能有点类似于MessageHandler。Dispatcher中定义了感兴趣的socket事件和对应的处理。
类的关系
本文提到的一些类的关系如下图所示。Thread类继承自MessageQue，可以通过Thread类来操作队列消息。PhysicalSocketServer除了实现SocketServer接口之外，还可以添加和删除Dispatcher。熟悉了这些类，基本上就了解Thread模块和SocketServer模块的工作原理。
使用
Thread的使用示例参见下面的代码。获取当前线程是通过Thread的Current函数，它会转调ThreadManager对象的CurrentThread函数。由于Thread继承自MessageQue，可以直接通过Thread对象来投递消息，Post函数的第一个参数是OnMessage所处的对象，会被保存于Message对象中。处理消息只要重载MessageHandler的OnMessage函数即可。main函数则调用Thread的Run函数进入默认的消息处理循环，默认的消息处理循环在本示例中就是：循环取消息，调用MessageHandler的OnMessage函数。
[cpp] view
plaincopy
#include <string>
#include <iostream>
#include "talk/base/thread.h"
class HelpData : public talk_base::MessageData
{
public:
std::string info_;
};
class Police : public talk_base::MessageHandler
{
public:
enum {
MSG_HELP,
};
void Help(const std::string& info) {
HelpData* data = new HelpData;
data->info_ = info;
talk_base::Thread::Current()->Post(this, MSG_HELP, data);
}
virtual void OnMessage(talk_base::Message* msg) {
switch (msg->message_id) {
case MSG_HELP:
HelpData* data = (HelpData*)msg->pdata;
std::cout << "MSG_HELP : " << data->info_ << std::endl;
break;
}
}
};
int main(int argc, char** argv)
{
Police p;
p.Help("Please help me!");
talk_base::Thread::Current()->Run();
return 0;
}
处理消息
Thread的默认消息处理流程可用下图表示。默认消息处理函数的入口为Thread::Run()，另一个内嵌循环是SocketServer::Wait()。箭头指向数据成员则表示，处理相关数据。
要执行默认消息处理循环，使用下列语句即可：
[cpp] view
plaincopy
talk_base::Thread::Current()->Run();
当然，你也可以自己处理消息，可以参见pcp例子中的代码，这段代码用在登录阶段，等待登录操作完成。
[cpp] view
plaincopy
// Wait until login succeeds.
std::vector<uint32> ids;
ids.push_back(MSG_LOGIN_COMPLETE);
ids.push_back(MSG_LOGIN_FAILED);
if (MSG_LOGIN_FAILED == Loop(ids))
FatalError("Failed to connect");
首先设置关系的消息id集合，然后进入自定义的消息循环。
[cpp] view
plaincopy
// Runs the current thread until a message with the given ID is seen.
uint32 Loop(const std::vector<uint32>& ids) {
talk_base::Message msg;
while (talk_base::Thread::Current()->Get(&msg)) {
if (msg.phandler == NULL) {
if (std::find(ids.begin(), ids.end(), msg.message_id) != ids.end())
return msg.message_id;
std::cout << "orphaned message: " << msg.message_id;
continue;
}
talk_base::Thread::Current()->Dispatch(&msg);
}
return 0;
}
每次循环中获取消息，然后判断消息的id是否后符合要求。符合要求就返回（表示登录或登录失败），否则就派发消息（就是执行OnMessage函数，和Thread的默认消息处理一样）。
一. Thread类是libjingle中比较核心的类。 我把他的功能主要分为三块：
1. Thread相关函数: 主要是对不同操作系统Thread的统一接口包装。 其中包括了Start(), Stop(), Join()等线程控制函数， 也包括了优先级控制相关的函数， 还有一个定位查找的功能。关于定位查找的功能主要是由ThreadManager控制的， ThreadManager类主要是对Thread进行管理。当然会有一个global的ThreadManager的对象来进行所有Thread对象的注册，定位，查找等操作。 Thread中的执行函数是PreRun()。 在PreRun()函数中就是用ThreadManager：：SetCurrent()函数来设置当前的Thread。
2. 消息机制: 这个功能主要是由继承MessageQueue中的来获取的。
Thread中的核心函数就是ProcessMessages():
bool Thread::ProcessMessages(int cmsLoop) {
uint32 msEnd;
if (cmsLoop != kForever)
msEnd = GetMillisecondCount() + cmsLoop;
int cmsNext = cmsLoop;
while (true) {
Message msg;
if (!Get(&msg, cmsNext))
return false;
Dispatch(&msg);
if (cmsLoop != kForever) {
uint32 msCur = GetMillisecondCount();
if (msCur >= msEnd)
return true;
cmsNext = msEnd - msCur;
}
}
}
此函数主要是循环地得到消息，分派消息。当然这边有超时的机制。这边主要用到了MessageQueue：：Get()和Dispatch()函数，
这个等下会着重介绍。
3. 进行异步网络事件监听的功能。主要是继承了SocketServer而获得的功能。 (这里要注意是因为MessageQueue继承SocketServer的原因，PS: 这边我个人觉得应该分开这两块功能接口更舒服。) 这个等下也会着重介绍。
二. MessageQueue: (这段懂得消息机制的朋友可以略过...)
MessageQueue主要是实现了一个消息队列来Post, Get, Peek消息。 主要用处么，将处理逻辑和内容解耦，在编程中经常用的手法。
1. 消息格式:
struct Message {
Message() {
memset(this, 0, sizeof(*this)); // 初始化
}
MessageHandler *phandler; // Message的回调函数，用来处理得到消息的操作。
uint32 message_id; // message id
MessageData *pdata; // message data是个标志接口
uint32 ts_sensitive; // ？
};
2. 消息队列类型:
这里主要有两种消息队列类型，一种是一般的先进先出的消息队列，还有一种是可以设置delayed时间根据delayed时间排列的优先级队列。
2. 发消息:
我们使用Post函数来发消息。
void MessageQueue::Post(MessageHandler *phandler, uint32 id,
MessageData *pdata, bool time_sensitive) {
if (fStop_)
return;
// Keep thread safe
// Add the message to the end of the queue
// Signal for the multiplexer to return
CritScope cs(&crit_);
EnsureActive();
Message msg;
msg.phandler = phandler;
msg.message_id = id;
msg.pdata = pdata;
if (time_sensitive) {
msg.ts_sensitive = Time() + kMaxMsgLatency;
}
msgq_.push(msg);
ss_->WakeUp();
}
这里主要就组成了一个Message，然后放到消息队列尾部了，此操作是线程安全的。 此外，还调用了socketserver多路复用器的WakeUp()方法来唤醒正在一定时间内循坏等待的多路复用器。
同样PostDelayed就是发消息到一个优先级队列中去。
3. 取消息:
我们使用Get（）来获得消息。
bool MessageQueue::Get(Message *pmsg, int cmsWait) {
// Return and clear peek if present
// Always return the peek if it exists so there is Peek/Get symmetry
if (fPeekKeep_) {
*pmsg = msgPeek_;
fPeekKeep_ = false;
return true;
}
// Get w/wait + timer scan / dispatch + socket / event multiplexer dispatch
int cmsTotal = cmsWait;
int cmsElapsed = 0;
uint32 msStart = Time();
uint32 msCurrent = msStart;
while (true) {
// Check for sent messages
ReceiveSends();
// Check queues
int cmsDelayNext = kForever;
{
CritScope cs(&crit_);
// Check for delayed messages that have been triggered
// Calc the next trigger too
while (!dmsgq_.empty()) {
if (msCurrent < dmsgq_.top().msTrigger_) {
cmsDelayNext = dmsgq_.top().msTrigger_ - msCurrent;
break;
}
msgq_.push(dmsgq_.top().msg_);
dmsgq_.pop();
}
// Check for posted events
while (!msgq_.empty()) {
*pmsg = msgq_.front();
if (pmsg->ts_sensitive) {
long delay = TimeDiff(msCurrent, pmsg->ts_sensitive);
if (delay > 0) {
LOG_F(LS_WARNING) << "id: " << pmsg->message_id << " delay: "
<< (delay + kMaxMsgLatency) << "ms";
}
}
msgq_.pop();
if (MQID_DISPOSE == pmsg->message_id) {
ASSERT(NULL == pmsg->phandler);
delete pmsg->pdata;
continue;
}
return true;
}
}
if (fStop_)
break;
// Which is shorter, the delay wait or the asked wait?
int cmsNext;
if (cmsWait == kForever) {
cmsNext = cmsDelayNext;
} else {
cmsNext = cmsTotal - cmsElapsed;
if (cmsNext < 0)
cmsNext = 0;
if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext))
cmsNext = cmsDelayNext;
}
// Wait and multiplex in the meantime
ss_->Wait(cmsNext, true);
// If the specified timeout expired, return
msCurrent = Time();
cmsElapsed = msCurrent - msStart;
if (cmsWait != kForever) {
if (cmsElapsed >= cmsWait)
return false;
}
}
return false;
}
Get方法如同注释所说主要做了以下几件事情。
1. 如果已经有消息被Peek，那么取出那条消息直接返回ok了，并且把PeekKeep的标志置为false。
2. 先去delayed队列取消息，看看有没有到时的消息。 如果有，就放到主消息队列中去。
3. 去主消息队列中取消息，如果有，就直接返回ok了，没有的话继续以下步骤。这边又有一个删除消息的消息，如果收到的是刚才那样一个消息的话，就需要执行dispose操作。
4. 查看stop标志，如果不结束的话就利用余下的时间就进行多路复用器的监听工作，来监听异步网络事件。看到这条的时候就明白为什么要在Post消息的时候WakeUp多路复用器了，快速响应消息是王道啊！
Peek消息:
Peek消息其实是查看消息，再去取的时候还能够得到该条消息。我一开始以为message不会从队列中Pop出来，不过这边他采取类似的做法。设一个消息变量来保存Peeked的msg，一个标志peekKeep_来表示消息是否被Peek出来保存到变量中。因此，当peekKeep_为true的时候，Get一下会取出变量里面的那条消息并且将标志置为false。
其他一些函数实现也比较基础，简单，这边就不说消息队列这个无聊的东西了。
三 SocketServer
接下去我们看看我最感兴趣的SocketServer了。
libjingle是gtalk的客户端，和服务器的底层socket通信主要就是使用了这个接口。
1. Socket:
SocketServer首先是个SocketFactory，所以它能Create出socket来，可以创建阻塞的和非阻塞的socket。
而创建出来的Socket则是对于不同os底层socket的高层抽象。
主要有Bind(), Connect(), Send(), Recv()等等。
而AsyncSocket则是有一些多了一些信号事件来给外界捕捉处理。
class AsyncSocket : public Socket, public sigslot::has_slots<> {
public:
virtual ~AsyncSocket() {}
sigslot::signal1<AsyncSocket*> SignalReadEvent; // ready to read
sigslot::signal1<AsyncSocket*> SignalWriteEvent; // ready to write
sigslot::signal1<AsyncSocket*> SignalConnectEvent; // connected
sigslot::signal2<AsyncSocket*,int> SignalCloseEvent; // closed
// TODO: error
};
这边用到了signal/slot机制，这个机制很棒，以前如果用过QT的朋友肯定能懂，和QT的signal/slot机制基本类似，libjingle里面的signal/slot库我还没来得及分析，不过短小精悍，应该不错，以后有机会再分析下，呵呵。 不过最重要的就是知道这个是一个类似与Observer模式的东西，最重要的就是其你发一个一开始有一个signal和一些slots用connect函数来绑定，signal就是被观察者，slot就是观察者。 如果有当signal触发的时候，slot就能接受到。
SocketServer是对监听网络事件的高层抽象，主要有两个函数:Wait()和Wakeup()。 监听的时候是Wait（），可以用WakeUp()来唤醒。
这里我们主要用到了异步的客户端。使用了select 多路复用器来监听异步网络事件。
当然抽象的好处就是可以有不同的实现，甚至是虚拟mock的实现。
而我们一般默认用到的就是PhysicalSocketServer，这个类封装了不同os之间网络库的差异性。
对于常用的一些socket函数的封装在这里我就不介绍了，主要介绍一下Wait()和Wakeup()函数中的工作机制。这边以linux为例子。
1. Dispatcher:
class Dispatcher {
public:
virtual uint32 GetRequestedEvents() = 0; // 返回感兴趣的事件
virtual void OnPreEvent(uint32 ff) = 0; // 在事件执行前
virtual void OnEvent(uint32 ff, int err) = 0; // 在事件执行时
virtual int GetDescriptor() = 0; // 返回Dispather的描述
};
这边引入了一个Dispatcher的概念，Dispatcher是用来派发事件的，因此对于多路复用器的主循环来说我们可以引入一个做法: 即当我们收到事件后我们使用不同的Dispatcher来进行派发，这里的事件可以是网络事件，也可以是其他可以被多路复用器选择出来的事件。
我们先来看网络事件派发器SocketDispatcher。
它主要是继承了PhysicalSocket和Dispatcher，然后在派发执行函数OnEvent()里面进行异步网络事件通知处理，signal感兴趣的 网络连接，读，写，关闭事件，并且将更新感兴趣的网络事件，去除已经做过的事件。
virtual void OnEvent(uint32 ff, int err) {
int cache_id = id_;
if ((ff & kfRead) != 0) {
enabled_events_ &= ~kfRead;
SignalReadEvent(this);
}
if (((ff & kfWrite) != 0) && (id_ == cache_id)) {
enabled_events_ &= ~kfWrite;
SignalWriteEvent(this);
}
if (((ff & kfConnect) != 0) && (id_ == cache_id)) {
if (ff != kfConnect)
LOG(LS_VERBOSE) << "Signalled with kfConnect: " << ff;
enabled_events_ &= ~kfConnect;
SignalConnectEvent(this);
}
if (((ff & kfClose) != 0) && (id_ == cache_id)) {
//LOG(INFO) << "SOCK[" << static_cast<int>(s_) << "] OnClose() Error: " << err;
signal_close_ = true;
signal_err_ = err;
}
}
还有一种就是其他的事件分派器， EventDispatcher在linux下面是用了pipe来模拟(windows则有event机制)。这样也可以被多路复用器select出来。
Signal的时候从一端往管道中写，而在PreOnEvent的时候从另一端读出来。
这边一个实例就是类Signaler。 该类继承了EventDispatcher来完成对退出wait，达到wakeup。
2. wait()函数详解:
这里我贴了一段wait()的linux实现代码，不包括里面的超时机制。
// Zero all fd_sets. Don't need to do this inside the loop since
// select() zeros the descriptors not signaled
fd_set fdsRead;
FD_ZERO(&fdsRead);
fd_set fdsWrite;
FD_ZERO(&fdsWrite);
fWait_ = true;
while (fWait_) {
int fdmax = -1;
{
CritScope cr(&crit_);
for (unsigned i = 0; i < dispatchers_.size(); i++) {
// Query dispatchers for read and write wait state
Dispatcher *pdispatcher = dispatchers_[i];
assert(pdispatcher);
if (!process_io && (pdispatcher != signal_wakeup_))
continue;
int fd = pdispatcher->GetDescriptor();
if (fd > fdmax)
fdmax = fd;
uint32 ff = pdispatcher->GetRequestedEvents();
if (ff & kfRead)
FD_SET(fd, &fdsRead);
if (ff & (kfWrite | kfConnect))
FD_SET(fd, &fdsWrite);
}
}
// Wait then call handlers as appropriate
// < 0 means error
// 0 means timeout
// > 0 means count of descriptors ready
int n = select(fdmax + 1, &fdsRead, &fdsWrite, NULL, ptvWait);
// If error, return error
// todo: do something intelligent
if (n < 0)
return false;
// If timeout, return success
if (n == 0)
return true;
// We have signaled descriptors
{
CritScope cr(&crit_);
for (unsigned i = 0; i < dispatchers_.size(); i++) {
Dispatcher *pdispatcher = dispatchers_[i];
int fd = pdispatcher->GetDescriptor();
uint32 ff = 0;
if (FD_ISSET(fd, &fdsRead)) {
FD_CLR(fd, &fdsRead);
ff |= kfRead;
}
if (FD_ISSET(fd, &fdsWrite)) {
FD_CLR(fd, &fdsWrite);
if (pdispatcher->GetRequestedEvents() & kfConnect) {
ff |= kfConnect;
} else {
ff |= kfWrite;
}
}
if (ff != 0) {
pdispatcher->OnPreEvent(ff);
pdispatcher->OnEvent(ff, 0);
}
}
}
1. 创建 fd_set并且初始化。
2. 对每个dispatcher的fd添加感兴趣的事件。
3. select()。
4. 更新每个dispatcher的感兴趣的并且发生的事件到ff，然后执行OnPreEvent和OnEvent。
本文主要分析了libjingle源码中的Thread和SocketServer模块，以及它们是如何协同工作的。首先，介绍了Thread和SocketServer的模型，给出了如何使用Thread的示例。然后，分析了Thread中的默认消息循环的处理流程和如何自己处理消息。
概述
libjingle源码中，Thread和SocketServer模块的原理如下图所示。整个模型实际上是一个消息模型，Thread主要负责处理消息，MessageQue表示的是当前的消息队列，MessageHandler由用户用来定义处理消息的动作。而ThreadManager为单实例，可以获取当前的Thread，这样用户可以往当前的Thread中投递消息。SocketServer代表的是用来侦听Socket的服务，它是一个独立的模块。
消息的处理流程主要由Thread负责。上图中有两条处理流程，分别用两根带箭头的线表示。左边箭头的处理流程为：当消息队列中没有消息时，Thread将控制权转交给SocketServer，直到有消息时会通知SocketServer返回到Thread。也就是说Thread优先的是处理消息，在空闲时，会让SocketServer侦听socket。右边的箭头是正常的消息处理流程，获取消息并处理用户定义的对应的OnMessage函数。
SocketServer模块只是用在libjingle内部，用户并不需要直接使用它。P2P中使用了PhysicalSocketServer作为SocketServer，它的原理如下图所示：
PhysicalSocketServer主要是侦听基于本地网卡的socket（libjingle中还有一些伪socket），然后分发socket事件到Dispatcher中。Dispatcher是PhysicalSocketServer的分发体，功能有点类似于MessageHandler。Dispatcher中定义了感兴趣的socket事件和对应的处理。
类的关系
本文提到的一些类的关系如下图所示。Thread类继承自MessageQue，可以通过Thread类来操作队列消息。PhysicalSocketServer除了实现SocketServer接口之外，还可以添加和删除Dispatcher。熟悉了这些类，基本上就了解Thread模块和SocketServer模块的工作原理。
使用
Thread的使用示例参见下面的代码。获取当前线程是通过Thread的Current函数，它会转调ThreadManager对象的CurrentThread函数。由于Thread继承自MessageQue，可以直接通过Thread对象来投递消息，Post函数的第一个参数是OnMessage所处的对象，会被保存于Message对象中。处理消息只要重载MessageHandler的OnMessage函数即可。main函数则调用Thread的Run函数进入默认的消息处理循环，默认的消息处理循环在本示例中就是：循环取消息，调用MessageHandler的OnMessage函数。
[cpp] view
plaincopy
#include <string>
#include <iostream>
#include "talk/base/thread.h"
class HelpData : public talk_base::MessageData
{
public:
std::string info_;
};
class Police : public talk_base::MessageHandler
{
public:
enum {
MSG_HELP,
};
void Help(const std::string& info) {
HelpData* data = new HelpData;
data->info_ = info;
talk_base::Thread::Current()->Post(this, MSG_HELP, data);
}
virtual void OnMessage(talk_base::Message* msg) {
switch (msg->message_id) {
case MSG_HELP:
HelpData* data = (HelpData*)msg->pdata;
std::cout << "MSG_HELP : " << data->info_ << std::endl;
break;
}
}
};
int main(int argc, char** argv)
{
Police p;
p.Help("Please help me!");
talk_base::Thread::Current()->Run();
return 0;
}
处理消息
Thread的默认消息处理流程可用下图表示。默认消息处理函数的入口为Thread::Run()，另一个内嵌循环是SocketServer::Wait()。箭头指向数据成员则表示，处理相关数据。
要执行默认消息处理循环，使用下列语句即可：
[cpp] view
plaincopy
talk_base::Thread::Current()->Run();
当然，你也可以自己处理消息，可以参见pcp例子中的代码，这段代码用在登录阶段，等待登录操作完成。
[cpp] view
plaincopy
// Wait until login succeeds.
std::vector<uint32> ids;
ids.push_back(MSG_LOGIN_COMPLETE);
ids.push_back(MSG_LOGIN_FAILED);
if (MSG_LOGIN_FAILED == Loop(ids))
FatalError("Failed to connect");
首先设置关系的消息id集合，然后进入自定义的消息循环。
[cpp] view
plaincopy
// Runs the current thread until a message with the given ID is seen.
uint32 Loop(const std::vector<uint32>& ids) {
talk_base::Message msg;
while (talk_base::Thread::Current()->Get(&msg)) {
if (msg.phandler == NULL) {
if (std::find(ids.begin(), ids.end(), msg.message_id) != ids.end())
return msg.message_id;
std::cout << "orphaned message: " << msg.message_id;
continue;
}
talk_base::Thread::Current()->Dispatch(&msg);
}
return 0;
}
每次循环中获取消息，然后判断消息的id是否后符合要求。符合要求就返回（表示登录或登录失败），否则就派发消息（就是执行OnMessage函数，和Thread的默认消息处理一样）。
一. Thread类是libjingle中比较核心的类。 我把他的功能主要分为三块：
1. Thread相关函数: 主要是对不同操作系统Thread的统一接口包装。 其中包括了Start(), Stop(), Join()等线程控制函数， 也包括了优先级控制相关的函数， 还有一个定位查找的功能。关于定位查找的功能主要是由ThreadManager控制的， ThreadManager类主要是对Thread进行管理。当然会有一个global的ThreadManager的对象来进行所有Thread对象的注册，定位，查找等操作。 Thread中的执行函数是PreRun()。 在PreRun()函数中就是用ThreadManager：：SetCurrent()函数来设置当前的Thread。
2. 消息机制: 这个功能主要是由继承MessageQueue中的来获取的。
Thread中的核心函数就是ProcessMessages():
bool Thread::ProcessMessages(int cmsLoop) {
uint32 msEnd;
if (cmsLoop != kForever)
msEnd = GetMillisecondCount() + cmsLoop;
int cmsNext = cmsLoop;
while (true) {
Message msg;
if (!Get(&msg, cmsNext))
return false;
Dispatch(&msg);
if (cmsLoop != kForever) {
uint32 msCur = GetMillisecondCount();
if (msCur >= msEnd)
return true;
cmsNext = msEnd - msCur;
}
}
}
此函数主要是循环地得到消息，分派消息。当然这边有超时的机制。这边主要用到了MessageQueue：：Get()和Dispatch()函数，
这个等下会着重介绍。
3. 进行异步网络事件监听的功能。主要是继承了SocketServer而获得的功能。 (这里要注意是因为MessageQueue继承SocketServer的原因，PS: 这边我个人觉得应该分开这两块功能接口更舒服。) 这个等下也会着重介绍。
二. MessageQueue: (这段懂得消息机制的朋友可以略过...)
MessageQueue主要是实现了一个消息队列来Post, Get, Peek消息。 主要用处么，将处理逻辑和内容解耦，在编程中经常用的手法。
1. 消息格式:
struct Message {
Message() {
memset(this, 0, sizeof(*this)); // 初始化
}
MessageHandler *phandler; // Message的回调函数，用来处理得到消息的操作。
uint32 message_id; // message id
MessageData *pdata; // message data是个标志接口
uint32 ts_sensitive; // ？
};
2. 消息队列类型:
这里主要有两种消息队列类型，一种是一般的先进先出的消息队列，还有一种是可以设置delayed时间根据delayed时间排列的优先级队列。
2. 发消息:
我们使用Post函数来发消息。
void MessageQueue::Post(MessageHandler *phandler, uint32 id,
MessageData *pdata, bool time_sensitive) {
if (fStop_)
return;
// Keep thread safe
// Add the message to the end of the queue
// Signal for the multiplexer to return
CritScope cs(&crit_);
EnsureActive();
Message msg;
msg.phandler = phandler;
msg.message_id = id;
msg.pdata = pdata;
if (time_sensitive) {
msg.ts_sensitive = Time() + kMaxMsgLatency;
}
msgq_.push(msg);
ss_->WakeUp();
}
这里主要就组成了一个Message，然后放到消息队列尾部了，此操作是线程安全的。 此外，还调用了socketserver多路复用器的WakeUp()方法来唤醒正在一定时间内循坏等待的多路复用器。
同样PostDelayed就是发消息到一个优先级队列中去。
3. 取消息:
我们使用Get（）来获得消息。
bool MessageQueue::Get(Message *pmsg, int cmsWait) {
// Return and clear peek if present
// Always return the peek if it exists so there is Peek/Get symmetry
if (fPeekKeep_) {
*pmsg = msgPeek_;
fPeekKeep_ = false;
return true;
}
// Get w/wait + timer scan / dispatch + socket / event multiplexer dispatch
int cmsTotal = cmsWait;
int cmsElapsed = 0;
uint32 msStart = Time();
uint32 msCurrent = msStart;
while (true) {
// Check for sent messages
ReceiveSends();
// Check queues
int cmsDelayNext = kForever;
{
CritScope cs(&crit_);
// Check for delayed messages that have been triggered
// Calc the next trigger too
while (!dmsgq_.empty()) {
if (msCurrent < dmsgq_.top().msTrigger_) {
cmsDelayNext = dmsgq_.top().msTrigger_ - msCurrent;
break;
}
msgq_.push(dmsgq_.top().msg_);
dmsgq_.pop();
}
// Check for posted events
while (!msgq_.empty()) {
*pmsg = msgq_.front();
if (pmsg->ts_sensitive) {
long delay = TimeDiff(msCurrent, pmsg->ts_sensitive);
if (delay > 0) {
LOG_F(LS_WARNING) << "id: " << pmsg->message_id << " delay: "
<< (delay + kMaxMsgLatency) << "ms";
}
}
msgq_.pop();
if (MQID_DISPOSE == pmsg->message_id) {
ASSERT(NULL == pmsg->phandler);
delete pmsg->pdata;
continue;
}
return true;
}
}
if (fStop_)
break;
// Which is shorter, the delay wait or the asked wait?
int cmsNext;
if (cmsWait == kForever) {
cmsNext = cmsDelayNext;
} else {
cmsNext = cmsTotal - cmsElapsed;
if (cmsNext < 0)
cmsNext = 0;
if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext))
cmsNext = cmsDelayNext;
}
// Wait and multiplex in the meantime
ss_->Wait(cmsNext, true);
// If the specified timeout expired, return
msCurrent = Time();
cmsElapsed = msCurrent - msStart;
if (cmsWait != kForever) {
if (cmsElapsed >= cmsWait)
return false;
}
}
return false;
}
Get方法如同注释所说主要做了以下几件事情。
1. 如果已经有消息被Peek，那么取出那条消息直接返回ok了，并且把PeekKeep的标志置为false。
2. 先去delayed队列取消息，看看有没有到时的消息。 如果有，就放到主消息队列中去。
3. 去主消息队列中取消息，如果有，就直接返回ok了，没有的话继续以下步骤。这边又有一个删除消息的消息，如果收到的是刚才那样一个消息的话，就需要执行dispose操作。
4. 查看stop标志，如果不结束的话就利用余下的时间就进行多路复用器的监听工作，来监听异步网络事件。看到这条的时候就明白为什么要在Post消息的时候WakeUp多路复用器了，快速响应消息是王道啊！
Peek消息:
Peek消息其实是查看消息，再去取的时候还能够得到该条消息。我一开始以为message不会从队列中Pop出来，不过这边他采取类似的做法。设一个消息变量来保存Peeked的msg，一个标志peekKeep_来表示消息是否被Peek出来保存到变量中。因此，当peekKeep_为true的时候，Get一下会取出变量里面的那条消息并且将标志置为false。
其他一些函数实现也比较基础，简单，这边就不说消息队列这个无聊的东西了。
三 SocketServer
接下去我们看看我最感兴趣的SocketServer了。
libjingle是gtalk的客户端，和服务器的底层socket通信主要就是使用了这个接口。
1. Socket:
SocketServer首先是个SocketFactory，所以它能Create出socket来，可以创建阻塞的和非阻塞的socket。
而创建出来的Socket则是对于不同os底层socket的高层抽象。
主要有Bind(), Connect(), Send(), Recv()等等。
而AsyncSocket则是有一些多了一些信号事件来给外界捕捉处理。
class AsyncSocket : public Socket, public sigslot::has_slots<> {
public:
virtual ~AsyncSocket() {}
sigslot::signal1<AsyncSocket*> SignalReadEvent; // ready to read
sigslot::signal1<AsyncSocket*> SignalWriteEvent; // ready to write
sigslot::signal1<AsyncSocket*> SignalConnectEvent; // connected
sigslot::signal2<AsyncSocket*,int> SignalCloseEvent; // closed
// TODO: error
};
这边用到了signal/slot机制，这个机制很棒，以前如果用过QT的朋友肯定能懂，和QT的signal/slot机制基本类似，libjingle里面的signal/slot库我还没来得及分析，不过短小精悍，应该不错，以后有机会再分析下，呵呵。 不过最重要的就是知道这个是一个类似与Observer模式的东西，最重要的就是其你发一个一开始有一个signal和一些slots用connect函数来绑定，signal就是被观察者，slot就是观察者。 如果有当signal触发的时候，slot就能接受到。
SocketServer是对监听网络事件的高层抽象，主要有两个函数:Wait()和Wakeup()。 监听的时候是Wait（），可以用WakeUp()来唤醒。
这里我们主要用到了异步的客户端。使用了select 多路复用器来监听异步网络事件。
当然抽象的好处就是可以有不同的实现，甚至是虚拟mock的实现。
而我们一般默认用到的就是PhysicalSocketServer，这个类封装了不同os之间网络库的差异性。
对于常用的一些socket函数的封装在这里我就不介绍了，主要介绍一下Wait()和Wakeup()函数中的工作机制。这边以linux为例子。
1. Dispatcher:
class Dispatcher {
public:
virtual uint32 GetRequestedEvents() = 0; // 返回感兴趣的事件
virtual void OnPreEvent(uint32 ff) = 0; // 在事件执行前
virtual void OnEvent(uint32 ff, int err) = 0; // 在事件执行时
virtual int GetDescriptor() = 0; // 返回Dispather的描述
};
这边引入了一个Dispatcher的概念，Dispatcher是用来派发事件的，因此对于多路复用器的主循环来说我们可以引入一个做法: 即当我们收到事件后我们使用不同的Dispatcher来进行派发，这里的事件可以是网络事件，也可以是其他可以被多路复用器选择出来的事件。
我们先来看网络事件派发器SocketDispatcher。
它主要是继承了PhysicalSocket和Dispatcher，然后在派发执行函数OnEvent()里面进行异步网络事件通知处理，signal感兴趣的 网络连接，读，写，关闭事件，并且将更新感兴趣的网络事件，去除已经做过的事件。
virtual void OnEvent(uint32 ff, int err) {
int cache_id = id_;
if ((ff & kfRead) != 0) {
enabled_events_ &= ~kfRead;
SignalReadEvent(this);
}
if (((ff & kfWrite) != 0) && (id_ == cache_id)) {
enabled_events_ &= ~kfWrite;
SignalWriteEvent(this);
}
if (((ff & kfConnect) != 0) && (id_ == cache_id)) {
if (ff != kfConnect)
LOG(LS_VERBOSE) << "Signalled with kfConnect: " << ff;
enabled_events_ &= ~kfConnect;
SignalConnectEvent(this);
}
if (((ff & kfClose) != 0) && (id_ == cache_id)) {
//LOG(INFO) << "SOCK[" << static_cast<int>(s_) << "] OnClose() Error: " << err;
signal_close_ = true;
signal_err_ = err;
}
}
还有一种就是其他的事件分派器， EventDispatcher在linux下面是用了pipe来模拟(windows则有event机制)。这样也可以被多路复用器select出来。
Signal的时候从一端往管道中写，而在PreOnEvent的时候从另一端读出来。
这边一个实例就是类Signaler。 该类继承了EventDispatcher来完成对退出wait，达到wakeup。
2. wait()函数详解:
这里我贴了一段wait()的linux实现代码，不包括里面的超时机制。
// Zero all fd_sets. Don't need to do this inside the loop since
// select() zeros the descriptors not signaled
fd_set fdsRead;
FD_ZERO(&fdsRead);
fd_set fdsWrite;
FD_ZERO(&fdsWrite);
fWait_ = true;
while (fWait_) {
int fdmax = -1;
{
CritScope cr(&crit_);
for (unsigned i = 0; i < dispatchers_.size(); i++) {
// Query dispatchers for read and write wait state
Dispatcher *pdispatcher = dispatchers_[i];
assert(pdispatcher);
if (!process_io && (pdispatcher != signal_wakeup_))
continue;
int fd = pdispatcher->GetDescriptor();
if (fd > fdmax)
fdmax = fd;
uint32 ff = pdispatcher->GetRequestedEvents();
if (ff & kfRead)
FD_SET(fd, &fdsRead);
if (ff & (kfWrite | kfConnect))
FD_SET(fd, &fdsWrite);
}
}
// Wait then call handlers as appropriate
// < 0 means error
// 0 means timeout
// > 0 means count of descriptors ready
int n = select(fdmax + 1, &fdsRead, &fdsWrite, NULL, ptvWait);
// If error, return error
// todo: do something intelligent
if (n < 0)
return false;
// If timeout, return success
if (n == 0)
return true;
// We have signaled descriptors
{
CritScope cr(&crit_);
for (unsigned i = 0; i < dispatchers_.size(); i++) {
Dispatcher *pdispatcher = dispatchers_[i];
int fd = pdispatcher->GetDescriptor();
uint32 ff = 0;
if (FD_ISSET(fd, &fdsRead)) {
FD_CLR(fd, &fdsRead);
ff |= kfRead;
}
if (FD_ISSET(fd, &fdsWrite)) {
FD_CLR(fd, &fdsWrite);
if (pdispatcher->GetRequestedEvents() & kfConnect) {
ff |= kfConnect;
} else {
ff |= kfWrite;
}
}
if (ff != 0) {
pdispatcher->OnPreEvent(ff);
pdispatcher->OnEvent(ff, 0);
}
}
}
1. 创建 fd_set并且初始化。
2. 对每个dispatcher的fd添加感兴趣的事件。
3. select()。
4. 更新每个dispatcher的感兴趣的并且发生的事件到ff，然后执行OnPreEvent和OnEvent。
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- libjingle源码分析之二：Thread和SocketServer
- libjingle源码分析之二：Thread和SocketServer
- python 从SocketServer到 WSGIServer 源码分析、
- skynet记录1：socket server源码分析
- apache kafka系列之源码分析走读-SocketServer分析
- Python之socketserver源码分析
- 《Java 源码分析》：Java NIO 之 ServerSocketChannel
- SocketServer模块源码分析
- Python标准库源码分析：SocketServer.py**********************8
- 《Java 源码分析》：Java NIO 之 ServerSocketChannel
- Python SocketServer源码分析
- 自动化运维Python系列之IO多路复用、SocketServer源码分析
- HBase源码分析之HRegionServer上MemStore的flush处理流程（二）
- live555源码分析----RSTPServer创建过程分析
- Full Multi-thread Client/Server Socket Class with ThreadPool
- Java NIO框架Netty教程（四） – ServerBootStrap启动流程源码分析
- zipkin-server源码分析（二）
- adb 深入分析一1、adb模块的源码路径，源码编译生成的结果。 2、adb模块分为adb client、adb server、services、adbd daemon几个模块。 3、adb通讯模
- 功能强大的IOCP Socket Server模块源码——完成端口通讯服务器（IOCP Socket Server）设计（六）
- Memcached源码分析之thread.c