使用内存映射开发高性能进程间消息通信组件

时间：2019-01-06 16:30:13 阅读：193 评论：0 收藏：0 [点我收藏+]

标签：err 就会 throw 客户中间取数 amp 超时 www.

一、背景

　　项目开发中免不了各模块或系统之间进行消息通信，目前热门的消息中间件有Redis、RabbitMQ、Kafka、RocketMQ等等。

以上几种组件中Redis在消息队列方面表现还可以，但是如果涉及发布订阅功能，就不行了，最近项目就使用了redis的发布订阅，

每秒只能发出几千条，虽然目前绰绰有余，但是瓶颈可以预期。

　　其余的几种都是比较重量级的消息中间件，什么跨平台、分布式、集群、支持N种协议等等，很高大全，

我们可能就只使用了其中1、2个功能。严格来说，项目中集成这几种MQ的工作量是不小的，对于中小型系统来说，可能维护MQ

稳定的工作量都比项目还大，难度也高，所有功能用全了的程序员恐怕不多。

　　从长远考虑出发，选择重量级MQ恐怕是板上钉钉的事，但是项目一开始就上这几种，我觉得那也是欠缺考虑的。如果项目

根本不要求跨机器通信，那杀鸡就不要用牛刀了。比如，你只是在模块之间、线程之间、进程之间，或者是在同一主机的各种不同系统之间，

其实都可以不用重量级MQ。当然你使用了也没事，看个人选择。

　　最近的项目有这么个场景，采集近所有底层设备，每个设备有点3000个，总共20多万个点需要采集上来。刚开始使用了Redis的发布订阅，

但是程序毫无疑问地挂了，根本带不起来；因为程序启动时每个点的值都是从0变成N，就需要发消息出来，那一开始消息是很多的，redis根本

处理不完，而且有很高频率的超时断线。以至于想换RabbitMQ，后来想想还是算了，因为那样增加项目难度不说，后期维护也是个难题。

说到底这是模块之间的通信，是主程序（Winform）调用采集C++的DLL类库，发出消息后主程序和web端订阅，在主程序与DLL这边，在DLL

方法上增加一个回调函数就搞定了，完全不用走消息中间件，Web端要哪些点的实时值就先ASK，先请求需要看哪些点，如何在主程序这边

发布那些点的实时值消息，这样发布订阅的数据量少了2、3个数量级不止。

二、需求

　　针对上边的业务场景，因为是模块之间的线程间通信，这样搞问题不大；如果是进程之间也要那么高频率的通信，那就不好办了，我们

不想使用重量级MQ，又想高频率传输消息，怎么办呢？网上搜索了一番，貌似没看到有成熟的速度又快、体量又小，部署又简单的中间件。

所以在下不才，针对这个问题抛砖引玉，做一个demo出来供大家讨论一下。

三、原理

　　应题，就是使用内存映射来做同一个机器下各种消息的通信，之前也写过一篇关于使用共享内存实现快速读写的文章，点击前往浏览

“.net环境下跨进程、高频率读写数据”，但是内存映射比较适合做消息队列，因为消息可以持久化在本地，没读完下次进来还可以接着读。

我预想是这样设计：

1、发布订阅涉及到2个主要方法：Publish(string channel)、Subscribe(string channel, Callback callback);

2、为每个channel生成一个文件：channel.db,默认每个db可以存储1000个同类型的结构体消息作为消息队列，从头部写入，尾部读出。

　每个db文件前面留一个索引区作为发布方与订阅方各自的读写位置。发布与订阅前，先读写这个索引区，因为是一对一读写，所以

可以完美避开读写锁，大大提高性能。

3、针对一对多需求，单独设计一个config.db文件存储种channel与其相关订阅信息，大概原理图如下：

技术分享图片

4、解决读写不加锁问题

我们看结构体：SIndex有三个属性

1) WriteIndex 记录发布方(Pubish)最后写入数据的位置

2) ReadIndex 记录订阅方(Subscribe)最后读取数据的位置

3) Over 表示WriteIndex已达到队列最大值，再WriteIndex小于等于队列最大值前，读写如下图：

技术分享图片

