devbox
小小林熬夜学编程
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

linux下共享内存和消息队列实现多进程间数据收发_通过共享内存实现消息队列

作者：小小林熬夜学编程 | 2024-05-21 08:12:25

赞

踩

通过共享内存实现消息队列

       linux下进程通信的方式有很多，共享内存，消息队列，管道等。共享内存可以传输大量数据，但是多个进程同时读取共享内存就会出现脏读，可以借助消息队列实现多进程消息发送和接收。这种组合方式在实际开发中应用还是很多的，接下来就看一下。

目录

1.共享内存操作api

 （1）创建共享内存

（2）挂载共享内存到当前进程

（3）取消挂载

（4） 共享内存控制函数-可以删除

2.消息队列操作api

（1）创建消息附列

（2）往消息队列中发送消息

（3）从消息队列取消息

（4）操作消息队列，删除，查看等

3.命令行操作

（1）共享内存查看和删除

（2）消息队列查看和删除

（3）信号量查看和删除

4.生产者进程和消费者进程互相通信实践

4.1 代码

4.2 运行效果

5.execve系统调用

1.共享内存操作api

 （1）创建共享内存

int shmget(key_t key, size_t size, int shmflg);

（2）挂载共享内存到当前进程

void *shmat(int shmid, const void *shmaddr, int shmflg);

（3）取消挂载

int shmdt(const void *shmaddr);

（4） 共享内存控制函数-可以删除

int shmctl(int shmid, int cmd, struct shmid_ds *buf);

2.消息队列操作api

（1）创建消息附列

int msgget(key_t key, int flags);

（2）往消息队列中发送消息

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

（3）从消息队列取消息

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

（4）操作消息队列，删除，查看等

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

3.命令行操作

（1）共享内存查看和删除

ipcs -m / ipcrm -m

（2）消息队列查看和删除

 ipcs -q / ipcrm -q

（3）信号量查看和删除

ipcs -s / ipcrm -s

4.生产者进程和消费者进程互相通信实践

接下来通过一个共享内存和两条消息队列来实现进程间互发数据，本在ubuntu环境下使用vscode开发，程序目录结构如下：

4.1 代码

消息定义：  msgInfo.h

  1. #pragma once
  2. #include <time.h>
  3.  
  4. #define	gAvailableMsgBufPos	100
  5. #define gOccupiedMsgBufPos	101
  6.  
  7. //共享内存结构
  8. struct TSharedMemoryModule
  9. {
  10. 	char	        name[40+1];		
  11. 	int	            id;			
  12. 	char	        reloadCommand[80+1];
  13. 	int	            index;			
  14. 	long	        sizeOfUserSpace;	
  15. 	unsigned char   *removed_puserSpace;	
  16. 	int	            users;			
  17. 	int	            newCreated;		
  18. 	int	            removed_writingLocks;
  19. 	int	            removed_readingLocks;
  20. 	int				resID;
  21. };
  22. typedef TSharedMemoryModule		*PTSharedMemoryModule;
  23.  
  24. //消息队列管理结构
  25. struct TMsgBufHDL
  26. {
  27. 	long			maxSizeOfMsg;		
  28. 	int				maxNumOfMsg;		
  29. 	int				maxStayTime;		
  30. 	int				userID;			
  31. 	int				queueIDOfFreePos;	
  32. 	int				queueIDOfOccupiedPos;
  33. };
  34. typedef TMsgBufHDL		*PTMsgBufHDL;
  35.  
  36. //消息头
  37. struct TMessageHeader
  38. {
  39. 	long			type;	
  40. 	int				provider;	
  41. 	time_t			time;		
  42. 	int				len;		
  43. 	long			msgIndex;	
  44. 	int				dealer;		
  45.     int 			forward_cnt;
  46. 	short			locked;	
  47. };
  48. typedef TMessageHeader	*PTMessageHeader;
  49.  
  50. // 空闲消息队列元素结构
  51. struct TFreePosOfMsgBuf
  52. {
  53. 	long 	statusOfPos;
  54. 	char	indexOfPos[4];
  55. };
  56.  
  57. // 任务消息队列结构
  58. struct TOccupiedPosOfMsgBuf
  59. {
  60. 	long 	typeOfMsg;
  61. 	char	indexOfPos[4];
  62. };

