使用共享內存(內存映射)實現(xiàn)發(fā)布訂閱模式

• 多進程實現(xiàn)PubSub發(fā)布訂閱模式，從而實現(xiàn)進程間的通信。
• 通信方式可以是TCP/UDP，管道Pipe/消息隊列，共享內存shared memory等等。其中TCP/UDP的方式是可以用作局域網以及跨平臺的通信，Pipe/消息隊列是進程間基于系統(tǒng)實現(xiàn)比較基礎的通信，這兩者有大量優(yōu)秀的第三方庫支持，如ZeroMQ，只要加入我們自定義數(shù)據(jù)的轉換方式即可方便實現(xiàn)；而共享內存是實現(xiàn)進程間通信最快的方式，但因為共享內存的設計并不是用來做類似PubSub這種模式的實現(xiàn)的，并且共享內存實質上就是一段進程間共享的內存空間，使用自由度是極高的，所以也很少有第三方庫來實現(xiàn)共享內存方式的進程間通信。
• 因此本文的重點是如何使用共享內存shared memory來實現(xiàn)高效的PubSub發(fā)布訂閱模式。

需求

• 消息通過事先分配好的共享內存空間來傳遞
• 需要有一定的機制來管理消息的發(fā)送（寫）和接收（讀）
• 需要實現(xiàn)發(fā)布訂閱模式，也就是一個發(fā)布者（一寫）多個訂閱者（多讀）
• 考慮到平臺的原因，最后采用了文件映射內存的這種方式，在各種系統(tǒng)中都有比較通用的實現(xiàn)

邏輯分析

• 顯然，只要創(chuàng)建了一個文件并且設置好需要的大小，即可以使用mmap映射到進程的內存空間，并且在退出時可以用munmap將映射釋放掉。但是空間真正的釋放是要把文件刪掉的，因此我們需要一個計數(shù)器來記錄使用這塊共享內存的進程數(shù)，類似共享指針shared_ptr的實現(xiàn)，在計數(shù)為零時把文件刪掉。在修改這個計數(shù)的時候還需要一把進程間讀寫鎖：
• 對于只有單個訂閱者，數(shù)據(jù)之后包含一個標志位，發(fā)布者寫完后置為true，訂閱者讀完之后置為false，可能再加上一個信號燈的控制，來避免頻繁讀寫；
• 對于多個訂閱者，數(shù)據(jù)中的這個標志位變成一個計數(shù)，發(fā)布者寫完之后將計數(shù)器置為訂閱者的數(shù)量，訂閱者讀完之后將計數(shù)器減1，再加上一個進程條件變量的控制，來避免頻繁讀寫。
• 這兩種方案都有一定的弊端，最大的問題在于，訂閱者還需要修改共享內存的內容，這樣就發(fā)揮不出讀寫鎖支持多讀的優(yōu)勢了。我們需要一個更好的機制。
• 一個簡單的實現(xiàn)是數(shù)據(jù)中帶有一個單調遞增的標簽，訂閱者讀到數(shù)據(jù)后本地保存一下這個標簽的值，如果下次讀到的這個值不比保存的值大，就認為讀到了舊數(shù)據(jù)，忽略之。這個標簽比較好的實現(xiàn)是用當前的系統(tǒng)時間而不是計數(shù)，因為發(fā)布者可能會重啟清零，就算重啟后可以從已經寫入的數(shù)據(jù)中讀取，但后面為了實現(xiàn)無鎖隊列會讓這個事情變得麻煩。這樣還有一個問題是，依然會頻繁地去讀取這個標簽。因此需要加入進程條件變量的控制來減少這種頻繁。接下來是2，實現(xiàn)消息發(fā)送（寫）和接收（讀）的管理。因為我們已經有了一把讀寫鎖，很自然地想到可以用它來管理讀寫啊。事實上并不是這樣，因為發(fā)布者寫完數(shù)據(jù)之后可能會有一段時間不會占有寫鎖，這時候就要一種機制來限制訂閱者不會重復來讀這個數(shù)據(jù)。對于這個實現(xiàn)，已有的方案有：
• 對于每一個訂閱者都開辟一塊共享內存，可以按一對一的方式同時復制多份數(shù)據(jù)；
• 使用生產消費模式，使用循環(huán)隊列來實現(xiàn)讀寫分離。
• 第1種方案是解決了讀寫鎖爭搶的問題，但是增加了內存復制的開銷，反而沒有第2種方案好。但是我們要稍微修改一下傳統(tǒng)的生產消費模式的實現(xiàn)，只用一個指針來指向最新的數(shù)據(jù)。之所以這樣做是因為內存是事先分配好的，我們把它改造成環(huán)形的內存緩沖區(qū)，很難保證數(shù)據(jù)讀取的序列性；再者就是循環(huán)的尾指針應該由訂閱者自己來維護，因為每個訂閱者處理的速度是不一樣的。
• 如此一來，所有數(shù)據(jù)的修改完全是由發(fā)布者來做的，也就是說對于訂閱者來說，這是個無鎖隊列：

代碼實現(xiàn)

#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; }

參考鏈接

• 共享內存一寫多讀無鎖實現(xiàn)
• 共享內存消息隊列
• 【轉載】同步和互斥的POSIX支持(互斥鎖，條件變量，自旋鎖)
• Linux線程-互斥鎖pthread_mutex_t
• C語言open()函數(shù)：打開文件函數(shù)
• C語言mmap()函數(shù)：建立內存映射

總結

以上是生活随笔為你收集整理的完善博文 共享内存一写多读无锁实现的代码逻辑部分的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。