WriteIndex达到最大值后再往下写Over就要取反，如由False变为True。WriteIndex=0

如果此时没有订阅方,那新消息就会被抛弃，因为已无空间存储。

技术分享图片

4) 如果ReadIndex数值到队列最大值，Over也取反，此时ReadIndex = 0,读写又变成图1所示

5) 读写过程中并不存在互斥的情况，只要管理好读写位置，就可以避免加锁。

四、接口设计

4.1、主要参数定义

#define FM_MAX_CHANNEL		100		// 暂定最多100个不同频道
#define FM_MAX_SUBSCRIBE	3		// 暂定最多3个订阅用户
#define FM_MAX_ROWS			1000	// 暂定最多队列大小为1000
#define FM_DISCONNECT_TIME  5000	// 超过5000毫秒无心跳更新视为订阅断开
#define FM_KEEP_CONN_CYCLE  1000	// 保持心跳连接的时间周期
#define FM_NOTHING			-1		// 空白,数组为0等
#define FM_WORD_SIZE		sizeof(WORD)	// WORD长度
#define FM_INDEX_SIZE		sizeof(SIndex)  // SIndex长度

4.2、结构体

 1 // 索引
 2 typedef struct
 3 {
 4     WORD WriteIndex;
 5     WORD ReadIndex;
 6     WORD Over;         // 当W或R超过MAX一次，Over取反一次,Over默认为False
 7 }SIndex;
 8 
 9 // 内存映射参数
10 typedef struct
11 {
12     HANDLE FileHandle;
13     HANDLE FileMappingHandle;
14     LPVOID MapViewOfFileHandle;
15     UINT StructSize;
16     char FileName[20];
17     UINT SubscribeIndex;
18     WORD Conned;
19 }SDbConnInfo;
20 
21 // 频道
22 typedef struct
23 {
24     char ChannelName[20];
25     UINT StructSize;
26     DWORD Subscribe1LastTime;
27     DWORD Subscribe2LastTime;
28     DWORD Subscribe3LastTime;
29 }SChannel;
30 
31 // 频道与订阅映射
32 typedef struct
33 {
34     char ChannelName[20];
35     SDbConnInfo DbConnInfo[FM_MAX_SUBSCRIBE];
36 }SChannelMapDbConnInfo;

View Code

4.3、主要方法

	// 发布信息
	template<typename T>
	int Publish(const char *channel, T* data);

	// 订阅信息
	template<typename T>
	void Subscribe(const char *channel, SubscribeCallBackHandle callback);

五、代码实现

