多级缓冲体系的构建（基于PSS)

freeeyes

由于业务需要，最近开始实现多级缓冲的数据体系。
因为是多台服务器的集群，所以，一开始考虑的是使用memcached。但是实际上，测试结果不是很尽如人意。这本身并不是memcached的问题，而是业务场景的问题。
我的业务是：
每时每刻都有成千上万的GPS定位数据到达服务器，用于分析，比如，比较是否进出各种区域（圆形，矩形和多边形），是否超速，是否按照路线行驶，疲劳驾驶时间，累计驾驶时间，休息提醒，油量累计，里程限制等等，这一个个的逻辑都需要处理，并且在某些条件触发的时候，及时通知下行的GPS终端做出事件响应。
如果一开始只有几十辆车，这个没什么问题。但是如果上到几十万辆，数据的处理速度就会凸显出问题。每秒到达的定位信息都在1万条以上，要对海量的数据对比分析，是一件很繁重而且复杂的工作。这里还涉及到了用户动态添加和删除指定的规则。服务器如何达到更快的相应程度？
前一段时间写了一篇二级缓冲的文章，就是源于解决此问题。
那么让我们以这个实际案例来看看，我们需要达到什么：
(1)达到数据加载最快，用户设置的新的规则会在最快的时间刷入各个服务器。
(2)必须能够准确反映各个事件的处理情况，数据加载时间尽可能的短暂。
(3)按照号段(手机号段)负载均衡，分压前端集群的数据到达处理压力。
(4)24*7的支持，必须做到在业务进程崩溃的情况下，最快速的恢复服务。
(5)达到每台服务器处理每秒至少1万条数据。
那么综合以上的问题,我会想:
(1)首先数据的加载必须分级别，什么是最重要的事件需要优先处理？
(2)分布式的服务器体系，按照某一个号段（手机号段）进行分割，不同的数据号段命中不同的业务服务器。
(3)采用什么方式能在进程宕机的时候，最快速的恢复服务？
我先看一下我的一个数据块，也就是一个手机号对应的数据规则有多大？
一个手机号，也就是一个终端的规则事件，需要75K的内存大小，看看还是不小的，主要这么大的内存规则是线路关键点，区域个数等等。一辆车最多可以设置50条线路，1000个区域。
这部分不对于我是重点，我一开始考虑memcached的实现方案，因为我考虑我的服务器都是内网服务器，我可以上千兆网卡，考虑到如果我需要达成每秒10000条数据的处理，那么，我至少需要750M每秒的带宽，但是，要命的是，我不可能让memcached吃满我的网络IO，毕竟，我还需要很多其他数据流入流出的IO带宽，而且，就算后面用memcached集群，在高密度命中的模式下，memcached的最大问题是IO，而不是命中。
需求分析的关键之一，就是要有能力把握实际业务场景，效率来源于实际的场景，而不是某些简单的测试。
经过仔细的分析，我觉得memcached方案并不能满足实际我的业务场景需求。
其实仔细想一下，我的业务也很简单，因为我的数据来源是固定的，那么我可以提前加载好数据来源，在本地共享内存中加载。找不到的，再去IO去找。
那么，共享内存本身没什么问题。之所以使用它，是我考虑一个watch的组件（具体可以参看我之前的一篇关于二级缓存的文章）。我需要定时去获得在IO(这里可能是数据库，也可能是文件，或者其他第三方的IO接口)的数据变化，同时，以最小的同步代价更新到现有的共享内存中去，第二个，watch提供的最重要的功能就是IO定向。实际业务逻辑进程只和共享内存打交道。找不到的，再去向watch要，watch拿到数据后放入共享内存并通知业务进程去读取。
这样做的好处是，完全剥离了数据源和逻辑之间的关系。
逻辑只负责和共享内存打交道，这一层封装好的话，给外面的接口，业务开发者都可以不用知道共享内存的存在，只当做一个单件来操作即可了。
这里还有一个要点，就是共享内存如果不够大，不够放入我所有的号段信息数据怎么办？要知道，我开发的平台，是给省级的车辆GPS管理服务的。我不可能让几百万的车辆数据全部流入服务器，这是不公平的，从数据的线性流量来看，在一段时间内车辆上线的情况都是稳定的，比如白天跑的车晚上一般不跑，跑夜班的车白天一般不跑，这是有规律的。那么也就是说，数据是有钝化的。在某一时刻，实时在线的车辆实际就那么十几万辆。我完全可以借助淘汰算法，达到共享内存的最优配置。
什么是淘汰算法？
如果我的服务器在启动的时候，watch先将最活跃，也就是最后使用过的车辆信息，加载到共享内存中，比如最后活跃的100000量车的事件数据。加载到共享内存，当有一个不在共享内存中的车到达的时候，我应该删除最后最不常用的车辆数据。加载当前最新的车辆数据。
这就是一个简单的LRU算法问题。
但是，在这样的交换过程中，什么是影响效能的主要杀手？
需求分析的关键之一，了解自己设计的架构在什么阶段会有瓶颈，并根据需求决定瓶颈是否是需要优先解决的。
毕竟，如果在命中不到我的共享内存的时候，我至少需要支付四次IO成本，顺序是，请求到达watch， watch到达第三方数据源并获得数据，数据源返回数据到watch，watch通知业务去获得。
IO的时间成本是最难预计的，因为你不知道你的IO会在多久返回。但是这部分成本又是必须的，考虑到综合情况，在服务器硬件正常的时候，IO流量是一个稳定的值，当这个值超过一定数量的时候才会引起时间波动，那么，掌握这个度就好了。控制watch的命中量。如果你的数据在一上线的时候大量的命中了watch，只能说明你的共享内存watch初始化策略是一个及其失败的作品。
忽略了IO成本，那么剩下的就是，如果watch成功替换了共享内存中的一个数据块，那么我如何以最小的代价通知业务逻辑进程更新它所在的索引？
我不可能让watch更新了一块共享内存的时候，业务机重新加载所有的共享内存节点，这样的消耗是我承受不起的。
那么我就需要一个与共享内存数据块和业务进程数据块链表的一个对应关系，当某一块数据发生改变，业务进程可以最快速的定位替换当前数据块的状态。
举个例子来说。
如果我在程序启动的时候，首先缓冲了100000个数据的共享内存大小。那么此时此刻，这100000个数据都不在命中状态。当业务进程启动，会有大量的数据请求流入，业务进程会根据这些请求，标记其中10000个车辆的数据已经正在使用。那么剩下的90000个数据实际是空闲的。当有一辆全新的车辆手机号到达的时候，我去这90000条数据里命中，如果命中了，则标记成使用中，如果没有命中，请求watch去数据源查找，并替换90000个数据中最后最不常用的数据。
这样，我就能实现最大程度的数据减压，尽量把90%的数据请求命中在本地缓冲中。就算随机出现一批车辆临时上线，我也能保证我的数据在最大可能的情况下减少访问IO的过程。
另一个问题是，如果数据源更新了，我如何通知业务共享内存更新？
这个工作也可以watch来做，因为我的实际情况是，用户是通过web来更新添加修改删除一些事件的。
我可以选择两种做法
(1)实时更新，前端页面发现用户修改了数据，盗用socket一个端口让watch去加载一条数据。
(2)定时更新，定时遍历一些数据的改变。
实际我取的是后者，毕竟，前端页面去写一个socket会比较耗时，不能增加前端工程师们的压力，先以达到目标为准。
需求分析的关键之一，什么是你现阶段可以舍弃的，什么是你必须保留的，必须清楚明确。
如果完全想做到完美，是不可能的，必须历经一次次的迭代升级，逐步优化。所以最怕的就是停而不做。只存在于空想，想的很美，和做的很好是两回事。
好了，以上面的计划，我开始实现一个可以满足上述需求的小例子，并测试性能是否能达到我的目标。
共享内存的基础调用不在这里讨论，有兴趣的话可以看我以前关于共享内存使用的帖子。
由于涉及到一些公司机密，所以实际逻辑判断我不在这里给出，只给出一个可以测试的需求用例（以用户名和密码验证为准）。
以下代码，在PSS作为一个测试插件提供，windows7和linux下测试通过。（详细代码可以从PSS的google code上获得）
首先，我要声明一个基类，这里明确了我要做的事情。