收据收发头文件：msgExchange.h

  1. #pragma once
  2.  
  3. #include "msgInfo.h"
  4.  
  5. /*
  6. ================ 消息队列 ====================
  7. // 创建共享内存
  8. int shmget(key_t key, size_t size, int shmflg);
  9. // 挂载
  10. void *shmat(int shmid, const void *shmaddr, int shmflg);
  11. // 取消挂载
  12. int shmdt(const void *shmaddr);
  13. // 共享内存控制函数-可以删除
  14. int shmctl(int shmid, int cmd, struct shmid_ds *buf);
  15. */
  16.  
  17. /*
  18. ================ 消息队列 ====================
  19.  // 创建和获取 ipc 内核对象
  20.  int msgget(key_t key, int flags);
  21.  // 将消息发送到消息队列
  22.  int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);
  23.  // 从消息队列获取消息
  24.  ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);
  25.  // 查看、设置、删除 ipc 内核对象（用法和 shmctl 一样）
  26.  int msgctl(int msqid, int cmd, struct msqid_ds *buf);
  28. 消息数据格式
  29.  struct Msg{
  30.      long type; // 消息类型。这个是必须的，而且值必须 > 0，这个值被系统使用
  31.      // 消息正文，多少字节随你而定
  32.  };
  33.  */
  34.  
  35. // 初始化共享内存和消息队列
  36. int InitSharedMemoryMsgBuf();
  37.  
  38. // 销毁共享内存和消息队列
  39. void UnInitSharedMemoryMsgBuf();
  40.  
  41. //创建消息队列 - 调用创建共享内存
  42. int CreateMsgBuf();
  43.  
  44. //创建共享内存
  45. PTSharedMemoryModule CreateSharedMemoryModule(long sizeOfUserSpace);
  46.  
  47. //初始化消息队列
  48. int InitAllMsgIndex();
  49.  
  50. //从空闲消息队列取位置
  51. int GetAvailableMsgBufPos();
  52.  
  53. // 空闲消息队列放入消息
  54. int FreeMsgBufPos(int index);
  55.  
  56. // 任务消息队列放入消息
  57. int OccupyMsgBufPos(int index, long typeOfMsg);
  58.  
  59. // 任务消息队列取消息
  60. int GetOccupiedMsgBufPos(long *typeOfMsg);
  61. // 从任务消息队列获取指定类型的消息
  62. int GetOccupiedMsgBufPosOfMsgType(long typeOfMsg);
  63.  
  64. // 发送消息
  65. int WriteBufferMessage(const unsigned char * msg,
  66.                 int lenOfMsg, 
  67.                 long typeOfMsg, 
  68.                 const TMessageHeader* poriHeader, 
  69.                 PTMessageHeader pNewHeader);
  70.  
  71. // 用指定位置发消息
  72. int WriteBufferMessageUsingIndex(const unsigned char * msg, 
  73.                             int lenOfMsg, long typeOfMsg,
  74.                             const TMessageHeader* poriHeader, 
  75.                             PTMessageHeader pNewHeader, 
  76.                             const int *piIndexOfMsg);
  77.  
  78. //接收消息 - 释放索引
  79. int ReadBufferMessagReleaseIndex(unsigned char * msg, int lenOfMsgBuf,
  80.                                 long typeOfMsg,
  81.                                 PTMessageHeader pOutputHeader);
  82.  
  83. //接收消息 - 不释放索引
  84. int ReadBufferMessageNotReleaseIndex(unsigned char *msg, int lenOfMsgBuf, 
  85.                                 long typeOfMsg, 
  86.                                 PTMessageHeader pOutputHeader, 
  87.                                 int *indexOfMsg, int iSigRetryFlag);