5.1 、FMDBManager，主要管理内存映射相关操作，因为是读写位置不一样，所以不需要加互斥量

  1 class FMDBManager
  2 {
  3 public:
  4     FMDBManager() {};
  5     ~FMDBManager() {};
  6 
  7 public:
  8     static int Create(SDbConnInfo *info)
  9     {
 10         CString fileName(info->FileName);
 11         DWORD totalSize = (FM_MAX_ROWS * info->StructSize) + FM_INDEX_SIZE;
 12 
 13         info->FileHandle = CreateFile(fileName, (GENERIC_READ | GENERIC_WRITE), (FILE_SHARE_READ | FILE_SHARE_WRITE),
 14             NULL, OPEN_ALWAYS, FILE_FLAG_SEQUENTIAL_SCAN, NULL);
 15 
 16         info->FileMappingHandle = CreateFileMapping(info->FileHandle, NULL, PAGE_READWRITE, 0, totalSize, NULL);
 17 
 18         if(info->FileMappingHandle == NULL || info->FileMappingHandle == INVALID_HANDLE_VALUE) 
 19         {
 20             Log("");
 21             CloseHandle(info->FileHandle);
 22             return enumFail;
 23         }
 24 
 25         if(GetLastError() == ERROR_ALREADY_EXISTS) 
 26         {
 27             Log("");
 28             return enumFail;
 29         }
 30 
 31         // init
 32         info->MapViewOfFileHandle = MapViewOfFile(info->FileMappingHandle, FILE_MAP_ALL_ACCESS, 0, 0, totalSize);
 33 
 34         if(info->MapViewOfFileHandle == NULL) 
 35         {
 36             Log("");
 37             CloseHandle(info->FileMappingHandle);
 38             CloseHandle(info->FileHandle);
 39             return enumFail;
 40         }
 41 
 42         return enumSuccess;
 43     }
 44 
 45 protected:
 46     int Write(void *data, UINT order, SDbConnInfo *info)
 47     {
 48         if(info->MapViewOfFileHandle == NULL) 
 49         {
 50             Log("");
 51             return enumFail;
 52         }
 53         else
 54             memcpy((char *)info->MapViewOfFileHandle + (order * info->StructSize) + FM_INDEX_SIZE, data, info->StructSize);
 55 
 56         return enumSuccess;
 57     }
 58     int Read(void *data, UINT order, SDbConnInfo *info)
 59     {
 60         if(info->MapViewOfFileHandle == NULL) 
 61         {
 62             Log("");
 63             return enumFail;
 64         }
 65         else
 66             memcpy(data, (char *)info->MapViewOfFileHandle + (order * info->StructSize) + FM_INDEX_SIZE, info->StructSize);
 67 
 68         return enumSuccess;
 69     }
 70     int Delete(UINT order, SDbConnInfo *info)
 71     {
 72         if(info->MapViewOfFileHandle == NULL) 
 73         {
 74             Log("");
 75             return enumFail;
 76         }
 77         else
 78             memset((char *)info->MapViewOfFileHandle + (order * info->StructSize) + FM_INDEX_SIZE, 0, info->StructSize);
 79 
 80         return enumSuccess;
 81     }
 82 
 83     int WriteConfig(void *data, UINT order, UINT pos, UINT size, SDbConnInfo *info)
 84     {
 85         if(info->MapViewOfFileHandle == NULL) 
 86         {
 87             Log("");
 88             return enumFail;
 89         }
 90         else
 91             memcpy((char *)info->MapViewOfFileHandle + (order * info->StructSize) + FM_INDEX_SIZE + pos, data, size);
 92 
 93         return enumSuccess;
 94     }
 95     int WriteIndex(void *data, UINT pos, UINT size, SDbConnInfo *info)
 96     {
 97         if(info->MapViewOfFileHandle == NULL) 
 98         {
 99             Log("");
100             return enumFail;
101         }
102         else
103             memcpy((char *)info->MapViewOfFileHandle + pos, data, size);
104 
105         return enumSuccess;
106     }
107     int ReadIndex(SIndex *sIndex, SDbConnInfo *info)
108     {
109         if(info->MapViewOfFileHandle == NULL) 
110         {
111             Log("");
112             return enumFail;
113         }
114         else
115             memcpy(sIndex, (char *)info->MapViewOfFileHandle, FM_INDEX_SIZE);
116 
117         return enumSuccess;
118     }
119 };

View Code