1. //抽象Cache层的管理
2. //抽象公共层
3. //add by freeeyes
5. #include "SMOption.h"
7. class CCacheManager
8. {
9. public:
10.         CCacheManager() {};
12.         ~CCacheManager() {};
14.         //***********************************************
15.         //进程第一次打开缓冲
16.         //u4CachedCount : 缓冲块的个数
17.         //objMemorykey  ：共享内存key
18.         //u4CheckSize   ：单块缓冲大小
19.         //***********************************************
20.         bool Init(uint32 u4CachedCount, key_t objMemorykey, uint32 u4CheckSize)
21.         {
22.                 bool blIsCreate = true;
23.                 bool blIsOpen   = true;
25.                 Set_Cache_Count(u4CachedCount);
27.                 blIsOpen = m_SMOption.Init(objMemorykey, u4CheckSize, (uint16)Get_Cache_Count(), blIsCreate);
28.                 if(false == blIsOpen)
29.                 {
30.                         return false;
31.                 }
33.                 if(blIsCreate == true)
34.                 {
35.                         //共享内存第一次创建，需要从文件里面重建共享内存
36.                         Read_All_Init_DataResoure();
37.                 }
38.                 else
39.                 {
40.                         //共享内存已经存在，遍历获得列表
41.                         Read_All_From_CacheMemory();
42.                 }
44.                 return true;
45.         };
48.         //***********************************************
49.         //关闭缓冲管理器，需要继承子类去实现之
50.         //***********************************************        
51.         virtual void Close() {};
53.         //***********************************************
54.         //定时同步数据源和缓冲中的数据，需要继承子类去实现之
55.         //***********************************************        
56.         virtual void Sync_DataReaource_To_Memory() {};
58. protected:
59.         //***********************************************
60.         //当缓冲不存在，第一次初始化缓冲，需要继承子类去实现之
61.         //***********************************************        
62.         virtual bool Read_All_Init_DataResoure() { return true; };
64.         //***********************************************
65.         //当缓冲存在，从缓冲中还原对应关系，需要继承子类去实现之
66.         //***********************************************        
67.         virtual bool Read_All_From_CacheMemory() { return true; };
69.         //***********************************************
70.         //开始同步的一些初始化动作
71.         //***********************************************
72.         virtual void Begin_Sync_DataReaource_To_Memory() {};
74.         //***********************************************
75.         //同步完成后的一些动作
76.         //***********************************************
77.         virtual void End_Sync_DataReaource_To_Memory() {};
79.         //***********************************************
80.         //设置缓冲个数
81.         //***********************************************
82.         void Set_Cache_Count(uint32 u4CacheCount)
83.         {
84.                 m_u4Count = u4CacheCount;
85.         };
87.         //***********************************************
88.         //得到当前缓冲个数
89.         //***********************************************
90.         uint32 Get_Cache_Count()
91.         {
92.                 return m_u4Count;
93.         };
95.         //***********************************************
96.         //得到指定位置的缓冲数据
97.         //***********************************************
98.         _CacheBlock* Get_CacheBlock_By_Index(uint32 u4Index)
99.         {
100.                 return (_CacheBlock* )m_SMOption.GetBuffer(u4Index);
101.         };
103.         //***********************************************
104.         //设置缓冲区全部加载成功标记位
105.         //***********************************************
106.         void Set_Memory_Init_Success()                       
107.         {
108.                 m_SMOption.SetMemoryState(READERINITSTATED);
109.         };
111. private:
112.         CSMOption    m_SMOption;
113.         uint32       m_u4Count;
115. };

