程序笔记

C++使用boost实现进程间通讯：消息队列，共享内存，信号

2024-06-29 42

前言

进程间通信（IPC）必然是C++程序员必掌握技能之一，而boost库是众多库中平台支持性很好，效果很高之一。做嵌入式或者服务器等应用的人肯定有所涉及。本文列举几种使用boost的常用IPC：共享内存，信号，以及消息队列的编程方式，供大家参考。

信号通信

信号是进程内通信，非常类似于Qt的信号槽，配合消息队列以及boost多线程使用效果很好，贴出一个常用带参数的写法，详细的其他用法可以参考文章末的参考文章：

#include boost/signal.hpp 
#include boost/thread/thread.hpp 
#include boost/date_time/posix_time/posix_time.hpp 
#include iostream 
 
using namespace std; 
using namespace boost; 
 
float print_sum(float x, float y) 
{ 
  std::cout  The sum is   x+y  std::endl; 
  return x+y; 
}
  
float print_product(float x, float y) 
{ 
  std::cout  The product is   x*y  std::endl; 
  return x*y; 
}
 
float print_difference(float x, float y) 
{ 
  std::cout  The difference is   x-y  std::endl; 
  return x-y; 
}
 
float print_quotient(float x, float y) 
{
   std::cout  The quotient is   x/y  std::endl; 
  return x/y; 
}
 
int main() 
{
  boost::signalfloat (float , float ) sig; 
  sig.connect(0, print_sum); 
  sig.connect(1, print_product); 
  sig.connect(2, print_difference); 
  sig.connect(3, print_quotient); 
  // Output 1.6667 because return by the last slot called. 
  cout  sig(5, 3)  endl;    
  return 0; 
}

信号槽删除及阻塞

Seg 1: Disconnecting slots.
 
        boost::signals::connection c = sig.connect(HelloWorld());
 
        if (c.connected()) {
 
        // c is still connected to the signal
 
        sig(); // Prints Hello, World!
 
        }
 
        c.disconnect(); // Disconnect the HelloWorld object
 
        assert(!c.connected()); //c isnt connected any more
 
        sig(); // Does nothing: there are no connected slots
 
Seg 2:
 
        boost::signals::connection c = sig.connect(HelloWorld());
 
        sig(); // Prints Hello, World!
 
 
 
        c.block(); // block the slot
 
        assert(c.blocked());
 
        sig(); // No output: the slot is blocked
 
 
 
        c.unblock(); // unblock the slot
 
        sig(); // Prints Hello, World!

共享内存

#include boost/interprocess/shared_memory_object.hpp  
#include boost/interprocess/mapped_region.hpp  
#include cstring  
#include cstdlib  
#include string  
   
int main(int argc, char *argv[])  
{  
   using namespace boost::interprocess;  
   
   if(argc == 1){  //Parent process  
      //Remove shared memory on construction and destruction  
      struct shm_remove  
      {  
         shm_remove() { shared_memory_object::remove(MySharedMemory); }  
         ~shm_remove(){ shared_memory_object::remove(MySharedMemory); }  
      } remover;  
   
      //Create a shared memory object.  
      shared_memory_object shm (create_only, MySharedMemory, read_write);  
   
      //Set size  
      shm.truncate(1000);  
   
      //Map the whole shared memory in this process  
      mapped_region region(shm, read_write);  
   
      //Write all the memory to 1  
      std::memset(region.get_address(), 1, region.get_size());  
   
      //Launch child process  
      std::string s(argv[0]); s +=  child ;  
      if(0 != std::system(s.c_str()))  
         return 1;  
   }  
   else{  
      //Open already created shared memory object.  
      shared_memory_object shm (open_only, MySharedMemory, read_only);  
   
      //Map the whole shared memory in this process  
      mapped_region region(shm, read_only);  
   
      //Check that memory was initialized to 1  
      char *mem = static_castchar*(region.get_address());  
      for(std::size_t i = 0; i  region.get_size(); ++i)  
         if(*mem++ != 1)  
            return 1;   //Error checking memory  
   }  
   return 0;  
}

消息队列

