高性能服务器架构设计总结¶
高性能服务器架构设计总结¶
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
这篇文章算是对这个系列的一个系统性地总结。我们将介绍服务器的开发，并从多个方面探究如何开发一款高性能高并发的服务器程序。
所谓 高性能 就是服务器能流畅地处理各个客户端的连接并尽量低延迟地应答客户端的请求；所谓 高并发，指的是服务器可以同时支持多的客户端连接，且这些客户端在连接期间内会不断与服务器有数据来往。
这篇文章将从两个方面来介绍，一个是 服务器的框架，即单个服务器程序的代码组织结构；另外一个是一组服务程序的 如何组织与交互，即架构。注意：本文以下内容中的客户端是相对概念，指的是连接到当前讨论的服务程序的终端，所以这里的客户端既可能是我们传统意义上的客户端程序，也可能是连接该服务的其他服务器程序。
一、框架篇¶
按上面介绍的思路，我们先从单个服务程序的组织结构开始介绍。
（一）、网络通信¶
既然是服务器程序肯定会涉及到网络通信部分，那么服务器程序的网络通信模块要解决哪些问题？
笔者认为至少要解决以下问题：
- 如何检测有新客户端连接？
- 如何接受客户端连接？
- 如何检测客户端是否有数据发来？
- 如何收取客户端发来的数据？
- 如何检测连接异常？发现连接异常之后，如何处理？
- 如何给客户端发送数据？
- 如何在给客户端发完数据后关闭连接？
稍微有点网络基础的人，都能回答上面说的其中几个问题，比如接收客户端连接用 socket API 的 accept 函数，收取客户端数据用 recv 函数，给客户端发送数据用 send 函数，检测客户端是否有新连接和客户端是否有新数据可以用 IO multiplexing 技术（IO 复用）的 select、poll、epoll 等 socket API。确实是这样的，这些基础的 socket API 构成了服务器网络通信的地基，不管网络通信框架设计的如何巧妙，都是在这些基础的 socket API 的基础上构建的。但是 如何巧妙地组织这些基础的 socket API，才是问题的关键。我们说服务器很高效，支持高并发，实际上只是一个技术实现手段，不管怎样从软件开发的角度来讲无非就是一个程序而已，所以，只要程序能最大可能地满足“尽量减少等待”就是高效。也就是说高效不是“忙的忙死，闲的闲死”，而是大家都可以闲着，但是如果有活要干，大家尽量一起干，而不是一部分忙着依次做事情 123456789，另外一部分闲在那里无所事事。说的可能有点抽象，下面我们来举一些例子具体来说明一下。
比如默认 recv 函数如果没有数据的时候，线程就会阻塞在那里； 默认 send 函数，如果 tcp 窗口不是足够大，数据发不出去也会阻塞在那里； connect 函数默认连接另外一端的时候，也会阻塞在那里； 又或者是给对端发送一份数据，需要等待对端回答，如果对方一直不应答，当前线程就阻塞在这里。 以上都不是高效服务器的开发思维方式，因为上面的例子都不满足“尽量减少等待”的原则，为什么一定要等待呢？有没用一种方法，这些过程不需要等待，最好是不仅不需要等待，而且这些事情完成之后能通知我。这样在这些本来用于等待的 cpu 时间片内，我就可以做一些其他的事情。有，也就是我们下文要讨论的 IO Multiplexing 技术（IO 复用技术）。
（二）、几种 IO 复用机制的比较¶
目前 windows 系统支持 select、WSAAsyncSelect、WSAEventSelect、完成端口（IOCP），linux 系统支持 select、poll、epoll。这里我们不具体介绍每个具体的函数的用法，我们来讨论一点深层次的东西，以上列举的 API 函数可以分为两个层次：
- 层次一 select 和 poll
- 层次二 WSAAsyncSelect、WSAEventSelect、完成端口（IOCP）、epoll
为什么这么分呢？先来介绍 第一层次，select 和 poll 函数本质上还是在一定时间内主动去查询 socket 句柄（可能是一个也可能是多个）上是否有事件，比如可读事件，可写事件或者出错事件，也就是说我们还是需要每隔一段时间内去主动去做这些检测，如果在这段时间内检测出一些事件来，我们这段时间就算没白花，但是倘若这段时间内没有事件呢？我们只能是做无用功了，说白了，还是在浪费时间，因为假如一个服务器有多个连接，在 cpu 时间片有限的情况下，我们花费了一定的时间检测了一部分 socket 连接，却发现它们什么事件都没有，而在这段时间内我们却有一些事情需要处理，那我们为什么要花时间去做这个检测呢？把这个时间用在做我们需要做的事情不好吗？所以对于服务器程序来说，要想高效，我们应该尽量避免花费时间主动去查询一些 socket 是否有事件，而是等这些 socket 有事件的时候告诉我们去处理。这也就是 层次二 的各个函数做的事情，它们实际相当于变主动查询是否有事件为当有事件时，系统会告诉我们，此时我们再去处理，也就是“好钢用在刀刃”上了。只不过层次二的函数通知我们的方式是各不相同，比如 WSAAsyncSelect 是利用 windows 消息队列的事件机制来通知我们设定的窗口过程函数，IOCP 是利用 GetQueuedCompletionStatus 返回正确的状态，epoll 是 epoll_wait 函数返回而已。
比如 connect 函数连接另外一端，如果连接 socket 是异步的，那么 connect 虽然不能立刻连接完成，但是也是会立刻返回，无需等待，等连接完成之后，WSAAsyncSelect 会返回 FD_CONNECT 事件告诉我们连接成功，epoll 会产生 EPOLLOUT 事件，我们也能知道连接完成。甚至 socket 有数据可读时，WSAAsyncSelect 产生 FD_READ 事件，epoll 产生 EPOLLIN 事件，等等。
所以有了上面的讨论，我们就可以得到网络通信检测可读可写或者出错事件的正确姿势。这是我这里提出的 第二个原则：尽量减少做无用功的时间。这个在服务程序资源够用的情况下可能体现不出来什么优势，但是如果有大量的任务要处理，个人觉得这个可能带来无用。
（三）、检测网络事件的正确姿势¶
根据上面的介绍，第一，为了避免无意义的等待时间，第二，不采用主动查询各个 socket 的事件，而是采用等待操作系统通知我们有事件的状态的策略。我们的socket 都要设置成异步的。在此基础上我们回到栏目（一）中提到的七个问题：
- 如何检测有新客户端连接？
- 如何接受客户端连接？
默认 accept 函数会阻塞在那里，如果 epoll 检测到侦听 socket 上有 EPOLLIN 事件，或者 WSAAsyncSelect 检测到有 FD_ACCEPT 事件，那么就表明此时有新连接到来，这个时候调用 accept 函数，就不会阻塞了。当然产生的新 socket 你应该也设置成非阻塞的。这样我们就能在新 socket 上收发数据了。
- 如何检测客户端是否有数据发来？
- 如何收取客户端发来的数据？
同理，我们也应该在 socket 上有可读事件的时候才去收取数据，这样我们调用 recv 或者 read 函数时不用等待。
至于一次性收多少数据好呢？
我们可以根据自己的需求来决定，甚至你可以在一个循环里面反复 recv 或者 read，对于非阻塞模式的 socket，如果没有数据了，recv 或者 read 也会立刻返回，错误码 EWOULDBLOCK 会表明当前已经没有数据了。示例：
1bool CIUSocket::Recv()
2{
3 int nRet = 0;
4
5 while(true)
6 {
7 char buff[512];
8 nRet = ::recv(m_hSocket, buff, 512, 0);
9 //一旦出现错误就立刻关闭Socket
10 if(nRet == SOCKET_ERROR)
11 {
12 if (::WSAGetLastError() == WSAEWOULDBLOCK)
13 break;
14 else
15 return false;
16 }
17 else if(nRet < 1)
18 return false;
19
20 m_strRecvBuf.append(buff, nRet);
21
22 ::Sleep(1);
23 }
24
25 return true;
26}
- 如何检测连接异常？发现连接异常之后，如何处理？
同样当我们收到异常事件后例如 EPOLLERR 或关闭事件 FD_CLOSE，我们就知道了有异常产生，我们对异常的处理一般就是关闭对应的 socket。另外，如果 send/recv 或者 read/write 函数对一个 socket 进行操作时，如果返回 0，那说明对端已经关闭了 socket，此时这路连接也没必要存在了，我们也可以关闭对应的 socket。
- 如何给客户端发送数据？
给客户端发送数据，比收数据要稍微麻烦一点，也是需要讲点技巧的。首先我们不能像检测数据可读一样检测数据可写，因为如果检测可写的话，一般情况下只要对端正常收取数据，我们的 socket 就都是可写的，如果我们设置监听可写事件，会导致频繁地触发可写事件，但是我们此时并不一定有数据需要发送。所以 正确的做法 是：如果有数据要发送，则先尝试着去发送，如果发送不了或者只发送出去部分，剩下的我们需要将其缓存起来，然后设置检测该 socket 上可写事件，下次可写事件产生时，再继续发送，如果还是不能完全发出去，则继续设置侦听可写事件，如此往复，一直到所有数据都发出去为止。一旦所有数据都发出去以后，我们要移除侦听可写事件，避免无用的可写事件通知。不知道你注意到没有，如果某次只发出去部分数据，剩下的数据应该暂且存起来，这个时候我们就需要一个缓冲区来存放这部分数据，这个缓冲区我们称为“发送缓冲区”。发送缓冲区不仅存放本次没有发完的数据，还用来存放在发送过程中，上层又传来的新的需要发送的数据。为了保证顺序，新的数据应该追加在当前剩下的数据的后面，发送的时候从发送缓冲区的头部开始发送。也就是说先来的先发送，后来的后发送。
- 如何在给客户端发完数据后关闭连接？
这个问题比较难处理，因为这里的“发送完”不一定是真正的发送完，我们调用 send 或者 write 函数即使成功，也只是向操作系统的协议栈里面成功写入数据，至于能否被发出去、何时被发出去很难判断，发出去对方是否收到就更难判断了。所以，我们目前只能简单地认为 send 或者 write 返回我们发出数据的字节数大小，我们就认为“发完数据”了。然后调用 close 等 socket API 关闭连接。关闭连接的话题，我们再单独开一个小的标题来专门讨论一下。
（四）被动关闭连接和主动关闭连接¶
在实际的应用中，被动关闭连接 是由于我们检测到了连接的异常事件，比如 EPOLLERR，或者对端关闭连接，send 或 recv 返回 0，这个时候这路连接已经没有存在必要的意义了，我们被迫关闭连接。
而 主动关闭连接，是我们主动调用 close/closesocket 来关闭连接。比如客户端给我们发送非法的数据，比如一些网络攻击的尝试性数据包。这个时候出于安全考虑，我们关闭 socket 连接。
（五）发送缓冲区和接收缓冲区¶
上面已经介绍了发送缓冲区了，并说明了其存在的意义。接收缓冲区也是一样的道理，当收到数据以后，我们可以直接进行解包，但是这样并不好，
- 理由一：除非一些约定俗称的协议格式，比如 http 协议，大多数服务器的业务的协议都是不同的，也就是说一个数据包里面的数据格式的解读应该是业务层的事情，和网络通信层应该解耦，为了网络层更加通用，我们无法知道上层协议长成什么样子，因为不同的协议格式是不一样的，它们与具体的业务有关。
- 理由二：即使知道协议格式，我们在网络层进行解包处理对应的业务，如果这个业务处理比较耗时，比如读取磁盘文件，或者连接数据库进行账号密码验证，那么我们的网络线程会需要大量时间来处理这些任务，这样其它网络事件可能没法及时处理。鉴于以上二点，我们确实需要一个接收缓冲区，将收取到的数据放到该缓冲区里面去，并由专门的业务线程或者业务逻辑去从接收缓冲区中取出数据，并解包处理业务。
说了这么多，那发送缓冲区和接收缓冲区该设计成多大的容量？这是一个老生常谈的问题了，因为我们经常遇到这样的问题：预分配的内存太小不够用，太大的话可能会造成浪费。怎么办呢？答案就是像 string、vector 一样，设计出一个可以动态增长的缓冲区，按需分配，不够还可以扩展。
需要特别注意的是，这里说的发送缓冲区和接收缓冲区是每一个 socket 连接都存在一个。这是我们最常见的设计方案。
（六）协议的设计¶
除了一些通用的协议，如 http、ftp 协议以外，大多数服务器协议都是根据业务制定的。协议设计好了，数据包的格式就根据协议来设置。我们知道 tcp/ip 协议是流式数据，所以流式数据就是像流水一样，数据包与数据包之间没有明显的界限。比如 A 端给 B 端连续发了三个数据包，每个数据包都是 50 个字节，B 端可能先收到 10 个字节，再收到 140 个字节；或者先收到 20 个字节，再收到 20 个字节，再收到 110 个字节；也可能一次性收到 150 个字节。这 150 个字节可以以任何字节数目组合和次数被 B 收到。
所以我们讨论协议的设计第一个问题就是如何界定包的界线，也就是接收端如何知道每个包数据的大小。目前常用有如下三种方法：
- 固定大小，这种方法就是假定每一个包的大小都是固定字节数目，比如上文中讨论的每个包大小都是 50 个字节，接收端每收气 50 个字节就当成一个包；
- 指定包结束符，比如以一个\r\n(换行符和回车符) 结束，这样对端只要收到这样的结束符，就可以认为收到了一个包，接下来的数据是下一个包的内容；
- 指定包的大小，这种方法结合了上述两种方法，一般包头是固定大小，包头中有一个字段指定包体或者整个大的大小，对端收到数据以后先解析包头中的字段得到包体或者整个包的大小，然后根据这个大小去界定数据的界线。
协议要讨论的第二个问题是，设计协议的时候要尽量方便解包，也就是说协议的格式字段应该尽量清晰明了。
协议要讨论的第三个问题是，根据协议组装的数据包应该尽量小，这样有如下好处：
- 第一、对于一些移动端设备来说，其数据处理能力和带宽能力有限，小的数据不仅能加快处理速度，同时节省大量流量费用；
- 第二、如果单个数据包足够小的话，对频繁进行网络通信的服务器端来说，可以大大减小其带宽压力，其所在的系统也能使用更少的内存。试想：假如一个股票服务器，如果一只股票的数据包是 100 个字节或者 1000 个字节，那 100 只股票和 10000 只股票区别呢？
协议要讨论的第四个问题是，对于数值类型，我们应该显式地指定数值的长度，比如 long 型，如果在 32 位机器上是 32 位的 4 个字节，但是如果在 64 位机器上，就变成了 64 位 8 个字节了。这样同样是一个 long 型，发送方和接收方可能会用不同的长度去解码。所以建议最好在涉及到跨平台使用的协议最好显式地指定协议中整型字段的长度，比如 int32,int64 等等。下面是一个协议的接口的例子：
1class BinaryReadStream
2{
3 private:
4 const char* const ptr;
5 const size_t len;
6 const char* cur;
7 BinaryReadStream(const BinaryReadStream&);
8 BinaryReadStream& operator=(const BinaryReadStream&);
9
10 public:
11 BinaryReadStream(const char* ptr, size_t len);
12 virtual const char* GetData() const;
13 virtual size_t GetSize() const;
14 bool IsEmpty() const;
15 bool ReadString(string* str,
16 size_t maxlen,
17 size_t& outlen);
18 bool ReadCString(char* str,
19 size_t strlen,
20 size_t& len);
21 bool ReadCCString(const char** str,
22 size_t maxlen,
23 size_t& outlen);
24 bool ReadInt32(int32_t& i);
25 bool ReadInt64(int64_t& i);
26 bool ReadShort(short& i);
27 bool ReadChar(char& c);
28 size_t ReadAll(char* szBuffer, size_t iLen) const;
29 bool IsEnd() const;
30 const char* GetCurrent() const{ return cur; }
31
32 public:
33 bool ReadLength(size_t & len);
34 bool ReadLengthWithoutOffset(size_t &headlen,
35 size_t & outlen);
36 };
37
38 class BinaryWriteStream
39 {
40 public:
41 BinaryWriteStream(string* data);
42 virtual const char* GetData() const;
43 virtual size_t GetSize() const;
44 bool WriteCString(const char* str, size_t len);
45 bool WriteString(const string& str);
46 bool WriteDouble(double value, bool isNULL = false);
47 bool WriteInt64(int64_t value, bool isNULL = false);
48 bool WriteInt32(int32_t i, bool isNULL = false);
49 bool WriteShort(short i, bool isNULL = false);
50 bool WriteChar(char c, bool isNULL = false);
51 size_t GetCurrentPos() const{ return m_data->length(); }
52 void Flush();
53 void Clear();
54 private:
55 string* m_data;
56 };
其中 BinaryWriteStream 是编码协议的类，BinaryReadStream 是解码协议的类。可以按下面这种方式来编码和解码。
编码：
1std::string outbuf;
2BinaryWriteStream writeStream(&outbuf);
3writeStream.WriteInt32(msg_type_register);
4writeStream.WriteInt32(m_seq);
5writeStream.WriteString(retData);
6writeStream.Flush();
解码：
1BinaryReadStream readStream(strMsg.c_str(),
2 strMsg.length());
3int32_t cmd;
4if (!readStream.ReadInt32(cmd))
5{
6 return false;
7}
8
9//int seq;
10if (!readStream.ReadInt32(m_seq))
11{
12 return false;
13}
14
15std::string data;
16size_t datalength;
17if (!readStream.ReadString(&data, 0, datalength))
18{
19 return false;
20}
（七）、服务器程序结构的组织¶
上面的六个标题，我们讨论了很多具体的细节问题，现在是时候讨论将这些细节组织起来了。根据我的个人经验，目前主流的思想是 one thread one loop 的策略。通俗点说就是在一个线程的函数里面不断地循环依次做一些事情，这些事情包括检测网络事件、解包数据产生业务逻辑。我们先从最简单地来说，设定一些线程在一个循环里面做网络通信相关的事情，伪码如下：
1while(退出标志)
2{
3 //IO复用技术检测socket可读事件、出错事件
4 //（如果有数据要发送，则也检测可写事件）
5
6 //如果有可读事件，对于侦听socket则接收新连接；
7 //对于普通socket则收取该socket上的数据，收取的数据存入对应的接收缓冲区，如果出错则关闭连接；
8
9 //如果有数据要发送，有可写事件，则发送数据
10
11 //如果有出错事件，关闭该连接
12}
` 另外设定一些线程去处理接收到的数据，并解包处理业务逻辑，这些线程可以认为是业务线程了，伪码如下：
1//从接收缓冲区中取出数据解包，分解成不同的业务来处理
上面的结构是目前最通用的服务器逻辑结构，但是能不能再简化一下或者说再综合一下呢？我们试试，你想过这样的问题没有：假如现在的机器有两个 cpu，我们的网络线程数量是 2个，业务逻辑线程也是 2个，这样可能存在的情况就是：业务线程运行的时候，网络线程并没有运行，它们必须等待，如果是这样的话，干嘛要多建两个线程呢？除了程序结构上可能稍微清楚一点，对程序性能没有任何实质性提高，而且白白浪费 cpu时间片在线程上下文切换上。所以，我们可以将网络线程与业务逻辑线程合并，合并后的伪码看起来是这样子的：
1while(退出标志)
2{
3 //IO复用技术检测socket可读事件、出错事件
4 //（如果有数据要发送，则也检测可写事件）
5
6 //如果有可读事件，对于侦听socket则接收新连接；
7 //对于普通socket则收取该socket上的数据，
8 //收取的数据存入对应的接收缓冲区，如果出错则关闭连接；
9
10 //如果有数据要发送，有可写事件，则发送数据
11
12 //如果有出错事件，关闭该连接
13
14 //从接收缓冲区中取出数据解包，分解成不同的业务来处理
15}
` 你没看错，其实就是简单的合并，合并之后和不仅可以达到原来合并前的效果，而且在没有网络 IO 事件的时候，可以及时处理我们想处理的一些业务逻辑，并且减少了不必要的线程上下文切换时间。
我们再更进一步，甚至我们可以在这个 while 循环增加其它的一些任务的处理，比如程序的逻辑任务队列、定时器事件等等，伪码如下：
1while(退出标志)
2{
3 //定时器事件处理
4
5 //IO复用技术检测socket可读事件、出错事件
6 //（如果有数据要发送，则也检测可写事件）
7
8 //如果有可读事件，对于侦听socket则接收新连接；
9 //对于普通socket则收取该socket上的数据，
10 //收取的数据存入对应的接收缓冲区，如果出错则关闭连接；
11
12 //如果有数据要发送，有可写事件，则发送数据
13
14 //如果有出错事件，关闭该连接
15
16 //从接收缓冲区中取出数据解包，分解成不同的业务来处理
17
18 //程序自定义任务1
19
20 //程序自定义任务2
21}
注意：之所以将定时器事件的处理放在网络 IO 事件的检测之前，是因为避免定时器事件过期时间太长。假如放在后面的话，可能前面的处理耗费了一点时间，等到处理定时器事件时，时间间隔已经过去了不少时间。虽然这样处理，也没法保证定时器事件百分百精确，但是能尽量保证。
（八）高性能服务器架构设计总结 2——以 flamigo 服务器代码为例¶
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
说了这么多，我们来以 flamingo 的服务器程序的网络框架设计为例来验证上述介绍的理论。flamingo 的网络框架是基于陈硕的 muduo 库，改成 C++11 的版本，并修改了一些 bug。在此感谢原作者陈硕。flamingo 的源码可以在这里下载：https://github.com/baloonwj/flamingo，打不开 github 的可以移步 csdn：http://download.csdn.net/detail/analogous_love/9805797。
上文介绍的核心线程函数的 while 循环位于 eventloop.cpp 中：
1void EventLoop::loop()
2{
3 assert(!looping_);
4 assertInLoopThread();
5 looping_ = true;
6 quit_ = false;
7 // FIXME: what if someone calls quit() before loop() ?
8 LOG_TRACE << "EventLoop " << this << " start looping";
9
10 while (!quit_)
11 {
12 activeChannels_.clear();
13 pollReturnTime_ = poller_->poll(kPollTimeMs,
14 &activeChannels_);
15 ++iteration_;
16 if (Logger::logLevel() <= Logger::TRACE)
17 {
18 printActiveChannels();
19 }
20 // TODO sort channel by priority
21 eventHandling_ = true;
22 for (ChannelList::iterator it = activeChannels_.begin();
23 it != activeChannels_.end(); ++it)
24 {
25 currentActiveChannel_ = *it;
26 currentActiveChannel_->handleEvent(pollReturnTime_);
27 }
28 currentActiveChannel_ = NULL;
29 eventHandling_ = false;
30 doPendingFunctors();
31
32 if (frameFunctor_)
33 {
34 frameFunctor_();
35 }
36 }
37
38 LOG_TRACE << "EventLoop " << this << " stop looping";
39 looping_ = false;
40}
poller_->poll 利用 epoll 分离网络事件，然后接着处理分离出来的网络事件，每一个客户端 socket 对应一个连接，即一个 TcpConnection 和 Channel 通道对象。currentActiveChannel_->handleEvent(pollReturnTime_)根据是可读、可写、出错事件来调用对应的处理函数，这些函数都是回调函数，程序初始化阶段设置进来的：
1void Channel::handleEvent(Timestamp receiveTime)
2{
3 std::shared_ptr<void> guard;
4 if (tied_)
5 {
6 guard = tie_.lock();
7 if (guard)
8 {
9 handleEventWithGuard(receiveTime);
10 }
11 }
12 else
13 {
14 handleEventWithGuard(receiveTime);
15 }
16}
17
18void Channel::handleEventWithGuard(Timestamp receiveTime)
19{
20 eventHandling_ = true;
21 LOG_TRACE << reventsToString();
22 if ((revents_ & POLLHUP) && !(revents_ & POLLIN))
23 {
24 if (logHup_)
25 {
26 LOG_WARN << "Channel::handle_event() POLLHUP";
27 }
28 if (closeCallback_) closeCallback_();
29 }
30
31 if (revents_ & POLLNVAL)
32 {
33 LOG_WARN << "Channel::handle_event() POLLNVAL";
34 }
35
36 if (revents_ & (POLLERR | POLLNVAL))
37 {
38 if (errorCallback_) errorCallback_();
39 }
40 if (revents_ & (POLLIN | POLLPRI | POLLRDHUP))
41 {
42 //当是侦听socket时，readCallback_指向Acceptor::handleRead
43 //当是客户端socket时，调用TcpConnection::handleRead
44 if (readCallback_) readCallback_(receiveTime);
45 }
46 if (revents_ & POLLOUT)
47 {
48 //如果是连接状态服的socket，
49 //则writeCallback_指向Connector::handleWrite()
50 if (writeCallback_) writeCallback_();
51 }
52 eventHandling_ = false;
53}
当然，这里 利用了 Channel 对象的“多态性”
- 如果是普通 socket，可读事件就会调用预先设置的回调函数；
- 但是如果是侦听 socket，则调用 Aceptor 对象的 handleRead() 来接收新连接：
1void Acceptor::handleRead()
2{
3 loop_->assertInLoopThread();
4 InetAddress peerAddr;
5 //FIXME loop until no more
6 int connfd = acceptSocket_.accept(&peerAddr);
7 if (connfd >= 0)
8 {
9 // string hostport = peerAddr.toIpPort();
10 // LOG_TRACE << "Accepts of " << hostport;
11 //newConnectionCallback_实际指向:
12 //TcpServer::newConnection(int sockfd, const InetAddress& peerAddr)
13 if (newConnectionCallback_)
14 {
15 newConnectionCallback_(connfd, peerAddr);
16 }
17 else
18 {
19 sockets::close(connfd);
20 }
21 }
22 else
23 {
24 LOG_SYSERR << "in Acceptor::handleRead";
25 // Read the section named "The special problem of
26 // accept()ing when you can't" in libev's doc.
27 // By Marc Lehmann, author of livev.
28 if (errno == EMFILE)
29 {
30 ::close(idleFd_);
31 idleFd_ = ::accept(acceptSocket_.fd(), NULL, NULL);
32 ::close(idleFd_);
33 idleFd_ = ::open("/dev/null", O_RDONLY | O_CLOEXEC);
34 }
35 }
36}
主循环里面的业务逻辑处理对应：
1doPendingFunctors();
2
3if (frameFunctor_)
4{
5 frameFunctor_();
6}
7
8
9void EventLoop::doPendingFunctors()
10{
11 std::vector<Functor> functors;
12 callingPendingFunctors_ = true;
13
14 {
15 std::unique_lock<std::mutex> lock(mutex_);
16 functors.swap(pendingFunctors_);
17 }
18
19 for (size_t i = 0; i < functors.size(); ++i)
20 {
21 functors[i]();
22 }
23 callingPendingFunctors_ = false;
24}
这里增加业务逻辑是增加执行任务的函数指针的，增加的任务保存在成员变量 pendingFunctors_ 中，这个变量是一个函数指针数组（vector 对象），执行的时候，调用每个函数就可以了。上面的代码 先利用一个栈变量将成员变量 pendingFunctors_ 里面的函数指针换过来，接下来对这个栈变量进行操作就可以了，这样减少了锁的粒度。
因为成员变量 pendingFunctors_ 在增加任务的时候，也会被用到，设计到多个线程操作，所以要 加锁，增加任务的地方是：
1void EventLoop::queueInLoop(const Functor& cb)
2{
3 {
4 std::unique_lock<std::mutex> lock(mutex_);
5 pendingFunctors_.push_back(cb);
6 }
7
8 if (!isInLoopThread() || callingPendingFunctors_)
9 {
10 wakeup();
11 }
12}
而 frameFunctor_ 就更简单了，就是通过设置一个函数指针就可以了。当然这里有个技巧性的东西，即增加任务的时候，为了能够立即执行，使用唤醒机制，通过往一个 fd 里面写入简单的几个字节，来唤醒 epoll，使其立刻返回，因为此时没有其它的 socke 有事件，这样接下来就执行刚才添加的任务了。
我们看一下数据收取的逻辑：
1void TcpConnection::handleRead(Timestamp receiveTime)
2{
3 loop_->assertInLoopThread();
4 int savedErrno = 0;
5 ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
6 if (n > 0)
7 {
8 /*messageCallback_指向
9 CTcpSession::OnRead(const std::shared_ptr<TcpConnection>& conn,
10 Buffer* pBuffer,
11 Timestamp receiveTime)*/
12 messageCallback_(shared_from_this(),
13 &inputBuffer_, receiveTime);
14 }
15 else if (n == 0)
16 {
17 handleClose();
18 }
19 else
20 {
21 errno = savedErrno;
22 LOG_SYSERR << "TcpConnection::handleRead";
23 handleError();
24 }
25}
将收到的数据放到接收缓冲区里面，将来我们来解包：
1void ClientSession::OnRead(const std::shared_ptr<TcpConnection>& conn, Buffer* pBuffer, Timestamp receivTime)
2{
3 while (true)
4 {
5 //不够一个包头大小
6 if (pBuffer->readableBytes() < (size_t)sizeof(msg))
7 {
8 LOG_INFO << "buffer is not enough for a package header, pBuffer->readableBytes()=" << pBuffer->readableBytes() << ", sizeof(msg)=" << sizeof(msg);
9 return;
10 }
11
12 //不够一个整包大小
13 msg header;
14 memcpy(&header, pBuffer->peek(), sizeof(msg));
15 if (pBuffer->readableBytes() < (size_t)header.packagesize + sizeof(msg))
16 return;
17
18 pBuffer->retrieve(sizeof(msg));
19 std::string inbuf;
20 inbuf.append(pBuffer->peek(), header.packagesize);
21 pBuffer->retrieve(header.packagesize);
22 if (!Process(conn, inbuf.c_str(), inbuf.length()))
23 {
24 LOG_WARN << "Process error, close TcpConnection";
25 conn->forceClose();
26 }
27 }// end while-loop
28
29}
先 判断接收缓冲区里面的数据是否够一个包头大小，如果够再判断够不够包头指定的包体大小，如果还是够的话，接着在 Process 函数里面处理该包。
再看看发送数据的逻辑：
1void TcpConnection::sendInLoop(const void* data,
2 size_t len)
3{
4 loop_->assertInLoopThread();
5 ssize_t nwrote = 0;
6 size_t remaining = len;
7 bool faultError = false;
8 if (state_ == kDisconnected)
9 {
10 LOG_WARN << "disconnected, give up writing";
11 return;
12 }
13 // if no thing in output queue, try writing directly
14 if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
15 {
16 nwrote = sockets::write(channel_->fd(), data, len);
17 if (nwrote >= 0)
18 {
19 remaining = len - nwrote;
20 if (remaining == 0 && writeCompleteCallback_)
21 {
22 loop_->queueInLoop(std::bind(writeCompleteCallback_,
23 shared_from_this()));
24 }
25 }
26 else // nwrote < 0
27 {
28 nwrote = 0;
29 if (errno != EWOULDBLOCK)
30 {
31 LOG_SYSERR << "TcpConnection::sendInLoop";
32 // FIXME: any others?
33 if (errno == EPIPE || errno == ECONNRESET)
34 {
35 faultError = true;
36 }
37 }
38 }
39 }
40
41 assert(remaining <= len);
42 if (!faultError && remaining > 0)
43 {
44 size_t oldLen = outputBuffer_.readableBytes();
45 if (oldLen + remaining >= highWaterMark_
46 && oldLen < highWaterMark_
47 && highWaterMarkCallback_)
48 {
49 loop_->queueInLoop(std::bind(highWaterMarkCallback_,
50 shared_from_this(),
51 oldLen + remaining));
52 }
53 outputBuffer_.append(static_cast<const char*>(data)+nwrote,
54 remaining);
55 if (!channel_->isWriting())
56 {
57 channel_->enableWriting();
58 }
59 }
60}
如果剩余的数据 remaining 大于则调用channel_->enableWriting();开始监听可写事件，可写事件处理如下：
1void TcpConnection::handleWrite()
2{
3 loop_->assertInLoopThread();
4 if (channel_->isWriting())
5 {
6 ssize_t n = sockets::write(channel_->fd(),
7 outputBuffer_.peek(),
8 outputBuffer_.readableBytes());
9 if (n > 0)
10 {
11 outputBuffer_.retrieve(n);
12 if (outputBuffer_.readableBytes() == 0)
13 {
14 channel_->disableWriting();
15 if (writeCompleteCallback_)
16 {
17 loop_->queueInLoop(std::bind(writeCompleteCallback_,
18 shared_from_this()));
19 }
20 if (state_ == kDisconnecting)
21 {
22 shutdownInLoop();
23 }
24 }
25 }
26 else
27 {
28 LOG_SYSERR << "TcpConnection::handleWrite";
29 // if (state_ == kDisconnecting)
30 // {
31 // shutdownInLoop();
32 // }
33 }
34 }
35 else
36 {
37 LOG_TRACE << "Connection fd = " << channel_->fd()
38 << " is down, no more writing";
39 }
40}
如果发送完数据以后调用 channel_->disableWriting(); 移除监听可写事件。
很多读者可能一直想问，文中不是说解包数据并处理逻辑是业务代码而非网络通信的代码，你这里貌似都混在一起了，其实没有，这里实际的业务代码处理都是框架曾提供的回调函数里面处理的，具体怎么处理，由框架使用者——业务层自己定义。
总结起来，实际上就是一个线程函数里一个 loop 那么点事情，不信你再看我曾经工作上的一个交易系统项目代码：
1void CEventDispatcher::Run()
2{
3 m_bShouldRun = true;
4 while(m_bShouldRun)
5 {
6 DispatchIOs();
7 SyncTime();
8 CheckTimer();
9 DispatchEvents();
10 }
11}
12
13
14void CEpollReactor::DispatchIOs()
15{
16 DWORD dwSelectTimeOut = SR_DEFAULT_EPOLL_TIMEOUT;
17 if (HandleOtherTask())
18 {
19 dwSelectTimeOut = 0;
20 }
21
22 struct epoll_event ev;
23 CEventHandlerIdMap::iterator itor = m_mapEventHandlerId.begin();
24 for(; itor!=m_mapEventHandlerId.end(); itor++)
25 {
26 CEventHandler *pEventHandler = (CEventHandler *)(*itor).first;
27 if(pEventHandler == NULL){
28 continue;
29 }
30 ev.data.ptr = pEventHandler;
31 ev.events = 0;
32 int nReadID, nWriteID;
33 pEventHandler->GetIds(&nReadID, &nWriteID);
34 if (nReadID > 0)
35 {
36 ev.events |= EPOLLIN;
37 }
38 if (nWriteID > 0)
39 {
40 ev.events |= EPOLLOUT;
41 }
42
43 epoll_ctl(m_fdEpoll, EPOLL_CTL_MOD, (*itor).second, &ev);
44 }
45
46 struct epoll_event events[EPOLL_MAX_EVENTS];
47
48 int nfds = epoll_wait(m_fdEpoll, events,
49 EPOLL_MAX_EVENTS,
50 dwSelectTimeOut/1000);
51
52 for (int i=0; i<nfds; i++)
53 {
54 struct epoll_event &evref = events[i];
55 CEventHandler *pEventHandler = (CEventHandler *)evref.data.ptr;
56 if ((evref.events|EPOLLIN)!=0 &&
57 m_mapEventHandlerId.find(pEventHandler)!=m_mapEventHandlerId.end())
58 {
59 pEventHandler->HandleInput();
60 }
61 if ((evref.events|EPOLLOUT)!=0 &&
62 m_mapEventHandlerId.find(pEventHandler)!=m_mapEventHandlerId.end())
63 {
64 pEventHandler->HandleOutput();
65 }
66 }
67}
68
69
70void CEventDispatcher::DispatchEvents()
71{
72 CEvent event;
73 CSyncEvent *pSyncEvent;
74 while(m_queueEvent.PeekEvent(event))
75 {
76 int nRetval;
77
78 if(event.pEventHandler != NULL)
79 {
80 nRetval = event.pEventHandler->HandleEvent(event.nEventID,
81 event.dwParam,
82 event.pParam);
83 }
84 else
85 {
86 nRetval = HandleEvent(event.nEventID,
87 event.dwParam,
88 event.pParam);
89 }
90
91 if(event.pAdd != NULL) //同步消息
92 {
93 pSyncEvent=(CSyncEvent *)event.pAdd;
94 pSyncEvent->nRetval = nRetval;
95 pSyncEvent->sem.UnLock();
96 }
97 }
98}
再看看蘑菇街开源的 TeamTalk 的源码（代码下载地址：https://github.com/baloonwj/TeamTalk）：
1void CEventDispatch::StartDispatch(uint32_t wait_timeout)
2{
3 fd_set read_set, write_set, excep_set;
4 timeval timeout;
5 timeout.tv_sec = 0;
6 // 10 millisecond
7 timeout.tv_usec = wait_timeout * 1000;
8
9 if(running)
10 return;
11 running = true;
12
13 while (running)
14 {
15 _CheckTimer();
16 _CheckLoop();
17
18 if (!m_read_set.fd_count &&
19 !m_write_set.fd_count &&
20 !m_excep_set.fd_count)
21 {
22 Sleep(MIN_TIMER_DURATION);
23 continue;
24 }
25
26 m_lock.lock();
27 memcpy(&read_set, &m_read_set,
28 sizeof(fd_set));
29 memcpy(&write_set, &m_write_set,
30 sizeof(fd_set));
31 memcpy(&excep_set, &m_excep_set,
32 sizeof(fd_set));
33 m_lock.unlock();
34
35 int nfds = select(0, &read_set,
36 &write_set,
37 &excep_set,
38 &timeout);
39
40 if (nfds == SOCKET_ERROR)
41 {
42 log("select failed, error code: %d", GetLastError());
43 Sleep(MIN_TIMER_DURATION);
44 continue; // select again
45 }
46
47 if (nfds == 0)
48 {
49 continue;
50 }
51
52 for (u_int i = 0; i < read_set.fd_count; i++)
53 {
54 //log("select return read count=%d\n", read_set.fd_count);
55 SOCKET fd = read_set.fd_array[i];
56 CBaseSocket* pSocket = FindBaseSocket((net_handle_t)fd);
57 if (pSocket)
58 {
59 pSocket->OnRead();
60 pSocket->ReleaseRef();
61 }
62 }
63
64 for (u_int i = 0; i < write_set.fd_count; i++)
65 {
66 //log("select return write count=%d\n", write_set.fd_count);
67 SOCKET fd = write_set.fd_array[i];
68 CBaseSocket* pSocket = FindBaseSocket((net_handle_t)fd);
69 if (pSocket)
70 {
71 pSocket->OnWrite();
72 pSocket->ReleaseRef();
73 }
74 }
75
76 for (u_int i = 0; i < excep_set.fd_count; i++)
77 {
78 //log("select return exception count=%d\n", excep_set.fd_count);
79 SOCKET fd = excep_set.fd_array[i];
80 CBaseSocket* pSocket = FindBaseSocket((net_handle_t)fd);
81 if (pSocket)
82 {
83 pSocket->OnClose();
84 pSocket->ReleaseRef();
85 }
86 }
87
88 }
89}
* 由于公众号文章字数有限，您可以接着阅读下*一篇:《服务器端编程心得（八）——高性能服务器架构设计总结 3——以 flamigo 服务器代码为例》。
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
（八）高性能服务器架构设计总结 3——以 flamigo 服务器代码为例¶
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
再看 filezilla，一款 ftp 工具的服务器端，它采用的是 Windows 的 WSAAsyncSelect 模型（代码下载地址：https://github.com/baloonwj/filezilla）：
1//Processes event notifications sent by the sockets or the layers
2 static LRESULT CALLBACK WindowProc(HWND hWnd,
3 UINT message,
4 WPARAM wParam,
5 LPARAM lParam)
6 {
7 if (message>=WM_SOCKETEX_NOTIFY)
8 {
9 //Verify parameters
10 ASSERT(hWnd);
11 CAsyncSocketExHelperWindow *pWnd=(CAsyncSocketExHelperWindow *)
12 GetWindowLongPtr(hWnd, GWLP_USERDATA);
13 ASSERT(pWnd);
14 if (!pWnd)
15 return 0;
16 //Index is within socket storage
17 if (message < static_cast<UINT>(WM_SOCKETEX_NOTIFY+pWnd->m_nWindowDataSize))
18 {
19 //Lookup socket and verify if it's valid
20 CAsyncSocketEx *pSocket=pWnd->m_pAsyncSocketExWindowData[message - WM_SOCKETEX_NOTIFY].m_pSocket;
21 SOCKET hSocket = wParam;
22 if (!pSocket)
23 return 0;
24 if (hSocket == INVALID_SOCKET)
25 return 0;
26 if (pSocket->m_SocketData.hSocket != hSocket)
27 return 0;
28
29 int nEvent = lParam & 0xFFFF;
30 int nErrorCode = lParam >> 16;
31
32 //Dispatch notification
33 if (!pSocket->m_pFirstLayer) {
34 //Dispatch to CAsyncSocketEx instance
35 switch (nEvent)
36 {
37 case FD_READ:
38#ifndef NOSOCKETSTATES
39 if (pSocket->GetState() == connecting && !nErrorCode)
40 {
41 pSocket->m_nPendingEvents |= FD_READ;
42 break;
43 }
44 else if (pSocket->GetState() == attached)
45 pSocket->SetState(connected);
46 if (pSocket->GetState() != connected)
47 break;
48
49 // Ignore further FD_READ events after FD_CLOSE has been received
50 if (pSocket->m_SocketData.onCloseCalled)
51 break;
52#endif //NOSOCKETSTATES
53
54#ifndef NOSOCKETSTATES
55 if (nErrorCode)
56 pSocket->SetState(aborted);
57#endif //NOSOCKETSTATES
58 if (pSocket->m_lEvent & FD_READ) {
59 pSocket->OnReceive(nErrorCode);
60 }
61 break;
62 case FD_FORCEREAD: //Forceread does not check if there's data waiting
63#ifndef NOSOCKETSTATES
64 if (pSocket->GetState() == connecting && !nErrorCode)
65 {
66 pSocket->m_nPendingEvents |= FD_FORCEREAD;
67 break;
68 }
69 else if (pSocket->GetState() == attached)
70 pSocket->SetState(connected);
71 if (pSocket->GetState() != connected)
72 break;
73#endif //NOSOCKETSTATES
74 if (pSocket->m_lEvent & FD_READ)
75 {
76#ifndef NOSOCKETSTATES
77 if (nErrorCode)
78 pSocket->SetState(aborted);
79#endif //NOSOCKETSTATES
80 pSocket->OnReceive(nErrorCode);
81 }
82 break;
83 case FD_WRITE:
84#ifndef NOSOCKETSTATES
85 if (pSocket->GetState() == connecting && !nErrorCode)
86 {
87 pSocket->m_nPendingEvents |= FD_WRITE;
88 break;
89 }
90 else if (pSocket->GetState() == attached && !nErrorCode)
91 pSocket->SetState(connected);
92 if (pSocket->GetState() != connected)
93 break;
94#endif //NOSOCKETSTATES
95 if (pSocket->m_lEvent & FD_WRITE)
96 {
97#ifndef NOSOCKETSTATES
98 if (nErrorCode)
99 pSocket->SetState(aborted);
100#endif //NOSOCKETSTATES
101 pSocket->OnSend(nErrorCode);
102 }
103 break;
104 case FD_CONNECT:
105#ifndef NOSOCKETSTATES
106 if (pSocket->GetState() == connecting)
107 {
108 if (nErrorCode && pSocket->m_SocketData.nextAddr)
109 {
110 if (pSocket->TryNextProtocol())
111 break;
112 }
113 pSocket->SetState(connected);
114 }
115 else if (pSocket->GetState() == attached && !nErrorCode)
116 pSocket->SetState(connected);
117#endif //NOSOCKETSTATES
118 if (pSocket->m_lEvent & FD_CONNECT)
119 pSocket->OnConnect(nErrorCode);
120#ifndef NOSOCKETSTATES
121 if (!nErrorCode)
122 {
123 if ((pSocket->m_nPendingEvents&FD_READ) &&
124 pSocket->GetState() == connected)
125 pSocket->OnReceive(0);
126 if ((pSocket->m_nPendingEvents&FD_FORCEREAD) &&
127 pSocket->GetState() == connected)
128 pSocket->OnReceive(0);
129 if ((pSocket->m_nPendingEvents&FD_WRITE) &&
130 pSocket->GetState() == connected)
131 pSocket->OnSend(0);
132 }
133 pSocket->m_nPendingEvents = 0;
134#endif
135 break;
136 case FD_ACCEPT:
137#ifndef NOSOCKETSTATES
138 if (pSocket->GetState() != listening &&
139 pSocket->GetState() != attached)
140 break;
141#endif //NOSOCKETSTATES
142 if (pSocket->m_lEvent & FD_ACCEPT)
143 pSocket->OnAccept(nErrorCode);
144 break;
145 case FD_CLOSE:
146#ifndef NOSOCKETSTATES
147 if (pSocket->GetState() != connected &&
148 pSocket->GetState() != attached)
149 break;
150
151 // If there are still bytes left to read,
152 // call OnReceive instead of
153 // OnClose and trigger a new OnClose
154 DWORD nBytes = 0;
155 if (!nErrorCode && pSocket->IOCtl(FIONREAD, &nBytes))
156 {
157 if (nBytes > 0)
158 {
159 // Just repeat message.
160 pSocket->ResendCloseNotify();
161 pSocket->m_SocketData.onCloseCalled = true;
162 pSocket->OnReceive(WSAESHUTDOWN);
163 break;
164 }
165 }
166
167 pSocket->SetState(nErrorCode ? aborted : closed);
168#endif //NOSOCKETSTATES
169 pSocket->OnClose(nErrorCode);
170 break;
171 }
172 }
173 else //Dispatch notification to the lowest layer
174 {
175 if (nEvent == FD_READ)
176 {
177 // Ignore further FD_READ events after FD_CLOSE has been received
178 if (pSocket->m_SocketData.onCloseCalled)
179 return 0;
180
181 DWORD nBytes;
182 if (!pSocket->IOCtl(FIONREAD, &nBytes))
183 nErrorCode = WSAGetLastError();
184 if (pSocket->m_pLastLayer)
185 pSocket->m_pLastLayer->CallEvent(nEvent, nErrorCode);
186 }
187 else if (nEvent == FD_CLOSE)
188 {
189 // If there are still bytes left to read,
190 // call OnReceive instead of
191 // OnClose and trigger a new OnClose
192 DWORD nBytes = 0;
193 if (!nErrorCode && pSocket->IOCtl(FIONREAD, &nBytes))
194 {
195 if (nBytes > 0)
196 {
197 // Just repeat message.
198 pSocket->ResendCloseNotify();
199 if (pSocket->m_pLastLayer)
200 pSocket->m_pLastLayer->CallEvent(FD_READ, 0);
201 return 0;
202 }
203 }
204 pSocket->m_SocketData.onCloseCalled = true;
205 if (pSocket->m_pLastLayer)
206 pSocket->m_pLastLayer->CallEvent(nEvent, nErrorCode);
207 }
208 else if (pSocket->m_pLastLayer)
209 pSocket->m_pLastLayer->CallEvent(nEvent, nErrorCode);
210 }
211 }
212 return 0;
213 }
214 else if (message == WM_USER) //Notification event sent by a layer
215 {
216 //Verify parameters, lookup socket and notification message
217 //Verify parameters
218 ASSERT(hWnd);
219 CAsyncSocketExHelperWindow *pWnd=(CAsyncSocketExHelperWindow *)
220 GetWindowLongPtr(hWnd, GWLP_USERDATA);
221 ASSERT(pWnd);
222 if (!pWnd)
223 return 0;
224 //Index is within socket storage
225 if (wParam >= static_cast<UINT>(pWnd->m_nWindowDataSize))
226 {
227 return 0;
228 }
229
230 CAsyncSocketEx *pSocket = pWnd->m_pAsyncSocketExWindowData[wParam].m_pSocket;
231 CAsyncSocketExLayer::t_LayerNotifyMsg *pMsg = (CAsyncSocketExLayer::t_LayerNotifyMsg *)lParam;
232 if (!pMsg || !pSocket || pSocket->m_SocketData.hSocket != pMsg->hSocket)
233 {
234 delete pMsg;
235 return 0;
236 }
237 int nEvent=pMsg->lEvent&0xFFFF;
238 int nErrorCode=pMsg->lEvent>>16;
239
240 //Dispatch to layer
241 if (pMsg->pLayer)
242 pMsg->pLayer->CallEvent(nEvent, nErrorCode);
243 else
244 {
245 //Dispatch to CAsyncSocketEx instance
246 switch (nEvent)
247 {
248 case FD_READ:
249#ifndef NOSOCKETSTATES
250 if (pSocket->GetState() == connecting && !nErrorCode)
251 {
252 pSocket->m_nPendingEvents |= FD_READ;
253 break;
254 }
255 else if (pSocket->GetState() == attached && !nErrorCode)
256 pSocket->SetState(connected);
257 if (pSocket->GetState() != connected)
258 break;
259#endif //NOSOCKETSTATES
260 if (pSocket->m_lEvent & FD_READ)
261 {
262#ifndef NOSOCKETSTATES
263 if (nErrorCode)
264 pSocket->SetState(aborted);
265#endif //NOSOCKETSTATES
266 pSocket->OnReceive(nErrorCode);
267 }
268 break;
269 //Forceread does not check if there's data waiting
270 case FD_FORCEREAD:
271#ifndef NOSOCKETSTATES
272 if (pSocket->GetState() == connecting && !nErrorCode)
273 {
274 pSocket->m_nPendingEvents |= FD_FORCEREAD;
275 break;
276 }
277 else if (pSocket->GetState() == attached && !nErrorCode)
278 pSocket->SetState(connected);
279 if (pSocket->GetState() != connected)
280 break;
281#endif //NOSOCKETSTATES
282 if (pSocket->m_lEvent & FD_READ)
283 {
284#ifndef NOSOCKETSTATES
285 if (nErrorCode)
286 pSocket->SetState(aborted);
287#endif //NOSOCKETSTATES
288 pSocket->OnReceive(nErrorCode);
289 }
290 break;
291 case FD_WRITE:
292#ifndef NOSOCKETSTATES
293 if (pSocket->GetState() == connecting && !nErrorCode)
294 {
295 pSocket->m_nPendingEvents |= FD_WRITE;
296 break;
297 }
298 else if (pSocket->GetState() == attached && !nErrorCode)
299 pSocket->SetState(connected);
300 if (pSocket->GetState() != connected)
301 break;
302#endif //NOSOCKETSTATES
303 if (pSocket->m_lEvent & FD_WRITE)
304 {
305#ifndef NOSOCKETSTATES
306 if (nErrorCode)
307 pSocket->SetState(aborted);
308#endif //NOSOCKETSTATES
309 pSocket->OnSend(nErrorCode);
310 }
311 break;
312 case FD_CONNECT:
313#ifndef NOSOCKETSTATES
314 if (pSocket->GetState() == connecting)
315 pSocket->SetState(connected);
316 else if (pSocket->GetState() == attached && !nErrorCode)
317 pSocket->SetState(connected);
318#endif //NOSOCKETSTATES
319 if (pSocket->m_lEvent & FD_CONNECT)
320 pSocket->OnConnect(nErrorCode);
321#ifndef NOSOCKETSTATES
322 if (!nErrorCode)
323 {
324 if (((pSocket->m_nPendingEvents&FD_READ) &&
325 pSocket->GetState() == connected) &&
326 (pSocket->m_lEvent & FD_READ))
327 pSocket->OnReceive(0);
328 if (((pSocket->m_nPendingEvents&FD_FORCEREAD) &&
329 pSocket->GetState() == connected) &&
330 (pSocket->m_lEvent & FD_READ))
331 pSocket->OnReceive(0);
332 if (((pSocket->m_nPendingEvents&FD_WRITE) &&
333 pSocket->GetState() == connected) &&
334 (pSocket->m_lEvent & FD_WRITE))
335 pSocket->OnSend(0);
336 }
337 pSocket->m_nPendingEvents = 0;
338#endif //NOSOCKETSTATES
339 break;
340 case FD_ACCEPT:
341#ifndef NOSOCKETSTATES
342 if ((pSocket->GetState() == listening || pSocket->GetState() == attached) && (pSocket->m_lEvent & FD_ACCEPT))
343#endif //NOSOCKETSTATES
344 {
345 pSocket->OnAccept(nErrorCode);
346 }
347 break;
348 case FD_CLOSE:
349#ifndef NOSOCKETSTATES
350 if ((pSocket->GetState() == connected ||
351 pSocket->GetState() == attached) &&
352 (pSocket->m_lEvent & FD_CLOSE))
353 {
354 pSocket->SetState(nErrorCode?aborted:closed);
355#else
356 {
357#endif //NOSOCKETSTATES
358 pSocket->OnClose(nErrorCode);
359 }
360 break;
361 }
362 }
363 delete pMsg;
364 return 0;
365 }
366 else if (message == WM_USER+1)
367 {
368 // WSAAsyncGetHostByName reply
369
370 // Verify parameters
371 ASSERT(hWnd);
372 CAsyncSocketExHelperWindow *pWnd = (CAsyncSocketExHelperWindow *)
373 GetWindowLongPtr(hWnd, GWLP_USERDATA);
374 ASSERT(pWnd);
375 if (!pWnd)
376 return 0;
377
378 CAsyncSocketEx *pSocket = NULL;
379 for (int i = 0; i < pWnd->m_nWindowDataSize; ++i) {
380 pSocket = pWnd->m_pAsyncSocketExWindowData[i].m_pSocket;
381 if (pSocket && pSocket->m_hAsyncGetHostByNameHandle &&
382 pSocket->m_hAsyncGetHostByNameHandle == (HANDLE)wParam &&
383 pSocket->m_pAsyncGetHostByNameBuffer)
384 break;
385 }
386 if (!pSocket || !pSocket->m_pAsyncGetHostByNameBuffer)
387 return 0;
388
389 int nErrorCode = lParam >> 16;
390 if (nErrorCode) {
391 pSocket->OnConnect(nErrorCode);
392 return 0;
393 }
394
395 SOCKADDR_IN sockAddr{};
396 sockAddr.sin_family = AF_INET;
397 sockAddr.sin_addr.s_addr = ((LPIN_ADDR)((LPHOSTENT)pSocket->m_pAsyncGetHostByNameBuffer)->h_addr)->s_addr;
398
399 sockAddr.sin_port = htons(pSocket->m_nAsyncGetHostByNamePort);
400
401 BOOL res = pSocket->Connect((SOCKADDR*)&sockAddr, sizeof(sockAddr));
402 delete [] pSocket->m_pAsyncGetHostByNameBuffer;
403 pSocket->m_pAsyncGetHostByNameBuffer = 0;
404 pSocket->m_hAsyncGetHostByNameHandle = 0;
405
406 if (!res)
407 if (GetLastError() != WSAEWOULDBLOCK)
408 pSocket->OnConnect(GetLastError());
409 return 0;
410 }
411 else if (message == WM_USER + 2)
412 {
413 //Verify parameters, lookup socket and notification message
414 //Verify parameters
415 if (!hWnd)
416 return 0;
417
418 CAsyncSocketExHelperWindow *pWnd=(CAsyncSocketExHelperWindow *)
419 GetWindowLongPtr(hWnd, GWLP_USERDATA);
420 if (!pWnd)
421 return 0;
422
423 if (wParam >= static_cast<UINT>(pWnd->m_nWindowDataSize)) //Index is within socket storage
424 return 0;
425
426 CAsyncSocketEx *pSocket = pWnd->m_pAsyncSocketExWindowData[wParam].m_pSocket;
427 if (!pSocket)
428 return 0;
429
430 // Process pending callbacks
431 std::list<t_callbackMsg> tmp;
432 tmp.swap(pSocket->m_pendingCallbacks);
433 pSocket->OnLayerCallback(tmp);
434
435 for (auto & cb : tmp) {
436 delete [] cb.str;
437 }
438 }
439 else if (message == WM_TIMER)
440 {
441 if (wParam != 1)
442 return 0;
443
444 ASSERT(hWnd);
445 CAsyncSocketExHelperWindow *pWnd=(CAsyncSocketExHelperWindow *)
446 GetWindowLongPtr(hWnd, GWLP_USERDATA);
447 ASSERT(pWnd && pWnd->m_pThreadData);
448 if (!pWnd || !pWnd->m_pThreadData)
449 return 0;
450
451 if (pWnd->m_pThreadData->layerCloseNotify.empty())
452 {
453 KillTimer(hWnd, 1);
454 return 0;
455 }
456 CAsyncSocketEx* socket = pWnd->m_pThreadData->layerCloseNotify.front();
457 pWnd->m_pThreadData->layerCloseNotify.pop_front();
458 if (pWnd->m_pThreadData->layerCloseNotify.empty())
459 KillTimer(hWnd, 1);
460
461 if (socket)
462 PostMessage(hWnd,
463 socket->m_SocketData.nSocketIndex + WM_SOCKETEX_NOTIFY,
464 socket->m_SocketData.hSocket,
465 FD_CLOSE);
466 return 0;
467 }
468 return DefWindowProc(hWnd, message, wParam, lParam);
469 }
关于单个服务程序的框架，我已经介绍完了，如果你能完全理解我要表达的意思，我相信你也能构建出一套高性能服务程序来。
（八）高性能服务器架构设计总结 4——以 flamigo 服务器代码为例¶
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
二、架构篇¶
一个项目的服务器端往往由很多服务组成，就算单个服务在性能上做到极致，支持的并发数量也是有限的。举个简单的例子，假如一个聊天服务器，每个用户的信息是 1k，那对于一个 8G 的内存的机器，在不考虑其它的情况下 8102410241024 / 100 = 1024，实际有 838 万，但实际这只是非常理想的情况。所以我们有时候需要需要某个服务部署多套，就单个服务的实现来讲还是《框架篇》中介绍的*。
1
我们举个例子：
这是蘑菇街 TeamTalk 的服务器架构。
MsgServer 是聊天服务，可以部署多套，每个聊天服务器启动时都会告诉 loginSever 和 routeSever 自己的 ip 地址和端口号，当有用户上线或者下线的时候，MsgServer 也会告诉 loginSever 和 routeSever 自己上面最新的用户数量和用户 id 列表。
现在一个用户需要 登录，先连接 loginServer，loginServer 根据记录的各个 MsgServer 上的用户情况，返回一个最小负载的 MsgServer 的 ip 地址和端口号给客户端，客户端再利用这个 ip 地址和端口号去登录 MsgServer。当 聊天 时，位于 A MsgServer 上的用户给另外一个用户发送消息，如果该用户不在同一个 MsgServer 上，MsgServer 将消息转发给 RouteServer，RouteServer 根据自己记录的用户 id 信息找到目标用户所在的 MsgServer 并转发给对应的 MsgServer。
2
上面是分布式部署的一个例子，我们再来看另外一个例子。这个例子是单个服务的策略，实际服务器在处理网络数据的时候，如果同时有多个 socket 上有数据要处理，可能会出现一直服务前几个 socket，直到前几个 socket 处理完毕后再处理后面几个 socket 的数据。这就相当于，你去饭店吃饭，大家都点了菜，但是有些桌子上一直在上菜，而有些桌子上一直没有菜。这样肯定不好，我们来看下如何避免这种现象：
1int CFtdEngine::HandlePackage(CFTDCPackage *pFTDCPackage,
2 CFTDCSession *pSession)
3{
4 //NET_IO_LOG0("CFtdEngine::HandlePackage\n");
5 FTDC_PACKAGE_DEBUG(pFTDCPackage);
6
7 if (pFTDCPackage->GetTID() != FTD_TID_ReqUserLogin)
8 {
9 if (!IsSessionLogin(pSession->GetSessionID()))
10 {
11 SendErrorRsp(pFTDCPackage, pSession, 1, "客户未登录");
12 return 0;
13 }
14 }
15
16 CalcFlux(pSession, pFTDCPackage->Length()); //统计流量
17
18 REPORT_EVENT(LOG_DEBUG, "Front/Fgateway", "登录请求%0x",
19 pFTDCPackage->GetTID());
20
21 int nRet = 0;
22 switch(pFTDCPackage->GetTID())
23 {
24
25 case FTD_TID_ReqUserLogin:
26 ///huwp：20070608：检查过高版本的API将被禁止登录
27 if (pFTDCPackage->GetVersion()>FTD_VERSION)
28 {
29 SendErrorRsp(pFTDCPackage, pSession, 1,
30 "Too High FTD Version");
31 return 0;
32 }
33 nRet = OnReqUserLogin(pFTDCPackage,
34 (CFTDCSession *)pSession);
35 FTDRequestIndex.incValue();
36 break;
37 case FTD_TID_ReqCheckUserLogin:
38 nRet = OnReqCheckUserLogin(pFTDCPackage,
39 (CFTDCSession *)pSession);
40 FTDRequestIndex.incValue();
41 break;
42 case FTD_TID_ReqSubscribeTopic:
43 nRet = OnReqSubscribeTopic(pFTDCPackage,
44 (CFTDCSession *)pSession);
45 FTDRequestIndex.incValue();
46 break;
47 }
48
49 return 0;
50}
当 有某个 socket 上有数据可读时，接着接收该 socket 上的数据，对接收到的数据进行解包，然后调用 CalcFlux(pSession, pFTDCPackage->Length()) 进行流量统计：
1void CFrontEngine::CalcFlux(CSession *pSession, const int nFlux)
2{
3 TFrontSessionInfo *pSessionInfo = m_mapSessionInfo.Find(pSession->GetSessionID());
4 if (pSessionInfo != NULL)
5 {
6 //流量控制改为计数
7 pSessionInfo->nCommFlux ++;
8 ///若流量超过规定，则挂起该会话的读操作
9 if (pSessionInfo->nCommFlux >= pSessionInfo->nMaxCommFlux)
10 {
11 pSession->SuspendRead(true);
12 }
13 }
14}
该函数会先让某个连接会话（Session）处理的包数量递增，接着判断是否超过最大包数量，则 设置读挂起标志：
1void CSession::SuspendRead(bool bSuspend)
2{
3 m_bSuspendRead = bSuspend;
4}
这样下次将会从检测的 socket 列表中 排除该 socket：
1void CEpollReactor::RegisterIO(CEventHandler *pEventHandler)
2{
3 int nReadID, nWriteID;
4 pEventHandler->GetIds(&nReadID, &nWriteID);
5 if (nWriteID != 0 && nReadID ==0)
6 {
7 nReadID = nWriteID;
8 }
9 if (nReadID != 0)
10 {
11 m_mapEventHandlerId[pEventHandler] = nReadID;
12 struct epoll_event ev;
13 ev.data.ptr = pEventHandler;
14 if(epoll_ctl(m_fdEpoll, EPOLL_CTL_ADD, nReadID, &ev) != 0)
15 {
16 perror("epoll_ctl EPOLL_CTL_ADD");
17 }
18 }
19}
20
21
22void CSession::GetIds(int *pReadId, int *pWriteId)
23{
24 m_pChannelProtocol->GetIds(pReadId,pWriteId);
25 if (m_bSuspendRead)
26 {
27 *pReadId = 0;
28 }
29}
也就是说不再检测该 socket 上是否有数据可读。然后在定时器里 1 秒后 重置该标志，这样这个 socket 上有数据的话又可以重新检测到了：
1const int SESSION_CHECK_TIMER_ID = 9;
2const int SESSION_CHECK_INTERVAL = 1000;
3
4
5SetTimer(SESSION_CHECK_TIMER_ID, SESSION_CHECK_INTERVAL);
6
7
8void CFrontEngine::OnTimer(int nIDEvent)
9{
10 if (nIDEvent == SESSION_CHECK_TIMER_ID)
11 {
12 CSessionMap::iterator itor = m_mapSession.Begin();
13 while (!itor.IsEnd())
14 {
15 TFrontSessionInfo *pFind = m_mapSessionInfo.Find((*itor)->GetSessionID());
16 if (pFind != NULL)
17 {
18 CheckSession(*itor, pFind);
19 }
20 itor++;
21 }
22 }
23}
24
25void CFrontEngine::CheckSession(CSession *pSession,
26 TFrontSessionInfo *pSessionInfo)
27{
28 ///重新开始计算流量
29 pSessionInfo->nCommFlux -= pSessionInfo->nMaxCommFlux;
30 if (pSessionInfo->nCommFlux < 0)
31 {
32 pSessionInfo->nCommFlux = 0;
33 }
34 ///若流量超过规定，则挂起该会话的读操作
35 pSession->SuspendRead(pSessionInfo->nCommFlux >= pSessionInfo->nMaxCommFlux);
36}
这就相当与饭店里面先给某一桌客人上一些菜，让他们先吃着，等上了一些菜之后不会再给这桌继续上菜了，而是给其它空桌上菜，大家都吃上后，继续回来给原先的桌子继续上菜。实际上我们的饭店都是这么做的。上面的例子是单服务流量控制的实现的一个非常好的思路，它保证了每个客户端都能均衡地得到服务，而不是一些客户端等很久才有响应。
3
另外加快服务器处理速度的策略可能就是 缓存 了，缓存实际上是以空间换取时间的策略。对于一些反复使用的，但是不经常改变的信息，如果从原始地点加载这些信息就比较耗时的数据（比如从磁盘中、从数据库中），我们就可以使用缓存。
所以时下像 redis、leveldb、fastdb 等各种内存数据库大行其道。我在 flamingo 中用户的基本信息都是缓存在聊天服务程序中的，而文件服务启动时会去加载指定目录里面的所有程序名称，这些文件的名称都是 md5，为该文件内容的 md5。这样当客户端上传了新文件请求时，如果其传上来的文件 md5 已经位于缓存中，则表明该文件在服务器上已经存在，这个时候服务器就不必再接收该文件了，而是告诉客户端文件已经上传成功了。
说了这么多，一般来说，一个服务器的架构，往往更多取决于其具体的业务，我们要在结合当前的情况来实际去组织铺排，没有一套系统是万能的。多思考，多实践，多总结，相信很快你也能拥有很不错的架构能力。
鉴于笔者能力和经验有限，文中难免有错漏之处，欢迎提意见。 交流 QQ 群：49114021
本系列完
系列目录
第 01 篇 主线程与工作线程的分工
第 02 篇 Reactor 模式
第 03 篇 一个服务器程序的架构介绍
第 04 篇 如何将 socket 设置为非阻塞模式
第 05 篇 如何编写高性能日志
第 06 篇 关于网络编程的一些实用技巧和细节
第 07 篇 开源一款即时通讯软件的源码
第 08 篇 高性能服务器架构设计总结 1
第 09 篇 高性能服务器架构设计总结 2
第 10 篇 高性能服务器架构设计总结 3
第 11 篇 高性能服务器架构设计总结 4
欢迎关注公众号『easyserverdev』。如果有任何技术或者职业方面的问题需要我提供帮助，可通过这个公众号与我取得联系，此公众号不仅分享高性能服务器开发经验和故事，同时也免费为广大技术朋友提供技术答疑和职业解惑，您有任何问题都可以在微信公众号直接留言，我会尽快回复您。