收据收发实现：msgExchange.cpp

  1. #include <sys/types.h>
  2. #include <sys/ipc.h>
  3. #include <sys/msg.h>
  4. #include <sys/shm.h>
  5. #include <string.h>
  6. #include <errno.h>
  7. #include <sys/types.h>
  8. #include <unistd.h>
  9. #include <iostream>
  10.  
  11. #include "msgExchange.h"
  12.  
  13. // ipcs -m / ipcrm -m
  14. // ipcs -q / ipcrm -q 
  15. // ipcs -s / ipcrm -s
  16.  
  17. // 以下三部分都是从共享内存申请
  18. static PTSharedMemoryModule    s_pSharedMemMDL = NULL;
  19. static PTMsgBufHDL             s_pMsgBufHDL = NULL;
  20. static unsigned char           *s_pMsgBuf = NULL;
  21.  
  22. //共享内存，消息队列初始化信息
  23. static TSharedMemoryModule s_sharedMemoryModule;
  24. static TMsgBufHDL          s_msgBufHDL;
  25.  
  26. static unsigned long       gMsgIndex = 1;
  27.  
  28. #if defined(__GNUC__)
  29. static unsigned long addAndFetchUlongAtomic(volatile unsigned long *p_var, unsigned long value) {
  30.     return __sync_add_and_fetch(p_var, value);
  31. }
  32. #endif
  33.  
  34. // 初始化共享内存和消息队列
  35. int InitSharedMemoryMsgBuf()
  36. {
  37.     s_sharedMemoryModule.id = 1234567;
  38.     s_sharedMemoryModule.newCreated = 1;
  39.     s_sharedMemoryModule.users = 0;
  40.  
  41.     s_msgBufHDL.maxSizeOfMsg = 8192;		
  42. 	s_msgBufHDL.maxNumOfMsg = 1024;
  43.  
  44.     CreateMsgBuf();
  45.  
  46.     return 0;
  47. }
  48.  
  49. void UnInitSharedMemoryMsgBuf() {
  50.     struct shmid_ds    buf;
  51.     shmdt(s_pSharedMemMDL);
  52.     shmctl(s_sharedMemoryModule.resID, IPC_RMID, &buf);
  53.     msgctl(s_msgBufHDL.queueIDOfOccupiedPos, IPC_RMID, NULL);
  54.     msgctl(s_msgBufHDL.queueIDOfFreePos, IPC_RMID, NULL);  
  55. }
  56.  
  57. //创建共享内存
  58. PTSharedMemoryModule CreateSharedMemoryModule(long sizeOfUserSpace)
  59. {
  60.     PTSharedMemoryModule   pSharedMem = nullptr;
  61.     int                    ret;
  62.     int                    resID;
  63.     unsigned char          *pmUserSpace = nullptr;
  64.  
  65.     if ((resID = shmget(s_sharedMemoryModule.id, sizeof(TSharedMemoryModule) + sizeOfUserSpace, 0660)) == -1) {
  66.         if ((resID = shmget(s_sharedMemoryModule.id, sizeof(TSharedMemoryModule) + sizeOfUserSpace, IPC_CREAT|0660)) == -1) {
  67.             return nullptr;
  68.         }
  69.         s_sharedMemoryModule.newCreated = 1;
  70.     }
  71.     else {
  72.         s_sharedMemoryModule.newCreated = 0;
  73.     }
  74.  
  75.     std::cout << "shared memory id is :: " << resID << std::endl;
  76.     s_sharedMemoryModule.resID = resID;
  77.  
  78.     if (((pSharedMem = (PTSharedMemoryModule)shmat(resID, 0, SHM_RND)) == NULL) || (pSharedMem == (void*)(-1))) {
  79.         if (errno == 24) {
  80.             pSharedMem = (PTSharedMemoryModule)shmat(resID, 0, SHM_RND);
  81.         }
  82.         if ((pSharedMem == NULL) || (pSharedMem == (void*)(-1))) {
  83.             return nullptr;
  84.         }
  85.     }
  86.  
  87.     if (!s_sharedMemoryModule.newCreated) {
  88.         ++(s_sharedMemoryModule.users);
  89.         s_sharedMemoryModule.newCreated = 0;
  90.         pSharedMem->newCreated = s_sharedMemoryModule.newCreated;
  91.     }
  92.     else {        
  93.         memcpy(pSharedMem, &s_sharedMemoryModule, sizeof(s_sharedMemoryModule));
  94.         pSharedMem->users = 1;
  95.         pSharedMem->removed_readingLocks = 0;
  96.         pSharedMem->removed_writingLocks = 0;
  97.         pSharedMem->sizeOfUserSpace = sizeOfUserSpace;
  98.     }
  99.     if (sizeOfUserSpace == 0) {
  100.         pmUserSpace = nullptr;
  101.     }
  102.     else {
  103.         pmUserSpace = (unsigned char *)((unsigned char *)pSharedMem + sizeof(*pSharedMem));
  104.     }
  105.     
  106.     // 清空消息内容
  107.     if ((pSharedMem->newCreated) && (pmUserSpace)) {
  108.         memset(pmUserSpace, 0, sizeOfUserSpace);
  109.     }
  110.  
  111.     return pSharedMem;
  112. }
  113.  
  114. //创建消息队列 - 调用创建共享内存
  115. int CreateMsgBuf() {
  116.     TMsgBufHDL         msgBufHDL;
  117.     int                ret;
  118.     int                newCreatedQueue;
  119.     
  120.     memset(&msgBufHDL,0,sizeof(msgBufHDL));
  121.     msgBufHDL = s_msgBufHDL;
  122.     
  123.     if ((s_pSharedMemMDL = CreateSharedMemoryModule(
  124.             sizeof(*s_pMsgBufHDL) + 
  125.             ((sizeof(TMessageHeader) + sizeof(unsigned char) * msgBufHDL.maxSizeOfMsg) * (msgBufHDL.maxNumOfMsg + 1)))) == NULL)
  126.     {
  127.         return -1;
  128.     }
  129.  
  130.     if ((s_pMsgBufHDL = (PTMsgBufHDL)((unsigned char *)s_pSharedMemMDL + sizeof(TSharedMemoryModule))) == NULL)
  131.     {
  132.         return -1;
  133.     }
  134.  
  135.     s_pMsgBuf = (unsigned char *)s_pMsgBufHDL + sizeof(*s_pMsgBufHDL);
  136.     
  137.     if (s_pSharedMemMDL->newCreated) {
  138.         memcpy(s_pMsgBufHDL, &msgBufHDL, sizeof(*s_pMsgBufHDL));
  139.     }
  140.     
  141.     s_pMsgBufHDL->userID = s_pSharedMemMDL->id;
  142.  
  143.     if ((s_pMsgBufHDL->queueIDOfFreePos = msgget(s_pMsgBufHDL->userID, 0660)) == -1)
  144.     {
  145.         if ((s_pMsgBufHDL->queueIDOfFreePos = msgget(s_pMsgBufHDL->userID, 0660 | IPC_CREAT)) == -1)
  146.         {
  147.             return -1;
  148.         }
  149.         else {
  150.             newCreatedQueue = 1;
  151.         }
  152.     }
  153.  
  154.     std::cout << "free msg queue id is :: " << s_pMsgBufHDL->queueIDOfFreePos << std::endl;
  155.     s_msgBufHDL.queueIDOfFreePos =  s_pMsgBufHDL->queueIDOfFreePos;
  156.  
  157.     if ((s_pMsgBufHDL->queueIDOfOccupiedPos = msgget(s_pMsgBufHDL->userID + 1,0660)) == -1)
  158.     {
  159.         if ((s_pMsgBufHDL->queueIDOfOccupiedPos = msgget(s_pMsgBufHDL->userID + 1,0660 | IPC_CREAT)) == -1)
  160.         {
  161.             return -1;
  162.         }
  163.         else {
  164.             newCreatedQueue = 1;
  165.         }
  166.             
  167.     }
  168.     
  169.     std::cout << "occupied msg queue id is :: " << s_pMsgBufHDL->queueIDOfOccupiedPos << std::endl;
  170.     s_msgBufHDL.queueIDOfOccupiedPos =  s_pMsgBufHDL->queueIDOfOccupiedPos;
  171.  
  172.     if (newCreatedQueue) {
  173.         InitAllMsgIndex();
  174.     }
  175.     
  176.     return 0;
  177. }
  178.  
  179.  
  180. //初始化消息队列
  181. int InitAllMsgIndex() {
  182.     TFreePosOfMsgBuf        freePos;
  183.     TOccupiedPosOfMsgBuf    occupiedPos;
  184.     PTMessageHeader         pheader;
  185.     int                     ret;
  186.     int                     index;
  187.         
  188.     while(true) {
  189.         if ((ret = msgrcv(s_pMsgBufHDL->queueIDOfFreePos, &freePos, sizeof(freePos.indexOfPos), 0, IPC_NOWAIT)) < 0) {
  190.             break;
  191.         }
  192.             
  193.     }
  194.  
  195.     while(true) {
  196.         if ((ret = msgrcv(s_pMsgBufHDL->queueIDOfOccupiedPos, &occupiedPos, sizeof(occupiedPos.indexOfPos), 0, IPC_NOWAIT)) < 0) {
  197.             break;
  198.         }    
  199.     }
  200.     
  201.     // 位置消息队列中放入共享内存索引编号
  202.     for (index = 0; index < s_pMsgBufHDL->maxNumOfMsg; index++)
  203.     {
  204.         FreeMsgBufPos(index);
  205.     }
  206.     
  207.     return(0);
  208. }
  209.  
  210. //从空闲消息队列取位置
  211. int GetAvailableMsgBufPos()
  212. {
  213.     TFreePosOfMsgBuf    posOfMsgBuf;
  214.     int                 ret;
  215.     int                 index;
  216.     
  217.     memset(posOfMsgBuf.indexOfPos, 0, sizeof(posOfMsgBuf.indexOfPos));
  218.     if ((ret = msgrcv(s_pMsgBufHDL->queueIDOfFreePos, &posOfMsgBuf, sizeof(int), 0, 0)) < 0) {
  219.         return -1;
  220.     }
  221.     memcpy(&index, posOfMsgBuf.indexOfPos, sizeof(int));
  222.     
  223.     return index;
  224. }
  225.  
  226. // 空闲消息队列放入消息
  227. int FreeMsgBufPos(int index)
  228. {
  229.     TFreePosOfMsgBuf    posOfMsgBuf;
  230.     int                 ret;
  231.     PTMessageHeader    pheader;
  232.     
  233.     if ((index < 0) || (index >= s_pMsgBufHDL->maxNumOfMsg)) {
  234.         return -1;
  235.     }
  236.     
  237.     pheader = (PTMessageHeader)(s_pMsgBuf + index * (sizeof(TMessageHeader) + s_pMsgBufHDL->maxSizeOfMsg));
  238.     pheader->type = 0;
  239.     
  240.     posOfMsgBuf.statusOfPos = gAvailableMsgBufPos;
  241.     memcpy(posOfMsgBuf.indexOfPos, &index, sizeof(int));
  242.     
  243.     if (msgsnd(s_pMsgBufHDL->queueIDOfFreePos, &posOfMsgBuf, sizeof(int), 0) < 0) {
  244.         return -1;
  245.     }
  246.     return 0;
  247. }
  248.  
  249. // 任务消息队列放入消息
  250. int OccupyMsgBufPos(int index, long typeOfMsg)
  251. {
  252.     TOccupiedPosOfMsgBuf    posOfMsgBuf;
  253.     int                     ret;
  254.     
  255.     posOfMsgBuf.typeOfMsg = typeOfMsg;
  256.     memcpy(posOfMsgBuf.indexOfPos, &index, sizeof(int));
  257.  
  258.     if (ret = msgsnd(s_pMsgBufHDL->queueIDOfOccupiedPos, &posOfMsgBuf, sizeof(posOfMsgBuf.indexOfPos), 0)) {
  259.         return -1;
  260.     }
  261.  
  262.     return 0;
  263. }
  264.  
  265. // 任务消息队列取消息
  266. int GetOccupiedMsgBufPos(long *typeOfMsg)
  267. {
  268.     TOccupiedPosOfMsgBuf   posOfMsgBuf;
  269.     int                         ret;
  270.     int                         index;
  271.     
  272.     memset(posOfMsgBuf.indexOfPos, 0, sizeof(posOfMsgBuf.indexOfPos));
  273.     if ((ret = msgrcv(s_pMsgBufHDL->queueIDOfOccupiedPos, &posOfMsgBuf, sizeof(int), 0, 0)) < 0) {
  274.         return -1;
  275.     }
  276.     memcpy(&index, posOfMsgBuf.indexOfPos, sizeof(int));
  277.     *typeOfMsg = posOfMsgBuf.typeOfMsg;
  278.         
  279.     return index;
  280. }
  281.  
  282. // 从任务消息队列获取指定类型的消息
  283. int GetOccupiedMsgBufPosOfMsgType(long typeOfMsg)
  284. {
  285.     TOccupiedPosOfMsgBuf    posOfMsgBuf;
  286.     int                     ret;
  287.     int                     index;
  288.     int                     iRetryTime = 0;
  289.  
  290.     memset(posOfMsgBuf.indexOfPos, 0, sizeof(posOfMsgBuf.indexOfPos));
  291.     posOfMsgBuf.typeOfMsg = typeOfMsg;
  292.  
  293.     do {
  294.         ret = msgrcv(s_pMsgBufHDL->queueIDOfOccupiedPos, &posOfMsgBuf, sizeof(int), typeOfMsg, 0);
  295.         iRetryTime ++;
  296.     }while(iRetryTime < 10 && 1 == ret && EINTR == errno);
  297.  
  298.     if (-1 != ret) {
  299.         memcpy(&index, posOfMsgBuf.indexOfPos, sizeof(int));
  300.         return index;
  301.     }
  302.  
  303.     return -1;
  304. }
  305.  
  306. // 发送消息
  307. int WriteBufferMessage(const unsigned char * msg,
  308.                 int lenOfMsg, 
  309.                 long typeOfMsg, 
  310.                 const TMessageHeader* poriHeader, 
  311.                 PTMessageHeader pNewHeader)
  312. {
  313.     int                    indexOfMsg;
  314.     unsigned char         *pAddr;
  315.     int                    ret;
  316.     PTMessageHeader    pheader;
  317.  
  318.     
  319.     if ((indexOfMsg = GetAvailableMsgBufPos()) < 0) {
  320.         return indexOfMsg;
  321.     }
  322.     if ((indexOfMsg >= s_pMsgBufHDL->maxNumOfMsg)) {
  323.         FreeMsgBufPos(indexOfMsg);
  324.         return(-1);
  325.     }
  326.     
  327.     // 得到存放消息的首地址
  328.     pAddr = s_pMsgBuf + indexOfMsg * (sizeof(TMessageHeader) + s_pMsgBufHDL->maxSizeOfMsg);
  329.     pheader = (PTMessageHeader)pAddr;
  330.     pheader->len = lenOfMsg;
  331.     if (NULL == poriHeader) {
  332.         time(&(pheader->time));
  333.         pheader->provider = getpid();
  334.         pheader->dealer = 0;
  335.         pheader->msgIndex = addAndFetchUlongAtomic(&gMsgIndex, 1);
  336.         pheader->forward_cnt = 0;
  337.     }
  338.     else
  339.     {
  340.         pheader->time = poriHeader->time;
  341.         pheader->provider = poriHeader->provider;
  342.         pheader->dealer = getpid();
  343.         pheader->msgIndex = poriHeader->msgIndex;
  344.         pheader->forward_cnt = poriHeader->forward_cnt;
  345.     }
  346.     pheader->type = typeOfMsg;
  347.     if(pNewHeader) {
  348.         memcpy(pNewHeader, pheader, sizeof(*pNewHeader));
  349.     }
  350.     memcpy(pAddr + sizeof(TMessageHeader), msg, lenOfMsg);
  351.     if ((ret = OccupyMsgBufPos(indexOfMsg, typeOfMsg)) < 0) {
  352.         FreeMsgBufPos(indexOfMsg);
  353.         return ret;
  354.     }
  355.  
  356.     std::cout << "WriteBufferMessage -> indexOfMsg:: " << indexOfMsg << " typeOfMsg::" << typeOfMsg << " pheader->len::" <<  pheader->len << std::endl;
  357.     return 0;
  358. }
  359.  
  360. // 用指定位置发消息
  361. int WriteBufferMessageUsingIndex(const unsigned char * msg, 
  362.                             int lenOfMsg, long typeOfMsg,
  363.                             const TMessageHeader* poriHeader, 
  364.                             PTMessageHeader pNewHeader, 
  365.                             const int *piIndexOfMsg)
  366. {
  367.     int indexOfMsg = *piIndexOfMsg;
  368.     int ret;
  369.     unsigned char *pAddr;
  370.     PTMessageHeader pheader;
  371.     
  372.     if ((indexOfMsg >= s_pMsgBufHDL->maxNumOfMsg)) {
  373.         return -1;
  374.     }
  375.     
  376.     pAddr = s_pMsgBuf + indexOfMsg * (sizeof(TMessageHeader) + s_pMsgBufHDL->maxSizeOfMsg);
  377.     pheader = (PTMessageHeader)pAddr;
  378.     pheader->len = lenOfMsg;
  379.     if (poriHeader == NULL)
  380.     {
  381.         time(&(pheader->time));
  382.         pheader->provider = getpid();
  383.         pheader->dealer = 0;
  384.         pheader->msgIndex = addAndFetchUlongAtomic(&gMsgIndex, 1);
  385.         pheader->forward_cnt = 0;
  386.     }
  387.     else
  388.     {
  389.         pheader->time = poriHeader->time;
  390.         pheader->provider = poriHeader->provider;
  391.         pheader->dealer = getpid();
  392.         pheader->msgIndex = poriHeader->msgIndex;
  393.         pheader->forward_cnt = poriHeader->forward_cnt;
  394.     }
  395.     pheader->type = typeOfMsg;
  396.     if(pNewHeader) {
  397.         memcpy(pNewHeader, pheader, sizeof(*pNewHeader));
  398.     }
  399.     memcpy(pAddr + sizeof(TMessageHeader), msg, lenOfMsg);
  400.     if ((ret = OccupyMsgBufPos(indexOfMsg, typeOfMsg)) < 0)
  401.     {
  402.         return ret;
  403.     }
  404.     return(0);
  405. }
  406.  
  407. //接收消息 - 释放索引
  408. int ReadBufferMessagReleaseIndex(unsigned char * msg, int lenOfMsgBuf,
  409.                                 long typeOfMsg,
  410.                                 PTMessageHeader pOutputHeader)
  411. {
  412.     int                 indexOfMsg;
  413.     unsigned char       *pAddr;
  414.     int                 lenOfMsg;
  415.     PTMessageHeader     pheader;
  416.     
  417.     if (((indexOfMsg = GetOccupiedMsgBufPosOfMsgType(typeOfMsg)) < 0)  || (indexOfMsg >= s_pMsgBufHDL->maxNumOfMsg))
  418.     {
  419.         return -1;
  420.     }
  421.     pAddr = s_pMsgBuf + indexOfMsg * (sizeof(TMessageHeader) + s_pMsgBufHDL->maxSizeOfMsg);
  422.     pheader = (PTMessageHeader)pAddr;
  423.     if (((lenOfMsg = pheader->len) < 0) || (lenOfMsg >= lenOfMsgBuf) || (lenOfMsg > s_pMsgBufHDL->maxSizeOfMsg))
  424.     {
  425.         return -1;
  426.     }
  427.     memcpy(msg, pAddr + sizeof(TMessageHeader), lenOfMsg);
  428.     if (pOutputHeader != NULL) {
  429.         memcpy(pOutputHeader, pheader, sizeof(*pheader));
  430.     }
  431.         
  432.     FreeMsgBufPos(indexOfMsg);
  433.  
  434.     std::cout << "ReadBufferMessagReleaseIndex -> indexOfMsg:: " << indexOfMsg << " typeOfMsg::" << typeOfMsg << " pheader->len::" <<  pheader->len << std::endl;
  435.     
  436.     return(lenOfMsg);
  437. }
  438.  
  439.  
  440. //接收消息 - 不释放索引
  441. int ReadBufferMessageNotReleaseIndex(unsigned char *msg, int lenOfMsgBuf, 
  442.                                 long typeOfMsg, 
  443.                                 PTMessageHeader pOutputHeader, 
  444.                                 int *indexOfMsg, int iSigRetryFlag)
  445. {
  446.     unsigned char         *pAddr;
  447.     int                    lenOfMsg;
  448.     PTMessageHeader    pheader;
  449.     
  450.     if (((*indexOfMsg = GetOccupiedMsgBufPosOfMsgType(typeOfMsg)) < 0)  || (*indexOfMsg >= s_pMsgBufHDL->maxNumOfMsg))
  451.     {
  452.         return -1;
  453.     }
  454.     pAddr = s_pMsgBuf + *indexOfMsg * (sizeof(TMessageHeader) + s_pMsgBufHDL->maxSizeOfMsg);
  455.     pheader = (PTMessageHeader)pAddr;
  456.     if (((lenOfMsg = pheader->len) < 0) || (lenOfMsg >= lenOfMsgBuf) || (lenOfMsg > s_pMsgBufHDL->maxSizeOfMsg))
  457.     {
  458.         FreeMsgBufPos(*indexOfMsg);
  459.         return -1;
  460.     }
  461.     memcpy(msg, pAddr + sizeof(TMessageHeader), lenOfMsg);
  462.     if (pOutputHeader != NULL) {
  463.         memcpy(pOutputHeader, pheader, sizeof(*pheader));
  464.     }
  465.         
  466.     return lenOfMsg;
  467. }
  468.  
  469.