消息队列发送：

#include boost/interprocess/ipc/message_queue.hpp
#include iostream
#include vector
 
using namespace boost::interprocess;
 
int main ()
{
   try{
      //Erase previous message queue
      message_queue::remove(message_queue);
 
      //Create a message_queue.
      message_queue mq
         (create_only               //only create
         ,message_queue           //name
         ,100                       //max message number
         ,sizeof(int)               //max message size
         );
 
      //Send 100 numbers
      for(int i = 0; i  100; ++i){
         mq.send(i, sizeof(i), 0);
      }
 
 
   }
   catch(interprocess_exception ex){
      std::cout  ex.what()  std::endl;
      return 1;
   }
 
   return 0;
}

消息队列接收：

#include boost/interprocess/ipc/message_queue.hpp
#include iostream
#include vector
 
using namespace boost::interprocess;
 
int main ()
{
   try{
      //Open a message queue.
      message_queue mq
         (open_only        //only create
         ,message_queue  //name
         );
 
      unsigned int priority;
      message_queue::size_type recvd_size;
 
      //Receive 100 numbers
      for(int i = 0; i  100; ++i)
	  {
         int number;
         mq.receive(number, sizeof(number), recvd_size, priority);
		 printf(I:%d Rec:%d\n,i,number);
         if(number != i || recvd_size != sizeof(number))
            return 1;
      }
   }
   catch(interprocess_exception ex){
      message_queue::remove(message_queue);
      std::cout  ex.what()  std::endl;
      return 1;
   }
   message_queue::remove(message_queue);
   return 0;
}

消息队列封装

将消息队列整理了一下，可以直接方便使用如下：
注（BoostMsg.h）用于将函数外部化，无其他用处。

#include BoostMsg.h
 
#include boost/interprocess/ipc/message_queue.hpp  
#include iostream  
#include vector  
 
using namespace boost::interprocess;  
 
int BoostMsg_CreateMsg(const char* MsgQueName,int MaxQueLen,int MaxMsgLen)
{
	message_queue::remove(MsgQueName);  
	//Create a message_queue.  
	message_queue mq  
		(create_only               //only create  
		,MsgQueName						//name  
		,MaxQueLen                       //max message number  
		,MaxMsgLen              //max message size  
		);  
 
	printf(Boost Msg Created Name:%s QueLen:%d MsgMaxLen:%d \n,MsgQueName,MaxQueLen,MaxMsgLen);
 
	return 0;
}
 
int BoostMsg_OpenMsgToSend(char* BufMsg,const char* MsgQueName,int MsgLen)
{  
	try{  
		//Erase previous message queue  
		//message_queue::remove(message_queue);  
		//Create a message_queue.  
		message_queue mq  
			(open_only               //only create  
			,MsgQueName						//name  
			);  
 
		mq.send(BufMsg, MsgLen, 0);  
	}  
	catch(interprocess_exception ex){  
		std::cout  ex.what()  std::endl;  
		return 1;  
	}  
	return 0;  
} 
 
 
 
int BoostMsg_OpenMsgToRec(char* RecivedMsg,const char* MsgQueName,int MaxReceivedLen)
{  
	try{  
		//Open a message queue.  
		message_queue mq  
			(open_only        //only create  
			,MsgQueName  //name  
			);  
 
		unsigned int priority;  
		message_queue::size_type recvd_size;  
 
		memset(RecivedMsg,0,MaxReceivedLen);
		mq.receive(RecivedMsg, MaxReceivedLen, recvd_size, priority);  
		return recvd_size;
 
	}  
	catch(interprocess_exception ex){  
		message_queue::remove(message_queue);  
		std::cout  ex.what()  std::endl;  
		return 1;  
	}  
	message_queue::remove(message_queue);  
	return 0;  
}

使用方法很简单，就是Create 然后Send 和Rec，进程之间通信很方便。

编译命令：

[root@localhost tmp]# g++ boost_queue_send.cpp -o queue_send -lboost_thread -lboost_system
[root@localhost tmp]# g++ boost_queue_rec.cpp -o queue_rec -lboost_thread -lboost_system