复制代码

这里列出了我需要做的几个关键步骤。
对于watch，它需要同步数据事件，数据源加载事件，第一次启动数据加载事件。
这部分的事件需要自己去实现处理，根据你的数据规则。
还有一些小东西是可以封装的，比如lru淘汰算法模块，这部分完全可以作为通用化剥离出来。

1. //用于LRU的算法类
2. //共享内存数据淘汰规则
3. template<class K>
4. class CCachedLRUList
5. {
6. public:
7.         CCachedLRUList()
8.         {
9.                 m_u4CheckIndex     = 0;
10.                 m_u4MaxCachedCount = 0;
11.         };
13.         ~CCachedLRUList()
14.         {
15.         };
17.         void Set_Lru_Max_Count(uint32 u4MaxCachedCount)
18.         {
19.                 m_u4MaxCachedCount = u4MaxCachedCount;
20.         }
22.         //检测是否需要开始LRU，如果需要则返回最后要删除的
23.         bool Check_Cached_Lru(K& lrukey)
24.         {
25.                 if(m_mapKey.size() < m_u4MaxCachedCount)
26.                 {
27.                         return false;
28.                 }
29.                 else
30.                 {
31.                         //找到Lru应该弃掉的key
32.                         lrukey = (K&)m_mapTimeStamp.begin()->second;
33.                         return true;
34.                 }
35.         };
37.         //从队列里删除指定的key
38.         bool Delete_Cached_Lru(K lrukey)
39.         {
40.                 mapKey::iterator f = m_mapKey.find(lrukey);
41.                 if(f == m_mapKey.end())
42.                 {
43.                         return false;
44.                 }
46.                 uint32 u4Index = (uint32)f->second;
47.                 m_mapKey.erase(f);
49.                 mapTimeStamp::iterator ft = m_mapTimeStamp.find(u4Index);
50.                 if(ft == m_mapTimeStamp.end())
51.                 {
52.                         return false;
53.                 }
55.                 m_mapTimeStamp.erase(ft);
57.                 //删除Index和key之间的对应关系
58.                 mapKey2Index::iterator fi = m_mapKey2Index.find(lrukey);
59.                 if(fi != m_mapKey2Index.end())
60.                 {
61.                         uint32 u4CachedIndex = (uint32)fi->second;
63.                         //删除对应关系
64.                         m_mapKey2Index.erase(fi);
66.                         //删除另一个map
67.                         mapIndex2Key::iterator fii = m_mapIndex2Key.find(u4CachedIndex);
68.                         if(fii != m_mapIndex2Key.end())
69.                         {
70.                                 m_mapIndex2Key.erase(fii);
71.                         }
72.                 }
74.                 return true;
75.         };
77.         //添加一个key，如果已经存在则提升到队列最前面去
78.         EM_LRUReturn Add_Cached_Lru(K lrukey, uint32 u4CachedIndex)
79.         {
80.                 mapKey::iterator f = m_mapKey.find(lrukey);
81.                 if(f == m_mapKey.end())
82.                 {
83.                         //如果是新key,判断是否需要进行LRU检测
84.                         if(m_mapKey.size() > m_u4MaxCachedCount)
85.                         {
86.                                 return LRU_NEED_CHECK;
87.                         }
89.                         //添加新的key
90.                         uint32 u4Index = m_u4CheckIndex++;
91.                         m_mapKey.insert(mapKey::value_type(lrukey, u4Index));
92.                         m_mapTimeStamp.insert(mapTimeStamp::value_type(u4Index, lrukey));
94.                         m_mapKey2Index.insert(mapKey2Index::value_type(lrukey, u4CachedIndex));
95.                         m_mapIndex2Key.insert(mapIndex2Key::value_type(u4CachedIndex, lrukey));
97.                         return LRU_UNNEED_CHECK;
98.                 }
99.                 else
100.                 {
101.                         //如果key已经存在，则更新key的时间戳，也就是u4Index;
102.                         uint32& u4CurrIndex = (uint32&)f->second;
104.                         //删除旧的key，添加新的key
105.                         mapTimeStamp::iterator ft = m_mapTimeStamp.find(u4CurrIndex);
106.                         m_mapTimeStamp.erase(ft);
108.                         u4CurrIndex = m_u4CheckIndex++;
109.                         m_mapTimeStamp.insert(mapTimeStamp::value_type(u4CurrIndex, lrukey));
111.                         return LRU_UNNEED_CHECK;
112.                 }
113.         }
115.         //根据实际情况更新Lru的对应index和key的列表
116.         bool Reload_Cached_IndexList(K lrukey, K& lruBeforekey, uint32 u4CachedIndex)
117.         {
118.                 //寻找之前的Index对应的key并修改之
119.                 mapIndex2Key::iterator fii = m_mapIndex2Key.find(u4CachedIndex);
120.                 if(fii == m_mapIndex2Key.end())
121.                 {
122.                         return false;
123.                 }
125.                 lruBeforekey = (K)fii->second;
127.                 mapKey2Index::iterator fi = m_mapKey2Index.find(lruBeforekey);
128.                 if(fi != m_mapKey2Index.end())
129.                 {
130.                         m_mapKey2Index.erase(fi);
132.                         //添加新的key对应关系
133.                         m_mapKey2Index.insert(mapKey2Index::value_type(lrukey, u4CachedIndex));
134.                 }
136.                 m_mapIndex2Key[u4CachedIndex] = lrukey;
138.                 //测试代码
139.                 //DisPlay_Index2Key();
140.                 //DisPlay_Key2Index();
142.                 return true;
143.         }
145.         //获得指定位置的Index对应信息
146.         bool Get_Cached_KeyByIndex(uint32 u4CachedIndex, K& lrukey)
147.         {
148.                 mapIndex2Key::iterator fii = m_mapIndex2Key.find(u4CachedIndex);
149.                 if(fii == m_mapIndex2Key.end())
150.                 {
151.                         return false;
152.                 }
153.                 else
154.                 {
155.                         lrukey = (K )fii->second;
156.                         return true;
157.                 }
158.         }
160. private:
161.         //用于测试显示数据映射内容
162.         void DisPlay_Index2Key()
163.         {
164.                 OUR_DEBUG((LM_INFO, "[DisPlay_Index2Key]*****Begin DisPlay*****\n"));
165.                 for(mapIndex2Key::iterator b = m_mapIndex2Key.begin(); b != m_mapIndex2Key.end(); b++)
166.                 {
167.                         OUR_DEBUG((LM_INFO, "[DisPlay_Index2Key]key=%s.\n", ((string)(b->second)).c_str()));
168.                 }
169.                 OUR_DEBUG((LM_INFO, "[DisPlay_Index2Key]*****End DisPlay*****\n"));
170.         }
172.         //用于测试显示数据映射内容
173.         void DisPlay_Key2Index()
174.         {
175.                 OUR_DEBUG((LM_INFO, "[DisPlay_Key2Index]*****Begin DisPlay*****\n"));
176.                 for(mapKey2Index::iterator b = m_mapKey2Index.begin(); b != m_mapKey2Index.end(); b++)
177.                 {
178.                         OUR_DEBUG((LM_INFO, "[DisPlay_Key2Index]key=%s.\n", ((string)(b->first)).c_str()));
179.                 }
180.                 OUR_DEBUG((LM_INFO, "[DisPlay_Key2Index]*****End DisPlay*****\n"));
181.         }
183. private:
184.         typedef map<K, uint32> mapKey;       //key与Version对应关系
185.         typedef map<uint32, K> mapTimeStamp; //Version和key的对应关系
187.         typedef map<K, uint32> mapKey2Index; //记录Index和key的关系
188.         typedef map<uint32, K> mapIndex2Key; //记录key和Index的关系
190.         mapKey       m_mapKey;
191.         mapTimeStamp m_mapTimeStamp;
193.         mapKey2Index m_mapKey2Index;
194.         mapIndex2Key m_mapIndex2Key;
196.         uint32       m_u4CheckIndex;
197.         uint32       m_u4MaxCachedCount;
198. };
200. #endif

