正文

详解boost库中的Message Queue .

IP云
IP云V管理员/6 阅读
09/12
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示
Message Queue（后文简写成MQ或消息队列）是boost库中用来封装进程间通信的一种实现，同一台机器上的进程或线程可以通过消息队列来进行通迅。消息队列中的消息由优先级、消息长度、消息数据三部分组成。这里需要注意的事，MQ只是简单的将要发送的数据在内存中进行拷贝，所以我们在发送复杂结构或对象时，我们需要将其序列化后再发送，接收端接收时要反序列化，也就是说我们要自己去定义区分一条消息（就是自定义网络通迅协议）。在MQ中，我们可以使用三模式去发送和接收消息：
  1. 阻塞：在发送消息时，若消息队列满了，那么发送接口将会阻塞直到队列没有满。在接收消息时，若队列为空，那么接收接口也会阻塞直到队列不空。
  2. 超时：用户可以自定义超时时间，在超时时间到了，那么发送接口或接收接口都会返回，无论队列满或空
  3. Try：在队列为空或满时，都能立即返回
MQ使用命名的共享内存来实现进程间通信。共享内存换句话来说，就是用户可以指定一个名称来创建一块共享内存，然后像打一个文件一样去打开这块共享内存，同样别的进程也可以根据这个名称来打开这块共享内存，这样一个进程向共享内存中写，另一个进程就可以从共享内存中读。这里两个进程的读写就涉及到同步问题。另外，在创建一个MQ时，我们需要指定MQ的最大消息数量以及消息的最大size。
  1. //Create a message_queue. If the queue
  2. //exists throws an exception
  3. message_queue mq
  4. (create_only         //only create
  5. ,"message_queue"     //name
  6. ,100                 //max message number
  7. ,100                 //max message size
  8. );
  9. using boost::interprocess;
  10. //Creates or opens a message_queue. If the queue
  11. //does not exist creates it, otherwise opens it.
  12. //Message number and size are ignored if the queue
  13. //is opened
  14. message_queue mq
  15. (open_or_create      //open or create
  16. ,"message_queue"     //name
  17. ,100                 //max message number
  18. ,100                 //max message size
  19. );
  20. using boost::interprocess;
  21. //Opens a message_queue. If the queue
  22. //does not exist throws an exception.
  23. message_queue mq
  24. (open_only           //only open
  25. ,"message_queue"     //name
  26. );
使用message_queue::remove("message_queue");来移除一个指定的消息队列。
接下来，我们看一个使用消息队列的生产者与消息者的例子。第一个进程做为生产者，第二个进程做为消费者。
生产者进程：
  1. #include <boost/interprocess/ipc/message_queue.hpp>
  2. #include <iostream>
  3. #include <vector>
  4. using namespace boost::interprocess;
  5. int main ()
  6. {
  7. try{
  8. //Erase previous message queue
  9. message_queue::remove("message_queue");
  10. //Create a message_queue.
  11. message_queue mq
  12. (create_only               //only create
  13. ,"message_queue"           //name
  14. ,100                       //max message number
  15. ,sizeof(int)               //max message size
  16. );
  17. //Send 100 numbers
  18. for(int i = 0; i < 100; ++i){
  19. mq.send(&i, sizeof(i), 0);
  20. }
  21. }
  22. catch(interprocess_exception &ex){
  23. std::cout << ex.what() << std::endl;
  24. return 1;
  25. }
  26. return 0;
  27. }

消费者进程：

  1. #include <boost/interprocess/ipc/message_queue.hpp>
  2. #include <iostream>
  3. #include <vector>
  4. using namespace boost::interprocess;
  5. int main ()
  6. {
  7. try{
  8. //Open a message queue.
  9. message_queue mq
  10. (open_only        //only create
  11. ,"message_queue"  //name
  12. );
  13. unsigned int priority;
  14. message_queue::size_type recvd_size;
  15. //Receive 100 numbers
  16. for(int i = 0; i < 100; ++i){
  17. int number;
  18. mq.receive(&number, sizeof(number), recvd_size, priority);
  19. if(number != i || recvd_size != sizeof(number))
  20. return 1;
  21. }
  22. }
  23. catch(interprocess_exception &ex){
  24. message_queue::remove("message_queue");
  25. std::cout << ex.what() << std::endl;
  26. return 1;
  27. }
  28. message_queue::remove("message_queue");
  29. return 0;
  30. }