快捷導(dǎo)航

C++簡(jiǎn)單實(shí)現(xiàn)消息隊(duì)列的示例代碼

更新時(shí)間：2025年11月07日 10:06:29 作者：兩片空白

消息隊(duì)列是一種應(yīng)用間的通訊方式,消息發(fā)送后可以立即放回,由消息系統(tǒng)來(lái)確保消息的可靠傳遞,本文主要介紹了C++簡(jiǎn)單實(shí)現(xiàn)消息隊(duì)列的示例代碼,感興趣的可以了解一下

簡(jiǎn)介

消息隊(duì)列是一種應(yīng)用間的通訊方式，消息發(fā)送后可以立即放回，由消息系統(tǒng)來(lái)確保消息的可靠傳遞。消息發(fā)布者只需要將消息發(fā)布到消息隊(duì)列中，而不需要管誰(shuí)來(lái)取。消息使用者只管從消息隊(duì)列中取消息而不管誰(shuí)發(fā)布的。這樣發(fā)布者和使用者都不同知道對(duì)方的存在。
消息隊(duì)列普遍使用在生產(chǎn)者和消費(fèi)者模型中。

優(yōu)點(diǎn)

應(yīng)用解耦：應(yīng)用之間不用那么多的同步調(diào)用，發(fā)消息到消息隊(duì)列就行，消費(fèi)者可以自己消費(fèi)，消費(fèi)生產(chǎn)者不用管了，降低應(yīng)用之間的耦合。
降低延時(shí)：應(yīng)用之間用同步調(diào)用，需要等待對(duì)方響應(yīng)，等待時(shí)間比較長(zhǎng)，用消息之后，發(fā)送消息到消息隊(duì)列就行，應(yīng)用就可以返回了，對(duì)客戶(hù)來(lái)講降低了應(yīng)用延時(shí)。
削峰填谷：請(qǐng)求比較多的時(shí)候，應(yīng)用處理不過(guò)來(lái)，會(huì)丟棄請(qǐng)求；請(qǐng)求比較少時(shí)，應(yīng)用不飽和。
請(qǐng)求比較多時(shí)，把請(qǐng)求放到消息隊(duì)列，消費(fèi)者按特定處理速度來(lái)處理，請(qǐng)求少時(shí)，也讓?xiě)?yīng)用有事情可以做；能做到忙時(shí)不丟請(qǐng)求，閑時(shí)不閑置應(yīng)用資源。

主流的消息隊(duì)列：Kafka、ActiveMQ、RabbitMQ、RocketMQ

下面使用C++實(shí)現(xiàn)一個(gè)簡(jiǎn)單的消息隊(duì)列。

具體實(shí)現(xiàn)

消息隊(duì)列類(lèi)
里面主要包含一個(gè)數(shù)組和一個(gè)隊(duì)列，都保存消息。
應(yīng)用從數(shù)組中拿消息處理，當(dāng)數(shù)組滿(mǎn)的時(shí)候，消息保存到隊(duì)列中，當(dāng)數(shù)組中消息處理完，從隊(duì)列中取消息，再處理。
數(shù)組實(shí)現(xiàn)的是一個(gè)環(huán)形數(shù)組，記錄下消息寫(xiě)游標(biāo)和讀游標(biāo)，超過(guò)數(shù)組大小，對(duì)數(shù)組大小取余。

//消息長(zhǎng)度和消息
typedef std::pair<size_t, char*> msgPair;
#define MsgQueueSize 102400

class zMsgQueue
{
  //消息對(duì)，first表示是否存放消息
  typedef std::pair<bool, msgPair> msgQueue;
public:
  zMsgQueue();
  ~zMsgQueue();
  void* msgMalloc(const size_t len);
  void msgFree(void* p);
  //獲得一個(gè)消息
  msgPair* get();
  //放入一個(gè)消息
  bool put(const void* msg, size_t msgLen);
  //將隊(duì)列中的消息放到消息數(shù)組中
  bool putMsgQueue2Arr();
  //刪除一個(gè)消息
  void erase();
  bool empty();

  bool msgArrEmpty();
private:
  void clear();
  