复制代码

对于共享内存的数据块，我需要里面有一些共享内存标记的参数，所以，所有涉及共享内存的数据对象，都要继承_CacheBlock类

1. //用于缓冲当前状态的数据结构
2. struct _CacheBlock
3. {
4. private:
5.         EM_CheckState       m_emState;           //当前维护状态，对应EM_CheckState
6.         EM_CACHED_USE_STATE m_emIsUsed;          //在线标记，对应EM_CACHED_USE_STATE状态
7.         bool                m_blDelete;          //删除标记为，Watch进程维护这个标记，false为正在使用，true为该数据已经删除
8.         uint32              m_u4CacheIndex;      //当前数据块的块ID
10. public:
11.         _CacheBlock()
12.         {
13.                 m_emState       = CHECKS_HIT;
14.                 m_blDelete      = false;
15.                 m_emIsUsed      = CACHEDUNUSED;
16.                 m_u4CacheIndex  = 0;
17.         }
19.         //设置块ID
20.         void SetCacheIndex(uint32 u4CacheIndex)
21.         {
22.                 m_u4CacheIndex = u4CacheIndex;
23.         }
25.         //获得块ID
26.         uint32 GetCacheIndex()
27.         {
28.                 return m_u4CacheIndex;
29.         }
30.         
31.         //被数据源命中
32.         void SetHit()
33.         {
34.                 m_blDelete      = false;
35.                 m_emState       = CHECKS_HIT;
36.         }
38.         //没有被数据源命中
39.         void SetUnHit()
40.         {
41.                 m_blDelete     = true;
42.                 m_emState      = CHECKS_UNHIT;
43.         }
45.         //设置删除状态
46.         void SetDelete(bool blDelete)
47.         {
48.                 m_blDelete = blDelete;
49.         }
51.         //设置共享内存没有被逻辑进程使用
52.         void SetUsed()
53.         {
54.                 m_emIsUsed = CACHEDUSED;
55.         }
57.         //设置共享内存没有被逻辑进程使用
58.         void SetUnUsed()
59.         {
60.                 m_emIsUsed = CACHEDUNUSED;
61.         }
63.         //得到删除状态
64.         bool GetDelete()
65.         {
66.                 return m_blDelete;
67.         }
69.         //得到是否命中状态
70.         EM_CheckState GetCheckState()
71.         {
72.                 return m_emState;
73.         }
75.         //设置命中状态
76.         void SetCheckState(EM_CheckState objCheckState)
77.         {
78.                 m_emState = objCheckState;
79.         }
80. };