生产者进程：productionProcess.cpp

  1. #include <stdio.h>
  2. #include <sys/shm.h>
  3. #include <string.h>
  4. #include <iostream>
  5. #include "msgExchange.h"
  6.  
  7. using namespace std;
  8.  
  9. int SimpleTestWrite() {
  10.     // 1. 创建共享内存
  11.     int shmid = shmget(1000, 4096, IPC_CREAT|0660);
  12.     if(shmid == -1)
  13.     {
  14.         perror("shmget error");
  15.         return -1;
  16.     }
  17.  
  18.     // 2. 进程和共享内存关联
  19.     void* ptr = shmat(shmid, NULL, 0);
  20.     if(ptr == (void *) -1)
  21.     {
  22.         perror("shmat error");
  23.         return -1;
  24.     }
  25.  
  26.     // 3. 写共享内存
  27.     const char* p = "hello, nice to meet you !";
  28.     memcpy(ptr, p, strlen(p)+1);
  29.  
  30.     getchar();
  31.  
  32.     // 4. 解除内存和共享内存关联
  33.     shmdt(ptr);
  34.  
  35.     // 5. 删除共享内存
  36.     shmctl(shmid, IPC_RMID, NULL);
  37.     std::cout << "delete share memory !" << std::endl;
  38. }
  39.  
  40. int main(int argc, char *argv[])
  41. {
  42.     InitSharedMemoryMsgBuf();
  43.  
  44.     // 发送消息
  45.     TMessageHeader newHeader;
  46.     std::string sendMsg("Hello, I am producer !");
  47.     WriteBufferMessage((unsigned char *)sendMsg.data(),
  48.                 sendMsg.size(), 
  49.                 10101010, 
  50.                 nullptr, 
  51.                 &newHeader);
  52.     // 读消息
  53.     unsigned char recvMsg[8192] = {0};
  54.     ReadBufferMessagReleaseIndex(recvMsg, 8192,
  55.                                 newHeader.provider,
  56.                                 &newHeader);
  57.     std::cout << "recvMsg:: " << recvMsg << std::endl;
  58.  
  59.     UnInitSharedMemoryMsgBuf();
  60.     
  61.     return 0;
  62. }