5.2、FMDBClient，内存映射客户端，主要封装Publish与Subscribe方法给前端调用，屏蔽复杂性

  1 class FMDBClient : public FMDBManager
  2 {
  3 private:
  4     mutable std::mutex mut;
  5     SChannelMapDbConnInfo channelMapDbConnInfo = { 0 };
  6 
  7     bool CanWrite(SIndex *sIndex)
  8     {
  9         int nextWriteIndex = sIndex->WriteIndex + 1;
 10         if(nextWriteIndex > FM_MAX_ROWS) nextWriteIndex = 0;
 11 
 12         return nextWriteIndex != sIndex->ReadIndex;
 13     }
 14     bool CanRead(SIndex *sIndex) {
 15         if(sIndex->Over) return sIndex->ReadIndex > sIndex->WriteIndex;
 16         else return sIndex->ReadIndex + 1 <= sIndex->WriteIndex;
 17     }
 18 
 19     int GetDbConnInfo(const char *channel, int size)
 20     {
 21         int rest = enumFail;
 22 
 23         for(int i = 0; i < FM_MAX_CHANNEL; i++)
 24         {
 25             char channelNameTmp[20] = { 0 };
 26             sprintf_s(channelNameTmp, "%s", channel);
 27 
 28             if(0 == strcmp(channelNameTmp, channelMapDbConnInfoArray[i].ChannelName))
 29             {
 30                 channelMapDbConnInfo = channelMapDbConnInfoArray[i];
 31                 rest = enumSuccess;
 32                 break;
 33             }
 34         }
 35 
 36         return rest;
 37     }
 38     int SetDbConnInfo(const char *channel, UINT *subscribeIndex, SDbConnInfo *dbConnInfo)
 39     {
 40         std::lock_guard<std::mutex> lk(mut);
 41 
 42         int nextSubscribeIndex = fmdbConfig->GetNextSubscribeIndex(channel);
 43         if(nextSubscribeIndex == FM_NOTHING)
 44         {
 45             SChannel sChannel = { 0 };
 46             sprintf_s(sChannel.ChannelName, "%s", channel);
 47             sChannel.Subscribe1LastTime = GetTickCount();
 48             sChannel.StructSize = dbConnInfo->StructSize;
 49 
 50             sprintf_s(dbConnInfo->FileName, "%s.1.db", channel);
 51             if(!fmdbConfig->IsFM_NOTHING(dbConnInfo->FileName) && dbConnInfo->StructSize > 0)
 52             {
 53                 dbConnInfo->SubscribeIndex = 1;
 54                 *subscribeIndex = dbConnInfo->SubscribeIndex;
 55 
 56                 if(Create(dbConnInfo) == enumSuccess) return fmdbConfig->Insert(&sChannel);
 57                 else sprintf_s(dbConnInfo->FileName, "%s", channel); //还原名称
 58             }
 59         }
 60 
 61         if(nextSubscribeIndex > 1)
 62         {
 63             sprintf_s(dbConnInfo->FileName, "%s.%d.db", channel, nextSubscribeIndex);
 64             if(!fmdbConfig->IsFM_NOTHING(dbConnInfo->FileName) && dbConnInfo->StructSize > 0)
 65             {
 66                 dbConnInfo->SubscribeIndex = nextSubscribeIndex;
 67                 *subscribeIndex = nextSubscribeIndex;
 68 
 69                 if(Create(dbConnInfo) == enumSuccess) return fmdbConfig->Save(channel, nextSubscribeIndex);
 70                 else sprintf_s(dbConnInfo->FileName, "%s", channel); //还原名称
 71             }
 72         }
 73 
 74         return enumFail;
 75     }
 76     bool SetSubscribeConned(const char *channel, int subscribeIndex, SDbConnInfo *dbConnInfo)
 77     {
 78         int rest = enumFail;
 79 
 80         if(subscribeIndex <= 0) return rest;
 81 
 82         for(int i = 0; i < FM_MAX_CHANNEL; i++)
 83         {
 84             char channelNameTmp[20] = { 0 };
 85             sprintf_s(channelNameTmp, "%s", channel);
 86 
 87             if(0 == strcmp(channelNameTmp, channelMapDbConnInfoArray[i].ChannelName))
 88             {
 89                 channelMapDbConnInfoArray[i].DbConnInfo[subscribeIndex - 1].SubscribeIndex = dbConnInfo->SubscribeIndex; 
 90                 channelMapDbConnInfoArray[i].DbConnInfo[subscribeIndex - 1].Conned = 1;
 91                 rest = enumSuccess;
 92                 break;
 93             }
 94         }
 95 
 96         return rest;
 97     }
 98     bool IsConning(SDbConnInfo *dbConnInfo) { return true; };
 99 
100 public:
101     FMDBClient()
102     {
103         while(!fmdbConfigLoadFinish) { Sleep(200); }
104     };
105     ~FMDBClient() {};
106 
107 public:
108     int failTimes = 0;
109 
110     template<typename T>
111     int Publish(const char *channel, T* data)
112     {
113         int rest = enumFail;
114 
115         // 查找
116         if(GetDbConnInfo(channel, sizeof(T)) == enumFail)
117         {
118             printf_s("发布%s失败.\n", channel);
119             return enumFail;
120         }
121 
122         for(int i = 0; i < FM_MAX_SUBSCRIBE; i++)
123         {
124             if(channelMapDbConnInfo.DbConnInfo[i].FileHandle == NULL) continue;
125 
126             while(IsConning(&channelMapDbConnInfo.DbConnInfo[i]))
127             {
128                 SIndex sIndex = { 0 };
129                 if(ReadIndex(&sIndex, &channelMapDbConnInfo.DbConnInfo[i]) == enumFail)
130                 {
131                     throw "映射文件加载失败";
132                 }
133 
134                 if(CanWrite(&sIndex))
135                 {
136                     WORD writeIndex = sIndex.WriteIndex;
137                     if(Write(data, writeIndex, &channelMapDbConnInfo.DbConnInfo[i]) == enumSuccess)
138                     {
139                         writeIndex++;
140                         if(writeIndex > FM_MAX_ROWS)
141                         {
142                             writeIndex = 0;
143 
144                             WORD Over = TRUE;
145                             WriteIndex(&Over, (FM_WORD_SIZE * 2), FM_WORD_SIZE, &channelMapDbConnInfo.DbConnInfo[i]);
146                         }
147 
148                         rest = WriteIndex(&writeIndex, 0, FM_WORD_SIZE, &channelMapDbConnInfo.DbConnInfo[i]);
149                         break;
150                     }
151                 }
152                 else
153                 {
154                     failTimes++;
155                 }
156             }
157         }
158 
159         return rest;
160     }
161 
162     template<typename T>
163     void Subscribe(const char *channel, SubscribeCallBackHandle callback)
164     {
165         SDbConnInfo dbConnInfo = { 0 };
166         dbConnInfo.StructSize = sizeof(T);
167 
168         UINT subscribeIndex = 0;
169         if(SetDbConnInfo(channel, &subscribeIndex, &dbConnInfo) == enumFail)
170         {
171             printf_s("订阅%s失败.\n", channel);
172             return;
173         }
174 
175         while(IsConning(&dbConnInfo))
176         {
177             SetSubscribeConned(channel, subscribeIndex, &dbConnInfo);
178 
179             SIndex sIndex = { 0 };
180             if(ReadIndex(&sIndex, &dbConnInfo) == enumFail) throw "映射文件加载失败";
181             if(!CanRead(&sIndex)) continue;
182 
183             T t = { 0 };
184             int readIndex = sIndex.ReadIndex;
185             if(Read(&t, readIndex, &dbConnInfo) == enumSuccess)
186             {
187                 readIndex++;
188                 if(readIndex > FM_MAX_ROWS)
189                 {
190                     readIndex = 0;
191 
192                     WORD Over = FALSE;
193                     WriteIndex(&Over, (FM_WORD_SIZE * 2), FM_WORD_SIZE, &dbConnInfo);
194                 }
195 
196                 if(WriteIndex(&readIndex, FM_WORD_SIZE, FM_WORD_SIZE, &dbConnInfo) == enumSuccess)
197                     if(Delete(sIndex.ReadIndex, &dbConnInfo) == enumSuccess)
198                         if(callback(&t) == enumBreak) break;
199             }
200         }
201     }
202 };

