message queue with boost and serialization

기존의 예제들은 완전한 예제가 아닌 것으로 보인다.

C++ boost message queue는 POSIX 기반의 message queue를 지원하지 않는다고 한다.

또한 message queue로 메세지를 전달 시에 기본 데이터 타입이 아닌 경우에는 Serialization의 과정이 필요하다고 한다

완전한 예제인지는 모르겠지만 기존 예제를 boost에 가장 근접하게 수정을 해 보았다.

따라서 기존 예제는 좀 억지스러운 면이 있는 것 같지만 사용환경에 따라서 사용 가능할 것으로 보인다.

메세지 큐는 Inter-process보다는 Intra-process에서 자사 제품에서는 많이 사용되고 있다.  

1. 소스트리

.

├── CMakeLists.txt

├── remove.sh

└── test.cpp

2. 구현 예제

Intra-process 기반으로 작성을 해 보았다. Inter-process에서 사용할지 의문이다.

#include <iostream> #include <string> #include <thread> #include <cstdio> #include <sstream> #include <boost/archive/text_oarchive.hpp> #include <boost/archive/text_iarchive.hpp> #include <boost/archive/binary_oarchive.hpp> #include <boost/archive/binary_iarchive.hpp> #include <boost/interprocess/ipc/message_queue.hpp> #include <boost/serialization/string.hpp> using namespace boost::archive; using namespace boost::interprocess; #define MAX_SIZE 1000 class Message { public : Message(std::string aname, int aage) : name(aname), age(aage) {}; void print(void) { std::cout << "name : " << name << ", "; std::cout << "age : " << age << std::endl; }; private : friend class boost::serialization::access; template <typename Archive> void serialize(Archive &ar, const unsigned int version) { ar &name; ar &age; std::cout << __FUNCTION__ << ", " << __LINE__ << std::endl; } std::string name; int age; }; int main(int argc, char* argv[]) { int i = 0; char ch; Message smessage("heesoon.kim", 41); Message rmessage("", 0); std::string msgQName = "test"; //Erase previous message queue message_queue::remove(msgQName.c_str()); //Create a message_queue. message_queue mqSend //(open_or_create //create or open (create_only //create only ,msgQName.c_str() //name ,1 //max message number //,sizeof(smessage) //max message size ,MAX_SIZE ); message_queue mqReceive (open_only //only create ,msgQName.c_str() //name ); try { // task sending message std::thread tr_send = std::thread([&]() { while(i++ < 10) { std::stringstream oss; boost::archive::text_oarchive oa(oss); oa << smessage; // serilize smessage to oss buffer std::string serialized_string(oss.str()); // send message by message queue with string buffer mqSend.send(serialized_string.data(), serialized_string.size(), 0); } }); // task receiving message from above task std::thread tr_receive = std::thread([&]() { unsigned int priority; message_queue::size_type recvd_size; while(1) { std::stringstream iss; std::string serialized_string; serialized_string.resize(MAX_SIZE); mqReceive.receive(&serialized_string[0], MAX_SIZE, recvd_size, priority); serialized_string.resize(recvd_size); iss << serialized_string; // save received serialized data to streamstring buffer boost::archive::text_iarchive ia(iss); ia >> rmessage; // de-serialization received data rmessage.print(); } }); tr_send.detach(); tr_receive.detach(); } catch(interprocess_exception &ex){ std::cerr << ex.what() << std::endl; return 1; } while(1) { std::cout << "In main thread" << std::endl; std::cin.get(ch); if(ch == 'x' || ch == 'X') break; } message_queue::remove(msgQName.c_str()); return 0; }

3. 실행 결과

hskim@hskim-VirtualBox:~/share/prj/c++/test1-2$ ./test

In main thread serialize, 37 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 serialize, 37 name : heesoon.kim, age : 41 serialize, 37 name : heesoon.kim, age : 41

4. 전체 소스 코드

serialization.zip

https://github.com/heesoon/mpeg/tree/master/test/messageQueueWithSerialization