消费者进程：consumptionProcess.cpp

  1. #include <sys/shm.h>
  2. #include <string.h>
  3. #include <stdio.h>
  4. #include <iostream>
  5.  
  6. #include "msgExchange.h"
  7.  
  8. using namespace std;
  9.  
  10. int SimpleTestRead() {
  11.      // 1. 创建共享内存
  12.     int shmid = shmget(1000, 0, 0660);
  13.     if(shmid == -1)
  14.     {
  15.         perror("shmget error");
  16.         return -1;
  17.     }
  18.  
  19.     // 2. 获取共享内存地址
  20.     void* ptr = shmat(shmid, NULL, 0);
  21.     if(ptr == (void *) -1)
  22.     {
  23.         perror("shmat error");
  24.         return -1;
  25.     }
  26.  
  27.     // 3. 读共享内存
  28.     std::cout << "read memory:: " << (char*)ptr << std::endl;
  29.  
  30.     // 4. 解除关联
  31.     shmdt(ptr);
  32.  
  33.     // 5. 删除
  34.     shmctl(shmid, IPC_RMID, NULL);
  35.     std::cout << "delete share memory !" << std::endl;
  36. }
  37.  
  38. int main(int argc, char *argv[])
  39. {
  40.     InitSharedMemoryMsgBuf();
  41.  
  42.      // 读消息
  43.     TMessageHeader recvHeader;
  44.     unsigned char recvMsg[8192] = {0};
  45.     ReadBufferMessagReleaseIndex(recvMsg, 8192,
  46.                                 10101010,
  47.                                 &recvHeader);
  48.     std::cout << "recvMsg:: " << recvMsg << std::endl;
  49.  
  50.     // 发送消息
  51.     TMessageHeader newHeader;
  52.     std::string sendMsg("Hello, I am consumer, I have receiving your msg, thank you!");
  53.     WriteBufferMessage((unsigned char *)sendMsg.data(),
  54.                 sendMsg.size(), 
  55.                 recvHeader.provider, 
  56.                 &recvHeader, 
  57.                 &newHeader);
  58.                 
  59.     UnInitSharedMemoryMsgBuf();
  60.  
  61.     return 0;
  62. }