View Code

请注意上边控制读写的2个方法

	bool CanWrite(SIndex *sIndex)
	{
		int nextWriteIndex = sIndex->WriteIndex + 1;
		if(nextWriteIndex > FM_MAX_ROWS) nextWriteIndex = 0;

		return nextWriteIndex != sIndex->ReadIndex;
	}
	bool CanRead(SIndex *sIndex) 
	{
		if(sIndex->Over) return sIndex->ReadIndex > sIndex->WriteIndex;
		else return sIndex->ReadIndex + 1 <= sIndex->WriteIndex;
	}

我们可以分析一下，下一个WriteIndex值如果大于队列最大值 WriteIndex置0，下一个WriteIndex数值如果不等于

正在读的位置ReadIndex就能写；如果WriteIndex没有超出最大值，只要ReadIndex小于等于WriteIndex就能读，

如果超出，就判断ReadIndex大于WriteIndex就能读。WriteIndex与ReadIndex数值在Publish与Subscribe中维护

5.3、建立新线程获取最新订阅的客户端信息，这个功能主要是动态地像多个Subscribe端发生消息，比如订阅发生在发布之后，

也应该能收到消息。

 1 void Update()
 2 {
 3     while(true)
 4     {
 5         if(fmdbConfig->GetChannelArray() == enumSuccess)
 6         {
 7             for(int i = 0; i < FM_MAX_CHANNEL; i++)
 8             {
 9                 if(fmdbConfig->IsFM_NOTHING(channelMapDbConnInfoArray[i].ChannelName)) continue;
10 
11                 for(int j = 0; j < FM_MAX_SUBSCRIBE; j++)
12                 {
13                     if(channelMapDbConnInfoArray[i].DbConnInfo[j].StructSize <= 0) continue;
14 
15                     // KeepConned
16                     if(channelMapDbConnInfoArray[i].DbConnInfo[j].Conned)
17                     {
18                         fmdbConfig->KeepConned(channelMapDbConnInfoArray[i].ChannelName,
19                             channelMapDbConnInfoArray[i].DbConnInfo[j].SubscribeIndex);
20 
21                         channelMapDbConnInfoArray[i].DbConnInfo[j].Conned = 0;
22                         //printf_s("%s.KeepConned.\n", channelDbParsArray[i].SDbPars[j].Channel);
23                     }
24 
25                     if(!fmdbConfig->Exists(channelMapDbConnInfoArray[i].DbConnInfo[j].FileName))
26                     {
27                         FMDBManager::Create(&channelMapDbConnInfoArray[i].DbConnInfo[j]);
28                         fmdbConfig->AddChannel(channelMapDbConnInfoArray[i].DbConnInfo[j].FileName);
29                     }
30                 }
31             }
32         }
33 
34         fmdbConfigLoadFinish = true;
35         Sleep(1000);
36     }
37 }
38 thread th(Update);