  // 保存正在處理的消息
  msgQueue msgArr_[MsgQueueSize];
  // 保存等待處理的消息
  std::queue<msgPair> msgQueue_;
  
  //消息寫(xiě)游標(biāo)
  size_t queueWrite_;
  //消息讀游標(biāo)
  size_t queueRead_;
};

實(shí)現(xiàn)：

zMsgQueue::zMsgQueue()
{
  bzero(msgArr_, sizeof(msgArr_));
  queueWrite_ = 0;
  queueRead_ = 0;
}


zMsgQueue::~zMsgQueue()
{
  clear();
}

void* zMsgQueue::msgMalloc(const size_t len)
{
  char* p = (char*)malloc(len + 1);
  return (void*)(p + 1);
}

void zMsgQueue::msgFree(void* p)
{
  free((char*)p - 1);
}

  //獲得一個(gè)消息
msgPair* zMsgQueue::get()
{
  if(queueRead_ >= MsgQueueSize)
    return NULL;

  if(msgArrEmpty())
    putMsgQueue2Arr();
  
  msgPair* ret = NULL;
  if(msgArr_[queueRead_].first)
    ret = &msgArr_[queueRead_].second;

  return ret;
}

//放入一個(gè)消息
bool zMsgQueue::put(const void* msg, size_t msgLen)
{
  char* buf = (char*)msgMalloc(msgLen);
  if(buf)
  {
    bcopy(msg, buf, msgLen);
    
    //先將隊(duì)列中的消息放到數(shù)組中
    //數(shù)組中還有位置直接放到數(shù)組中
    //沒(méi)有位置放到隊(duì)列中
    if(!putMsgQueue2Arr() && !msgArr_[queueWrite_].first)
    {
      msgArr_[queueWrite_].first = true;
      msgArr_[queueWrite_].second.first = msgLen;
      msgArr_[queueWrite_].second.second = buf;
      queueWrite_++;
      queueWrite_ %= MsgQueueSize;
    }
    else 
    {
      msgQueue_.push(std::make_pair(msgLen, buf));
    }
    return true;
  } 
  return false;
}

//將隊(duì)列中的消息放到消息數(shù)組中
bool zMsgQueue::putMsgQueue2Arr()
{
  bool isLeft = false;
  while(!msgQueue_.empty())
  {
    if(!msgArr_[queueWrite_].first)
    {
      msgArr_[queueWrite_].first = true;
      msgArr_[queueWrite_].second = msgQueue_.front();
      queueWrite_++;
      queueWrite_ %= MsgQueueSize;
      msgQueue_.pop();
    }
    else 
    {
      isLeft = true;
      break;
    }
  }
  return isLeft;
}

//刪除一個(gè)消息
void zMsgQueue::erase()
{
  if(!msgArr_[queueRead_].first)
    return;

  msgFree(msgArr_[queueRead_].second.second);
  msgArr_[queueRead_].second.second = NULL;
  msgArr_[queueRead_].second.first = 0;
  msgArr_[queueRead_].first = false;

  queueRead_++;
  queueRead_ %= MsgQueueSize;
}

void zMsgQueue::clear()
{
  //隊(duì)列中還有消息
  while(putMsgQueue2Arr())
  {
    //數(shù)組中還有消息
    while(get())
    {
      erase();
    }
  }
  //數(shù)組中還有消息
  while(get())
  {
    erase();
  }
}

bool zMsgQueue::empty()
{
  if(putMsgQueue2Arr()) return false;

  return msgArrEmpty();
}

bool zMsgQueue::msgArrEmpty()
{
  if(queueRead_ == queueWrite_ && !msgArr_[queueRead_].first)
  {
    return true;
  }
  return false;
}

消息隊(duì)列的封裝
對(duì)消息隊(duì)列的封裝主要是為了對(duì)消息進(jìn)行解析和處理。
消息解析和處理函數(shù)定義成了虛函數(shù)，當(dāng)需要使用消息隊(duì)列并處理消息時(shí)，只需要繼承消息隊(duì)列，然后重寫(xiě)虛函數(shù)，進(jìn)行對(duì)應(yīng)處理即可。
類(lèi)中還使用到了讀寫(xiě)鎖，當(dāng)多線(xiàn)程的情況下，消息隊(duì)列是一個(gè)臨界資源，線(xiàn)程共享，需要進(jìn)行上鎖。單線(xiàn)程的情況下不需要加鎖。