复制代码

比如，我的数据对象是一个简单的用户名和密码验证的数据结构。
包含了用户名和密码
那么我的结构是：

1. //这里如果是想用缓冲功能，必须继承_CacheBlock对象
2. struct _UserValid : public _CacheBlock
3. {
4.         //缓冲的数据结构
5.         char   m_szUserName[MAX_BUFF_50];   //用户名
6.         char   m_szUserPass[MAX_BUFF_50];   //密码
7.         uint32 m_u4LoginCount;              //登陆次数
8. };

复制代码

watch每隔60秒更新一下缓冲中正在使用的数据。

1. ACE_thread_t  threadId;
2.         ACE_hthread_t threadHandle;
4.         //初始化共享内存
5.         App_UserValidManager::instance()->Init((uint32)MAX_LOGIN_VALID_COUNT, SHM_USERVALID_KEY, (uint32)sizeof(_UserValid));
7.         //首先创建工作线程
8.         ACE_Thread::spawn(
9.                 (ACE_THR_FUNC)worker,        //线程执行函数
10.                 NULL,                        //执行函数参数
11.                 THR_JOINABLE | THR_NEW_LWP,
12.                 &threadId,
13.                 &threadHandle
14.                 );

复制代码

worker的实际代码为：

1. void* worker(void *arg)
2. {
3.         if(NULL != arg)
4.         {
5.                 OUR_DEBUG((LM_INFO, "[worker]have param.\n"));
6.         }
7.        
8.         while(true)
9.         {
10.                 OUR_DEBUG((LM_INFO, "[Watch]Valid Begin.\n"));
11.                 App_UserValidManager::instance()->Sync_DataReaource_To_Memory();
12.                 OUR_DEBUG((LM_INFO, "[Watch]Valid End.\n"));
13.                 App_UserValidManager::instance()->Display();
14.                 ACE_Time_Value tvSleep(60, 0);
15.                 ACE_OS::sleep(tvSleep);
16.         }
18.         return NULL;
19. }