View Code

六、Demo测试

6.1、Producer.cpp

 1 #include "pch.h"
 2 #include "../FMDB.h"
 3 
 4 using namespace std;
 5 
 6 int main()
 7 {
 8     FMClient * client = new FMClient();
 9 
10     int times = 0;
11     int index = 0;
12     int total = 0;
13     UINT structSize = sizeof(SPerson);
14     DWORD dwStartTmp = GetTickCount();
15 
16     while(TRUE)
17     {
18         times++;
19         if(index == 0)
20         {
21             dwStartTmp = GetTickCount();
22         }
23 
24         SPerson sPerson = { 0 };
25         sPerson.Idx = index;
26         sprintf_s(sPerson.Name, "Name.%d", index);
27         sPerson.Age = index;
28 
29         if(client->Publish("Person", &sPerson) == enumSuccess)
30         {
31             if(index % 2 == 0) total = total + sPerson.Idx;
32             else total = total - sPerson.Idx;
33 
34             index++;
35             if(index % 50000 == 0)
36                 printf_s("发送条数: %d, 耗时:%d \n", index, (GetTickCount() - dwStartTmp));
37         }
38 
39         if(index >= 2000000) break;
40     }
41 
42     printf_s("调用次数: %d, 成功条数: %d, 检验值: %d \n", times, index, total);
43     system("pause");
44 }

6.2、Consumer.cpp

 1 #include "pch.h"
 2 #include "../FMDB.h"
 3 
 4 using namespace std;
 5 
 6 int index = 0;
 7 int total = 0;
 8 DWORD dwStartTmp = GetTickCount();
 9 
10 int SubscribeCallback(void *msg)
11 {
12     SPerson * person = (SPerson *)msg;
13 
14     if(index == 0)
15     {
16         dwStartTmp = GetTickCount();
17     }
18 
19     if(index % 2 == 0) total = total + person->Idx;
20     else total = total - person->Idx;
21 
22     index++;
23     if(index % 50000 == 0)
24     {
25         printf("接收条数: %d, 耗时:%d, Idx:%d, Name:%s, Age:%d\n",
26             index, (GetTickCount() - dwStartTmp), person->Idx, person->Name, person->Age);
27     }
28 
29     if(index >= 2000000)
30     {
31         return enumBreak;
32     }
33 
34     return enumSuccess;
35 };
36 
37 int main()
38 {
39     FMClient * client = new FMClient();
40     client->Subscribe<SPerson>("Person", SubscribeCallback);
41 
42     printf("接收条数: %d, 检验值: %d \n", index, total);
43     system("pause");
44 }