//T表示使用的消息隊(duì)列
//msgT表示消息的類(lèi)型，有的需要消息頭，消息正文等，需要解析，這里是直接使用
template<class T=zMsgQueue, class msgT=char>
class messageQueue : public rwLocker
{
public:
  messageQueue()
  {}
  ~messageQueue()
  {}

  bool putMsg(const msgT* msg, const size_t msgLen)
  {
    rwLocker::wlock();
    msgQueue_.put(msg, msgLen);
    rwLocker::unlock();
    return true;
  }
  //解析消息，處理消息
  virtual bool msgParse(const msgT* msg, const size_t msgLen) = 0;
  //獲取消息，解析消息，處理消息
  bool doCmd()
  {
    rwLocker::wlock();
    msgPair* msg = msgQueue_.get();
    while(msg)
    {
      msgParse(msg->second, msg->first);  
      msgQueue_.erase();
      msg = msgQueue_.get();
    }
    rwLocker::unlock();
    return true;
  }

  bool empty()
  {
    return msgQueue_.empty();
  }

private:
  T msgQueue_;
};

讀寫(xiě)鎖的封裝
讀寫(xiě)鎖：可以多個(gè)線(xiàn)程進(jìn)行讀，只能一個(gè)線(xiàn)程進(jìn)行寫(xiě)。寫(xiě)時(shí)獨(dú)享資源，讀時(shí)共享資源。寫(xiě)鎖的優(yōu)先級(jí)高。
為什么讀寫(xiě)鎖需要讀鎖？

為了防止其他線(xiàn)程請(qǐng)求寫(xiě)鎖。一個(gè)線(xiàn)程請(qǐng)求了讀鎖，其他線(xiàn)程在請(qǐng)求寫(xiě)鎖會(huì)阻塞，但是請(qǐng)求讀鎖不會(huì)阻塞。一個(gè)線(xiàn)程請(qǐng)求了寫(xiě)鎖，其他線(xiàn)程請(qǐng)求讀鎖和寫(xiě)鎖都會(huì)阻塞。

#include <pthread.h>

class rwLock 
{
public:
  rwLock()
  {
    pthread_rwlock_init(&rwlc_, NULL);
  }

  ~rwLock()
  {
    pthread_rwlock_destroy(&rwlc_);
  }

  void rlock()
  {
    pthread_rwlock_rdlock(&rwlc_);
  }

  void wlock()
  {
    pthread_rwlock_wrlock(&rwlc_);
  }

  void unlock()
  {
    pthread_rwlock_unlock(&rwlc_);
  }

private:
    pthread_rwlock_t rwlc_;
};


class rwLocker
{
public:
  void rlock()
  {
    rwlc_.rlock();
  }

  void wlock()
  {
    rwlc_.wlock();
  }

  void unlock()
  {
    rwlc_.unlock();
  }

private:
  rwLock rwlc_;
};

Makefile：

# ini1=main.cpp 
# in2=messageQueue.cpp
out=main 
cc=g++
std=-std=c++11 -lpthread

#$(out):$(in1) $(in2)
$(out): main.cpp messageQueue.cpp rwlock.h
	$(cc) $^ -o $@ $(std)


.PHONY:clean
clean:
	rm -rf $(out)

代碼測(cè)試
實(shí)現(xiàn)一個(gè)類(lèi)繼承消息隊(duì)列，重寫(xiě)消息處理函數(shù)。
定義對(duì)象，調(diào)用doCmd函數(shù)即可。

#include "messageQueue.h"

class test : public messageQueue<>
{
  bool msgParse(const char* msg, const size_t msgLen)
  {
    std::cout << msgLen << ":" << msg << std::endl;
    return true;
  }
};

int main()
{
  //模擬客戶(hù)端發(fā)送消息
  char buf[256] = "hello world!";
  test t;
  //消息隊(duì)列放消息
  t.putMsg(buf, strlen(buf));
  //處理消息
  t.doCmd();
  return 0;
}