复制代码

我会开启一个socket，让逻辑进程可以通过这个接口获得它所需要的数据。具体代码太多了，就在这里贴一个大概吧。

1. virtual int handle_input (ACE_HANDLE fd )
2.         {
3.                 if(fd == ACE_INVALID_HANDLE)
4.                 {
5.                         OUR_DEBUG((LM_ERROR, "[handle_input]fd is ACE_INVALID_HANDLE.\n"));
6.                         return -1;
7.                 }
8.                
9.                 ACE_Time_Value nowait(0, MAX_QUEUE_TIMEOUT);
11.                 //处理接收逻辑
12.                 //接收字节，先接收4字节包长度，然后接收用户名字符串
13.                 char szPacketSize[4] = {'\0'};
14.                 int nDataLen = this->peer().recv(szPacketSize, 4, MSG_NOSIGNAL, &nowait);
15.                 if(nDataLen != 4 && nDataLen <= 0)
16.                 {
17.                         return -1;
18.                 }
20.                 int nPacketSize = 0;
21.                 ACE_OS::memcpy(&nPacketSize, szPacketSize, 4);
23.                 char* pBuff = new char[nPacketSize];
24.                 ACE_OS::memset(pBuff, 0, nPacketSize);
26.                 //因为是内网程序，目前不考虑分包和粘包，这里在大流量下可以优化的，先以完成功能为主。
27.                 //组包规则，2字节用户名长度+用户名+4字节ConnectID
28.                 nDataLen = this->peer().recv(pBuff, nPacketSize, MSG_NOSIGNAL, &nowait);
29.                 if(nDataLen != nPacketSize && nDataLen <= 0)
30.                 {
31.                         SAFE_DELETE_ARRAY(pBuff);
32.                         return -1;
33.                 }
35.                 //解析数据
36.                 int nRecvPos      = 0;
37.                 int nUserNameSize = 0;
38.                 int nUserPassSize = 0;
39.                 int nConnectID    = 0;
41.                 ACE_OS::memcpy((char* )&nUserNameSize, (char* )&pBuff[nRecvPos], 2);
42.                 nRecvPos += 2;
43.                 char* pUserName = new char[nUserNameSize + 1];
44.                 ACE_OS::memset(pUserName, 0, nUserNameSize + 1);
45.                 ACE_OS::memcpy((char* )pUserName, (char* )&pBuff[nRecvPos], nUserNameSize);
46.                 nRecvPos += nUserNameSize;
48.                 ACE_OS::memcpy((char* )&nUserPassSize, (char* )&pBuff[nRecvPos], 2);
49.                 nRecvPos += 2;
50.                 char* pUserPass = new char[nUserPassSize + 1];
51.                 ACE_OS::memset(pUserPass, 0, nUserPassSize + 1);
52.                 ACE_OS::memcpy((char* )pUserPass, (char* )&pBuff[nRecvPos], nUserPassSize);
53.                 nRecvPos += nUserPassSize;
55.                 ACE_OS::memcpy((char* )&nConnectID, (char* )&pBuff[nRecvPos], 4);
56.                 nRecvPos += 4;
58.                 int nSendSize = 4 + 2 + nUserNameSize + 2 + nUserPassSize + 1 + 4 + 4;
59.                 char* pSend = new char[nSendSize];
60.                 int nSendPos = 0;
62.                 //处理接收数据
63.                 uint32 u4CacheIndex = 0;
64.                 bool blState = App_UserValidManager::instance()->Load_From_DataResouce(pUserName, u4CacheIndex);
65.                 if(blState == false)
66.                 {
67.                         //没有找到这个用户数据，组成返回包
68.                         int nSendPacketSize = 2 + nUserNameSize + 2 + nUserPassSize + 1 + 4 + 4;
69.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nSendPacketSize, 4);
70.                         nSendPos += 4;
71.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nUserNameSize, 2);
72.                         nSendPos += 2;
73.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )pUserName, nUserNameSize);
74.                         nSendPos += nUserNameSize;
75.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nUserPassSize, 2);
76.                         nSendPos += 2;
77.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )pUserPass, nUserPassSize);
78.                         nSendPos += nUserPassSize;
79.                         int nRet = 1;
80.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nRet, 1);
81.                         nSendPos += 1;
82.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&u4CacheIndex, 4);
83.                         nSendPos += 4;
84.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nConnectID, 4);
85.                         nSendPos += 4;
86.                 }
87.                 else
88.                 {
89.                         //找到了这个用户数据，组成返回包
90.                         int nSendPacketSize = 2 + nUserNameSize + 2 + nUserPassSize + 1 + 4 + 4;
91.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nSendPacketSize, 4);
92.                         nSendPos += 4;
93.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nUserNameSize, 2);
94.                         nSendPos += 2;
95.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )pUserName, nUserNameSize);
96.                         nSendPos += nUserNameSize;
97.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nUserPassSize, 2);
98.                         nSendPos += 2;
99.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )pUserPass, nUserPassSize);
100.                         nSendPos += nUserPassSize;
101.                         int nRet = 0;
102.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nRet, 1);
103.                         nSendPos += 1;
104.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&u4CacheIndex, 4);
105.                         nSendPos += 4;
106.                         ACE_OS::memcpy((char* )&pSend[nSendPos], (char* )&nConnectID, 4);
107.                         nSendPos += 4;
108.                 }
110.                 //发送返回数据
111.                 this->peer().send(pSend, nSendSize, &nowait);
113.                 SAFE_DELETE_ARRAY(pUserPass);
114.                 SAFE_DELETE_ARRAY(pUserName);
115.                 SAFE_DELETE_ARRAY(pSend);
116.                 SAFE_DELETE_ARRAY(pBuff);
117.                 return 0;
118.         }

复制代码

PSS的插件作为业务逻辑进程，watch作为监控者，达成。
实际测试结果，在10个用户缓冲下，每秒到达10000个随机请求，其中10%是新的ID，这样的情况CPU在30%左右，数据IO稳定。watch每秒处理800个左右新的请求（剩下200个被缓冲LRU挡住，非常好）。实际效果Linux优于windows，达到了我的设计需求。