使用共享内存(内存映射)实现发布订阅模式

多进程实现PubSub发布订阅模式，从而实现进程间的通信。
通信方式可以是TCP/UDP，管道Pipe/消息队列，共享内存shared memory等等。其中TCP/UDP的方式是可以用作局域网以及跨平台的通信，Pipe/消息队列是进程间基于系统实现比较基础的通信，这两者有大量优秀的第三方库支持，如ZeroMQ，只要加入我们自定义数据的转换方式即可方便实现；而共享内存是实现进程间通信最快的方式，但因为共享内存的设计并不是用来做类似PubSub这种模式的实现的，并且共享内存实质上就是一段进程间共享的内存空间，使用自由度是极高的，所以也很少有第三方库来实现共享内存方式的进程间通信。
因此本文的重点是如何使用共享内存shared memory来实现高效的PubSub发布订阅模式。

需求

消息通过事先分配好的共享内存空间来传递
需要有一定的机制来管理消息的发送（写）和接收（读）
需要实现发布订阅模式，也就是一个发布者（一写）多个订阅者（多读）
考虑到平台的原因，最后采用了文件映射内存的这种方式，在各种系统中都有比较通用的实现

逻辑分析

显然，只要创建了一个文件并且设置好需要的大小，即可以使用mmap映射到进程的内存空间，并且在退出时可以用munmap将映射释放掉。但是空间真正的释放是要把文件删掉的，因此我们需要一个计数器来记录使用这块共享内存的进程数，类似共享指针shared_ptr的实现，在计数为零时把文件删掉。在修改这个计数的时候还需要一把进程间读写锁：
对于只有单个订阅者，数据之后包含一个标志位，发布者写完后置为true，订阅者读完之后置为false，可能再加上一个信号灯的控制，来避免频繁读写；
对于多个订阅者，数据中的这个标志位变成一个计数，发布者写完之后将计数器置为订阅者的数量，订阅者读完之后将计数器减1，再加上一个进程条件变量的控制，来避免频繁读写。
这两种方案都有一定的弊端，最大的问题在于，订阅者还需要修改共享内存的内容，这样就发挥不出读写锁支持多读的优势了。我们需要一个更好的机制。
一个简单的实现是数据中带有一个单调递增的标签，订阅者读到数据后本地保存一下这个标签的值，如果下次读到的这个值不比保存的值大，就认为读到了旧数据，忽略之。这个标签比较好的实现是用当前的系统时间而不是计数，因为发布者可能会重启清零，就算重启后可以从已经写入的数据中读取，但后面为了实现无锁队列会让这个事情变得麻烦。这样还有一个问题是，依然会频繁地去读取这个标签。因此需要加入进程条件变量的控制来减少这种频繁。接下来是2，实现消息发送（写）和接收（读）的管理。因为我们已经有了一把读写锁，很自然地想到可以用它来管理读写啊。事实上并不是这样，因为发布者写完数据之后可能会有一段时间不会占有写锁，这时候就要一种机制来限制订阅者不会重复来读这个数据。对于这个实现，已有的方案有：
对于每一个订阅者都开辟一块共享内存，可以按一对一的方式同时复制多份数据；
使用生产消费模式，使用循环队列来实现读写分离。
第1种方案是解决了读写锁争抢的问题，但是增加了内存复制的开销，反而没有第2种方案好。但是我们要稍微修改一下传统的生产消费模式的实现，只用一个指针来指向最新的数据。之所以这样做是因为内存是事先分配好的，我们把它改造成环形的内存缓冲区，很难保证数据读取的序列性；再者就是循环的尾指针应该由订阅者自己来维护，因为每个订阅者处理的速度是不一样的。
如此一来，所有数据的修改完全是由发布者来做的，也就是说对于订阅者来说，这是个无锁队列：

代码实现