编译脚本：Makefile

  1. app: productionProcess consumptionProcess
  2.  
  3. #说明：$^代表依赖项
  4. productionProcess: productionProcess.cpp msgExchange.cpp
  5. 	g++ -g $^ -o productionProcess
  6.  
  7. consumptionProcess: consumptionProcess.cpp msgExchange.cpp
  8. 	g++ -g $^ -o consumptionProcess
  9.  
  10. clean:
  11. 	-rm productionProcess consumptionProcess -f

调试配置：launch.json

  1. {
  2.   "version": "0.2.0",
  3.   "configurations": [
  4.     {
  5.       "name": "C/C++ Runner: Debug Session",
  6.       "type": "cppdbg",
  7.       "request": "launch",
  8.       "args": [],
  9.       "stopAtEntry": false,
  10.       "externalConsole": false,
  11.       "cwd": "/home/tiger/ipc",
  12.       "program": "/home/tiger/ipc/productionProcess",
  13.       "MIMode": "gdb",
  14.       "miDebuggerPath": "gdb",
  15.       "setupCommands": [
  16.         {
  17.           "description": "Enable pretty-printing for gdb",
  18.           "text": "-enable-pretty-printing",
  19.           "ignoreFailures": true
  20.         }
  21.       ]
  22.     }
  23.   ]
  24. }

4.2 运行效果

从运行效果看，程序创建了共享内存5308433，空闲消息队列12，任务队列13，生产者进程从空闲队列去取出位置0，按照位置在共享内存写入信息，然后发送消息到任务队列，然后等待消费者进程答复，消费着进程启动后从任务队列中取出位置，按照位置从共享内存读取数据，并打印，然后从空闲队列取出1号位置，按照位置在共享内存中写入要发送的信息。然后将位置信息放入任务队列，生产者进程从任务队列取出消息位置，按照位置从共享内存中取出消费者进程的消息并打印出来，同时归还位置信息到空闲队列，下次再继续使用。

5.execve系统调用

进程调度之 4:系统调用execve - 我叫半桶水的个人空间 - OSCHINA - 中文开源技术交流社区

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小小林熬夜学编程/article/detail/601603
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号