성우님 블로그

#define MAX_MESSAGE_NUMBER  100
#define MAX_MESSAGE_SIZE    128
#define MAX_PROCESS_NAME    32

#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/interprocess/creation_tags.hpp>
#include <iostream>
using namespace boost::interprocess;

class CMessageQ
{

public:
  CMessageQ(const char* pName)
  {
    m_pMsgQ = NULL;
    strcpy(m_Name, pName);
  }
  ~CMessageQ(void)
  {
	  Remove();
  }

public:
  bool Create()
  {
	  try{
			Remove();
			// 메모리에서 제거되면 안되므로 static으로 선언
			static message_queue mq(create_only, m_Name, MAX_MESSAGE_NUMBER, MAX_MESSAGE_SIZE);
			m_pMsgQ = &mq;
	  }catch(interprocess_exception &ex){
		  std::cout << "MessageQ Create() error : " << ex.what() << std::endl;
		  return false;
	  }
	  return true;
  }

  bool Open()
  {
	  try{
		  static message_queue mq(open_only, m_Name);
      m_pMsgQ = &mq;
	  }catch(interprocess_exception &ex){
		  std::cout << "MessageQ Open() error : " << ex.what() << std::endl;
		  return false;
	  }
	  return true;
  }

  size_t Get()
  { 
    size_t temp;
    try{
    temp = m_pMsgQ->get_num_msg();
    return temp;
    }catch(interprocess_exception &ex){
      std::cout << "MessageQ Get() error : " << ex.what() << std::endl;
    }
    return 0;
  }

  template <typename T>
  bool Send(T* Data,size_t Size, unsigned int priority = 0)
  {
	  try{
		  m_pMsgQ->send(Data, Size, priority);
	  }catch(interprocess_exception &ex){
		  std::cout << "MessageQ Send() error : " << ex.what() << std::endl;
		  return false;
	  }
	  return true;
  }

  template <typename T>
  bool Recv(T* Data, size_t Size, unsigned int& priority)
  {
    size_t recvd_size=0;
	  try{
			//m_pMsgQ->receive(Data, Size, recvd_size, priority);
			m_pMsgQ->receive(Data, MAX_MESSAGE_SIZE, recvd_size, priority);
			if( Size != recvd_size )
			{
				std::cout << "MessageQ Recv() Size Wrong : " << std::endl;
				return false;
			}
	  }catch(interprocess_exception &ex){
		  std::cout << "MessageQ Recv() error : " << ex.what() << std::endl;
		  return false;
	  }
	  return true;
  }

  void Remove()
  {
	  message_queue::remove(m_Name);
  }

private:
	char m_Name[MAX_PROCESS_NAME]; // 프로세스 이름
	message_queue* m_pMsgQ; // 메시지큐 포인터
};

원문 :
http://www.boost.org/doc/libs/1_40_0/doc/html/interprocess/synchronization_mechanisms.html#interprocess.synchronization_mechanisms.message_queue


메시지큐란 무엇인가?

메시지큐는 메시지들의 목록과 유사하다. 스레드들은 메시지들을 큐에 넣을 수 있고 큐로 부터 제거할 수도 있다. 각각의 메시지들은 또한 우선순위가 낮은 메시지들보다 먼저 수행될 수 있는 높은 우선순위와 같은 우선순위를 가진다. 각각의 메시지들은 몇가지의 특성을 가지고 있다.

• 우선순위
• 메시지의 길이
• 데이터 (길이가 0보다 크다면)

스레드는 아래의 3가지 방법을 통해 메시지를 보내거나 받을 수 있다.

• Blocking : 만약 송신시에 메시지큐가 가득찼다거나 수신시에 비었다면, 스레드는 새로운 메시지를 위한 공간이나 날때까지 또는 새로운 메시지가 생길때까지 멈춰(Blocked) 있을 것이다.
• Try : 만약 송신시에 메시지큐가 가득찼다거나 수신시에 비었다면, 스레드는 그 즉시 에러를 리턴한다.
• Timed : 만약 송신시에 메시지큐가 가득찼다거나 수신시에 비었다면, 스레드는 성공신호(성공적인 상태를 리턴) 혹은 중단신호(실패를 리턴)가 도착할때까지 행동을 재시도 한다.

메시지큐는 단지 프로세스들 사이에 raw bytes를 복사하고 개체(object)를 전송하지는 않는다. 이것은 만약 우리가 메시지큐를 이용하여 개체(object)를 전송하고 싶다면, 그 개체는 반드시 binary serializable 되어야 한다. 예를 들면, 우리는 프로세스들 사이에 정수(integer)를 전송할 수 있지만 std::string은 전송 할 수 없다. 이때는 Boost.Serialization 또는 Boost.Interprocess 라는 향상된 메커니즘을 사용해서 프로세스들 사이에 복합적인 데이터(complex data)를 전송해야 한다.

Boost.Interprocess 메시지큐는 프로세스간 통신이라고 불린다. 메시지큐는 이름을 사용해서 만들어지고 파일처럼 이름으로 불러온다. 메시지큐를 생성할 때, 사용자는 반드시 메시지의 최대 사이즈와 메시지큐를 저장할 수 있는 최대 메시지 수를 정의하여야 한다. 이러한 파라미터들은 리소스들을 정의 할 것이다.(한 예로, 공유메모리가 사용되고 있다면, 공유메모리(shared memory)의 크기는 메시지큐에 제공된다)
 

using boost::interprocess;
//Create a message_queue. If the queue
//exists throws an exception
message_queue mq
   (create_only         //생성 전용(only create)
   ,"message_queue"     //이름
   ,100                 //최대 메시지 수(max message number)
   ,100                 //최대 메시지 크기(max message size)
   );

using boost::interprocess;
//Creates or opens a message_queue. If the queue
//does not exist creates it, otherwise opens it.
//Message number and size are ignored if the queue
//is opened
message_queue mq
   (open_or_create      //읽기 또는 생성(open or create)
   ,"message_queue"     //이름(name)
   ,100                 //최대 메시지 수(max message number)
   ,100                 //최대 메시지 크기(max message size)
   );

using boost::interprocess;
//Opens a message_queue. If the queue
//does not exist throws an exception.
message_queue mq
   (open_only           //읽기전용(only open)
   ,"message_queue"     //이름(name)
   );

using boost::interprocess;
message_queue::remove("message_queue");

#include <boost/interprocess/ipc/message_queue.hpp>

#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;
      std::size_t recvd_size;

      //Receive 100 numbers
      for(int i = 0; i < 100; ++i){
         int number;
         mq.receive(&number, sizeof(number), recvd_size, priority);
         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;
}