#include <iostream>
#include <cstring>
#include <vector>
#include <functional>
#include <memory>
#include <sys/mman.h>
#include <atomic>
#include <thread>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/types.h>
#include <fcntl.h>struct ShmData{bool written_;long timestamp_;size_t size_;char data_[1];ShmData():written_(false){}void Write(const char *data,const size_t len){written_ = false;memcpy(data_,data,len);size_ = len;timestamp_ = GetTimestamp();written_ = true;}bool Read(std::vector<char>* data,long* time = nullptr){if (!written_){return false;}if (time){*time = timestamp_;}data->resize(size_);memcpy(data->data(),data_,size_);return true;}static long GetTimestamp(){struct timespec ts;clock_gettime(CLOCK_REALTIME,&ts);return ts.tv_sec * 1000000 + ts.tv_nsec / 1000;}
};struct ShmQueue{size_t size_;int count_;int head_;char data_[1];ShmQueue(const size_t size,const int count):size_(sizeof(ShmData) + size),count_(count),head_(0){new(data_)ShmData;}void Write(const char* data,const size_t len){const int next = (head_ + 1) % count_;(reinterpret_cast<ShmData *>(data_ + next * size_))->Write(data,len);head_ = next;}bool Read(std::vector<char>*data,long* time){return (reinterpret_cast<ShmData *>(data_ + head_ * size_))->Read(data,time);}
};struct ShmSlice{int attached_;pthread_rwlock_t rwlock_;pthread_mutex_t mutex_;pthread_cond_t cond_;char data_[1];ShmSlice(const size_t size,const int count,const bool init = false){if (init){//init rwlockpthread_rwlockattr_t rwattr;pthread_rwlockattr_init(&rwattr);pthread_rwlockattr_setpshared(&rwattr,PTHREAD_PROCESS_SHARED);pthread_rwlock_init(&rwlock_,&rwattr);//init mutexpthread_mutexattr_t mattr;pthread_mutexattr_init(&mattr);pthread_mutexattr_setpshared(&mattr,PTHREAD_PROCESS_SHARED);pthread_mutex_init(&mutex_,&mattr);//init condition variablepthread_condattr_t cattr;pthread_condattr_init(&cattr);pthread_condattr_setpshared(&cattr,PTHREAD_PROCESS_SHARED);pthread_cond_init(&cond_,&cattr);//init shm queuenew(data_)ShmQueue(size,count);}LockWrite();if (init){attached_ = 1;} else{++attached_;}UnLockWrite();}~ShmSlice(){LockWrite();UnLockWrite();if (0 == attached_){pthread_cond_destroy(&cond_);pthread_mutex_destroy(&mutex_);pthread_rwlock_destroy(&rwlock_);}}int count(){LockRead();const int count = attached_;UnlockRead();return count;}void Write(const char* data,const size_t len){LockWrite();(reinterpret_cast<ShmQueue*>(data_))->Write(data,len);UnLockWrite();}bool Read(std::vector<char> *data,long* time){return (reinterpret_cast<ShmQueue *>(data_))->Read(data,time);}void LockWrite(){pthread_rwlock_wrlock(&rwlock_);}void UnLockWrite(){pthread_rwlock_unlock(&rwlock_);}void LockRead(){pthread_rwlock_rdlock(&rwlock_);}void UnlockRead(){pthread_rwlock_unlock(&rwlock_);}void LockMutex(){while (EOWNERDEAD == pthread_mutex_lock(&mutex_)){UnlockMutex();}}void UnlockMutex(){pthread_mutex_unlock(&mutex_);}void NotifyOne(){pthread_cond_signal(&cond_);}void NotifyAll(){pthread_cond_broadcast(&cond_);}void wait(){LockMutex();pthread_cond_wait(&cond_,&mutex_);UnlockMutex();}bool WaitFor(struct timespec *ts,const std::function<bool()>&cond){if (cond && cond()){return true;}LockMutex();pthread_cond_timedwait(&cond_,&mutex_,ts);UnlockMutex();bool ret;if (cond){ret = cond();} else{struct timespec now;clock_gettime(CLOCK_REALTIME,&now);ret = now.tv_sec < ts->tv_sec || (now.tv_sec == ts->tv_sec && now.tv_nsec <= ts->tv_nsec);}return ret;}
};class ShmManger{
public:ShmManger(std::string file_name,const int size): name_(std::move(file_name)),size_(sizeof(ShmSlice) + sizeof(ShmQueue) + 3 * (sizeof(ShmData) + size)){bool init = false;//open file descriptorint fd = open(name_.c_str(),O_RDWR | O_CREAT | O_EXCL,0600);if(fd < 0){fd = open(name_.c_str(),O_RDWR,0600);}else{//set file sizestruct stat fs;fstat(fd,&fs);if (fs.st_size < 1){ftruncate(fd,size_);}init = true;}//mmapvoid *shmaddr = mmap(NULL,size_,PROT_READ | PROT_WRITE,MAP_SHARED,fd,0);new (shmaddr) ShmSlice(size,3,init);auto deleter = [](ShmSlice *ptr){ptr->~ShmSlice();};slice_ = std::shared_ptr<ShmSlice>(reinterpret_cast<ShmSlice *>(shmaddr),deleter);close(fd);}~ShmManger(){running_ = false;slice_->NotifyAll();if (read_thread_.joinable()){read_thread_.join();}const int count = slice_->count();auto ptr = slice_.get();slice_.reset();if(count > 1){//unmapmunmap(ptr,size_);} else{//remove fileremove(name_.c_str());}}void Publish(const std::vector<char> &data){slice_->Write(data.data(),data.size());slice_->NotifyAll();}void Subscribe(std::function<void (const std::vector<char>&)>callback){callback = std::move(callback);running_ = true;read_thread_ = std::thread(&ShmManger::ReadThread,this);}private:void ReadThread(){long read_time = 0;while (running_){std::vector<char> data;long time;struct timespec ts;clock_gettime(CLOCK_REALTIME,&ts);ts.tv_sec += 5;if (!slice_->WaitFor(&ts,[&]{return slice_->Read(&data,&time) && time > read_time;})){continue;}read_time = time;//deal with datacallback_(data);}}std::string name_;int size_;std::shared_ptr<ShmSlice>slice_;std::function<void(const std::vector<char>&)>callback_;std::atomic_bool running_;std::thread read_thread_;
};
int main() {std::cout << "Hello, World!" << std::endl;return 0;
}

参考链接

共享内存一写多读无锁实现
共享内存消息队列
【转载】同步和互斥的POSIX支持(互斥锁，条件变量，自旋锁)
Linux线程-互斥锁pthread_mutex_t
C语言open()函数：打开文件函数
C语言mmap()函数：建立内存映射

完善博文共享内存一写多读无锁实现的代码逻辑部分相关推荐

java锁（公平锁和非公平锁、可重入锁(又名递归锁)、自旋锁、独占锁(写)/共享锁(读)/互斥锁、读写锁）
前言本文对Java的一些锁的概念和实现做个整理,涉及:公平锁和非公平锁.可重入锁(又名递归锁).自旋锁.独占锁(写)/共享锁(读)/互斥锁.读写锁公平锁和非公平锁概念公平锁是指多个线程按照申请 ...
java 共享内存获取写权限_跨进程访问共享内存的权限问题
问:我在服务器上用 CreateFileMapping 创建了一段共享内存.让这个exe始终在服务器上跑. 同时,别的用户在客户端用IE访问服务器,将要查询的数据通过C#制作的网页提交上来,服务器得到 ...
无盘服务器内存回写速度,解决无盘回写压力方法杂谈！
无盘回写压力缓解,乱写的,不对的地方请跟贴! 那么从文章开始到现在,我们已经取得了很多非常有用的数据,那现在让我们来总结下根据这些数据我们都可以得出哪些结论: 在测试的一周中,所有客户机平均每天的回写 ...
进程间通信IPC之--共享内存
每个进程各自有不同的用户地址空间,任何一个进程的全局变量在另一个进程中都看不到,所以进程之间要交换数据必须通过内核,在内核中开辟一块缓冲区,进程1把数据从用户空间拷到内核缓冲区,进程2再从内核缓冲 ...
Linux进程间的通信-----＞共享内存
共享内存: 顾名思义,共享内存就是允许两个不相关的进程访问同一个逻辑内存.共享内存是在两个正在运行的进程之间共享和传递数据的一种非常有效的方式.不同进程之间共享的内存通常安排为同一段物 ...
linux进程间通信：POSIX 共享内存
文章目录思维导图通信原理优势 POSIX 共享内存编程接口编程案例思维导图之前学习过sysemV 的共享内存的实现及使用原理,参考linux进程间通信:system V 共享内存 POS ...
linux进程间通讯-共享内存
文章目录共享内存概述共享内存的特点共享内存操作在ubuntu 12.04中共享内存限制值如下获得一个共享存储标识符创建共享内存共享内存映射(attach) 解除共享内存映射(detach ...
【Linux 应用编程】进程管理 - 进程间通信IPC之共享内存 mmap
IPC(InterProcess Communication,进程间通信)是进程中的重要概念.Linux 进程之间常用的通信方式有: 文件:简单,低效,需要代码控制同步管道:使用简单,默认阻塞匿名 ...
linux存储--共享内存机制shm（十四）
共享存储允许两个或多个进程共享一个给定的存储区,是进程间通信最快的一种方式. 不要同时对共享存储空间进行写操作,通常,信号量用于同步共享存储访问. 最简单的共享内存的使用流程 ①ftok函数生成键值 ...

完善博文共享内存一写多读无锁实现的代码逻辑部分

使用共享内存(内存映射)实现发布订阅模式

需求

逻辑分析

参考链接

完善博文共享内存一写多读无锁实现的代码逻辑部分相关推荐

最新文章

热门文章

完善博文 共享内存一写多读无锁实现的代码逻辑部分

使用共享内存(内存映射)实现发布订阅模式

需求

逻辑分析

参考链接

完善博文 共享内存一写多读无锁实现的代码逻辑部分相关推荐

最新文章

热门文章

完善博文共享内存一写多读无锁实现的代码逻辑部分

完善博文共享内存一写多读无锁实现的代码逻辑部分相关推荐