使用共享內存(內存映射)實現發布訂閱模式
- 多進程實現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()函數:建立內存映射