[转帖] FIFO消息队列（内存、文件双缓冲模式）_OpenStack, 云计算等杂项讨论区_Weblogic技术|Tuxedo技术|中间件技术|Oracle论坛|JAVA论坛|Linux/Unix技术|hadoop论坛_联动北方技术论坛  
网站首页 | 关于我们 | 服务中心 | 经验交流 | 公司荣誉 | 成功案例 | 合作伙伴 | 联系我们 |
联动北方技术论坛-国内领先的Weblogic、Tuxedo中间件技术论坛
联动北方-国内领先的云技术服务提供商
»  游客             当前位置：  论坛首页 »  自由讨论区 »  OpenStack, 云计算等杂项讨论区 »
总帖数
1
每页帖数
101/1页1
返回列表
0
发起投票  发起投票 发新帖子
查看: 4602 | 回复: 0   主题： [转帖] FIFO消息队列（内存、文件双缓冲模式）        上一篇   下一篇 
张三丰
注册用户
等级：上尉
经验：511
发帖：53
精华：0
注册：2013-11-19
状态：离线
发送短消息息给张三丰 加好友    发送短消息息给张三丰 发消息
发表于： IP：您无权察看 2013-11-21 11:00:30 | [全部帖] [楼主帖] 楼主

1          概述
本文档适用于C语言开发人员，用于在开发过程中进行参考，文档详细讲述每个
函数的用法以及源码示例。

2          支持文件缓存消息队列介绍
一般的FIFO消息队列都是基于内存，即所有的消息都保存在内存中。但是在很多情况下，消息队列中的消息数目难以确定，可能超过系统内存的容量，导致传统的FIFO消息队列无法满足应用的需求，所以推出基于内存加磁盘文件的FIFO消息队列。

基于缓存文件的消息队列是消息队列的消息保存在内存和缓存文件中，当内存中
的消息超过指定的数量后，把消息队列中的部分消息保存到一个或多个消息文件中。但是消息队列中的头部若干消息和尾部的若干消息始终保持在内存，用于提高整体性能，由于使用磁盘文件作为二次缓冲，而目前的磁盘容量巨大，可认为消息队列几乎没有容量的限制。

如果消息队列中的消息始终不超过指定数量，即不再使用缓存文件的情况下则和传统的FIFO消息队列功能上时相同的。

3          支持文件缓存消息队列技术特点
1、支持FIFO方式的消息队列
    即是一个先进先出的消息队列，在使用内存和缓存文件的情况下也严格保证
FIFO方式。

2、支持磁盘文件的二次缓冲
即消息队列是基于内存和磁盘缓冲文件。

3、支持多优先级
    在消息队列内存有多个子消息队列，每个优先级对应一个子消息队列。每个
子消息队列都可支持内存缓冲加磁盘缓冲消息，在从消息队列中取消息时，
始终优先处理级别高的子消息队列。

4、容量大
   由于使用基于文件的二次缓冲，消息队列的容量几乎没有限制

5、��度快
消息队列在没有使用磁盘缓冲的情况下（即没有磁盘IO）运行方式和传统模式的FIFO消息队列是一样的，即使使用磁盘缓冲由于一次加载一个缓冲文件，而每个磁盘文件在拥有很多消息的情况下，整体性能不会降低。

6、消息队列初始化时可从磁盘缓存文件加载消息，在删除队列时也可把内存中的消息保存到磁盘缓存文件，确保消息不丢失。

7、功能强大的API函数
   0）生成消息队列，并可指定消息队列的属性。
    1）支持从尾部追加消息
    2）支持从头部取消息
    3）也支持在头部插入消息
    4）支持多优先级
    5）删除消息队列

4          消息队列应用场景
1、短消息系统
2、彩信系统
3、其它存储转发消息系统
4、批处理系统
5、传统FIFO消息队列应用领域
6、其它情况
5          消息队列功能函数
5.1     MQCreateQueue()

1、功能说明
此函数用来生成一个FIFO消息队列，执行成功后返回消息队列句柄，之后对
此消息队列的操作都引用此句柄。

2、函数原型

int MQCreateQueue(LPQUEUEINIT pQueueInit, HANDLE& hQueue);


3、参数说明
1）LPHTREEDESC pTreeDesc
输入参数
描述消息队列属性的结构体，定义如下：

struct _QUEUEINIT
{
      char* sRootDir;
      char* sName;
      int nPriority;
      int nFileMsgNum;
      bool bPreDeleteFile;
      bool bEnableSyncObj;
}


    结构体成员说明：

l        sRootDir


消息队列缓存文件所在的目录，目录必须是全路径且已经存在。

l        sName


消息队列的名字，名字必须满足操作系统对文件名的要求。不同的
消息队列用不同的名字,否则可能会造成不同的消息队列在同一目
录下生成相同文件名的缓存文件。
说明：缓存文件名的名字构成如下规则：消息对列名_优先级_序列     号.afx，例如：

sms_0_00000001.afx
sms_1_00000001.afx
sms_2_00000001.afx
l        nPriority


消息队列优先级的数量,该变量必须大于0。消息队列内部优先级从
0开始依次递增。

l        nFileMsgNum


每个缓存文件内消息的最大数目。如果该变量<= 0，则消息队列不
使用文件缓存，所有的消息都保存在内存中；否则当内存中缓存的
超过此变量值后，即写入缓存文件。

l        bPreDeleteFile


初始化消息队列时，是否先删除目录下对应的缓存文件。
true: 也就是消息队列初始化后，消息队列中的消息为空。
false: 初始化消息队列缓冲文件。

l        bEnableSyncObj


是否启用内部同步对象。
true: 启用内部同步对象。
false: 禁用内部同步对象，由应用程序进行同步控制，一般使用互
斥锁或者临界区。

    2) HANDLE& hQueue
       输出参数，执行成功返回消息队列的句柄。

4、返回值
执行成功返回0，失败返回非0。

5、相关函数

 HQDeleteQueue()


示例1：
功能：生成一个消息队列，此消息队列有如下属性
1、消息队列对应的缓存文件目录为e:\cache目录
2、消息队列名字为sms
3、消息队列有3个优先级
4、每个缓冲文件缓冲1024个消息
5、消息队列在初始化时加载对应缓存目录中的缓存文件消息
6、使用消息队列内部的同步对象

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      // 消息队列缓存文件路径，此路径必须存在
      pInitInfo.sRootDir = “e:\\cache”;
      // 消息队列的名称
      pInitInfo.sName = "sms";
      // 消息队列优先级的数量
      pInitInfo.nPriority = 3;
      // 每个消息缓冲文件的消息最大数
      pInitInfo.nFileMsgNum = 1024;
      // 初始化时使用缓冲文件
      pInitInfo.bPreDeleteFile = false;
      // 启用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;// 消息队列句柄
      nRet = MQCreateQueue(&pInitInfo, hMsgQueue);
      return 0;
}


示例2：
功能：生成一个消息队列，此消息队列有如下属性
1、消息队列对应的缓存文件目录为e:\cache目录
2、消息队列名字为sms
3、消息队列有3个优先级
4、不适用文件缓存，所有的消息都保存在内存中。
5、消息队列在初始化时加载对应缓存目录中的缓存文件消息
6、使用消息队列内部的同步对象

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      // 消息队列缓存文件路径，此路径必须存在
      pInitInfo.sRootDir = “e:\\cache”;
      // 消息队列的名称
      pInitInfo.sName = "sms";
      // 消息队列优先级的数量
      pInitInfo.nPriority = 3;
      // 消息队列不使用缓存文件，消息全部保存在内存中
      pInitInfo.nFileMsgNum = 0;
      // 初始化时使用缓冲��件
      pInitInfo.bPreDeleteFile = false;
      // 启用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;// 消息队列句柄
      nRet = MQCreateQueue(&pInitInfo, hMsgQueue);
      return 0;
}


示例3：
功能：生成一个消息队列，此消息队列有如下属性
1、消息队列对应的缓存文件目录为e:\cache目录
2、消息队列名字为sms
3、消息队列有3个优先级
4、每个缓冲文件缓冲1024个消息
5、消息队列在初始化时先删除对应的缓存文件，初始化时消息数量为0。
6、使用消息队列内部的同步对象

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      // 消息队列缓存文件路径，此路径必须存在
      pInitInfo.sRootDir = “e:\\cache”;
      // 消息队列的名称
      pInitInfo.sName = "sms";
      // 消息队列优先级的数量
      pInitInfo.nPriority = 3;
      // 每个消息缓冲文件的消息最大数
      pInitInfo.nFileMsgNum = 1024;
      // 初始化时先删除缓存文件，初始化消息数量为0
      pInitInfo.bPreDeleteFile = true;
      // 启用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;// 消息队列句柄
      nRet = MQCreateQueue(&pInitInfo, hMsgQueue);
      return 0;
}
5.2 MQDeleteQueue()


1、功能说明
此函数用来删除消息队列，删除后的消息队列句柄不能再被引用。

2、函数原型

void MQDeleteQueue(HANDLE hQueue, bool bDeleteFile);


3、参数说明
   1）HANDLE hQueue
      输入参数，消息队列句柄。

2）bool bDeleteFile;
输入参数，删除消息队列时是否同时删除对应的缓存文件。
     true: 删除缓存文件
     false: 不删除缓存文件

4、返回值
    无

5、相关函数

 MQCreateQueue()


示例1：

功能：删除一个消息队列，删除队列时，内存中的消息直接删除，同时也删除对应的缓存文件。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      // 消息队列缓存文件路径，此路径必须存在
      pInitInfo.sRootDir = “e:\\cache”;
      // 消息队列的名称
      pInitInfo.sName = "sms";
      // 消息队列优先级的数量
      pInitInfo.nPriority = 3;
      // 每个消息缓冲文件的消息最大数
      pInitInfo.nFileMsgNum = 1024;
      // 初始化时使用缓冲文件
      pInitInfo.bPreDeleteFile = false;
      // 启用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;// 消息队列句柄
      nRet = MQCreateQueue(&pInitInfo, hMsgQueue);
      int i;
      for(i = 0; i < 1024*10 + 100; i++)
      {
            MQAddTail(hMsgQueue, 0, "1234567890", 10);
            MQAddTail(hMsgQueue, 1, "1234567891", 10);
            MQAddTail(hMsgQueue, 2, "1234567892", 10);
      }
      // 删除消息队列时删除缓存文件
      MQDeleteQueue(hMsgQueue, true);
      return 0;
}


示例2：

功能：删除一个消息队列，删除队列时，内存中的消息保存到缓存文件，不删除缓存文件，缓存文件用于下次初始化消息队列。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      // 消息队列缓存文件路径，此路径必须存在
      pInitInfo.sRootDir = “e:\\cache”;
      // 消息队列的名称
      pInitInfo.sName = "sms";
      // 消息队列优先级的数量
      pInitInfo.nPriority = 3;
      // 每个消息缓冲文件的消息最大数
      pInitInfo.nFileMsgNum = 1024;
      // 初始化时使用缓冲文件
      pInitInfo.bPreDeleteFile = false;
      // 启用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;// 消息队列句柄
      nRet = MQCreateQueue(&pInitInfo, hMsgQueue);
      int i;
      for(i = 0; i < 1024*10 + 100; i++)
      {
            MQAddTail(hMsgQueue, 0, "1234567890", 10);
            MQAddTail(hMsgQueue, 1, "1234567891", 10);
            MQAddTail(hMsgQueue, 2, "1234567892", 10);
      }
      // 删除消息队列时不删除缓存文件
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}
5.3 MQAddTail ()


1、功能说明
此函数用来在消息队列的尾部追加消息。

2、函数原型

int MQAddTail(HANDLE hQueue, int nPriority, char* pData, int nLength);


3、参数说明
   1）HANDLE hQueue
      消息队列句柄，此句柄由MQCreateQueue函数生成。

2）int nPriority
  指明此消息的优先级，优先级从0开始，nPriority值越大说明此消息的
优先级越高，越优先处理。
举例：如果消息队列的优先级为3，此参数的值应该从0~2，如果
nPriority >= 3，则消息被放入到一个特殊内部子消息队列，此子消息队
列优先级最高，且不支持文件缓存，只在特殊应用中使用此优先级和对应的子消息队列。

3）char* pData
  消息数据指针,不能是NULL。

4）int nLength
pData指针指向消息的长度。

4、返回值
    1）0，追加消息成功。
    2）其它，追加消息失败。

5、相关函数   

MQRemoveHead()
MQGetMsgCount()


示例1：
功能：在消息队列中插入消息。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      // 创建支持文件缓存的消息队列,消息队列有3个优先级
      MQCreateQueue(&pInitInfo, hMsgQueue);
      // 插入优先级为0的消息到消息队列。消息内容是字符串为1234567890，长度为10个
      // 字节
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      // 插入优先级为1的消息到消息队列
      MQAddTail(hMsgQueue, 1, "1234567891", 10);
      // 插入优先级为2的消息到消息队列
      MQAddTail(hMsgQueue, 2, "1234567892", 10);
      // 插入优先级为3的消息到消息队列，此消息被放入到内部优先级最高的子消息队列。
      MQAddTail(hMsgQueue, 3, "1234567893", 10);
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}
5.4 MQRemoveHead ()


1、功能说明
此函数用来从消息队列中获取一个消息,因为消息队列内部有多个不同优先级的子消息队列,此函数从优先级从高到低检查消息队列，一但有非空的子消息队列，则从此子消息队列的头部获取消息，这意味着高优先级的消息总是被优先处理。

2、函数原型

int MQRemoveHead(HANDLE hQueue, int& nPriority, char* pData,
int& nLength, unsigned long dwTimeOut = 0xFFFFFFFF);


3、参数说明
   1）HANDLE hTree
      输入参数
      消息队列句柄，此句柄由MQCreateQueue函数生成。

2）int& nPriority
输出参数
返回获取到消息的优先级

3）char* pData
输出参数
用于接收返回的消息数据，指针指向的空间由外部调用程序分配。

4）int& nLength
输出参数
返回获取到消息的长度。

5）unsigned long dwTimeOut = 0xFFFFFFFF
输入参数
此参数为等待消息队列为非空的时长，此参数只对使用消息队列内部同步
对象时有效，如果不使用内部同步对象，则此参数无效。

如果该值为0，则不等待;如果为0xFFFFFFFF,则一直等待直到消息队列中
有消息。

4、返回值
    1）0 从消息队列中成功获取到一条消息。
   2）其他值失败。

5、相关函数

 MQAddTail()


示例1：
功能：从消息队列中获取消息。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      // 创建支持文件缓存的消息队列,消息队列有3个优先级
      MQCreateQueue(&pInitInfo, hMsgQueue);
      // 插入优先级为0的消息到消息队列。消息内容是字符串为1234567890，长度为10个
      // 字节
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      // 插入优先级为1的消息到消息队列
      MQAddTail(hMsgQueue, 1, "1234567891", 10);
      // 插入优先级为2的消息到消息队列
      MQAddTail(hMsgQueue, 2, "1234567892", 10);
      // 插入优先级为3的消息到消息队列，此消息被放入到内部优先级最高的子消息队列。
      MQAddTail(hMsgQueue, 3, "1234567893", 10);
      int nPriority;
      char pBuf[1024];
      int nLength;
      if(MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      if(MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      if(MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      if(MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}
5.5 MQAddHead ()


1、功能说明
此函数用来在消息队列的头部追加消息，此函数只有在非常情况的使用，一
般是在删除消息队列前，把之前从头部获取且没有处理的消息重新放入到队
列的头部，用来下次初始化消息队列时，保持原有队列中消息的正确消息。

2、函数原型

int MQAddHead(HANDLE hQueue, int nPriority, char* pData, int nLength);


3、参数说明
   1）HANDLE hQueue
      消息队列句柄，此句柄由MQCreateQueue函数生成。

2）int nPriority
消息的优先级，参考MQAddTail函数中相关参数的说明。

3）char* pData
  消息数据指针,不能是NULL。

4）int nLength
pData指针指向消息的长度。

4、返回值
    1）0，追加消息成功。
    2）其它，追加消息失败。

5、相关函数   

MQAddTail()
MQRemoveHead()
MQGetMsgCount()


示例1：
功能：在消息队列的头部添加消息。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      // 创建支持文件缓存的消息队列,消息队列有3个优先级
      MQCreateQueue(&pInitInfo, hMsgQueue);
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      MQAddTail(hMsgQueue, 0, "1234567891", 10);
      MQAddTail(hMsgQueue, 0, "1234567892", 10);
      MQAddTail(hMsgQueue, 0, "1234567893", 10);
      MQAddTail(hMsgQueue, 0, "1234567894", 10);
      MQAddTail(hMsgQueue, 0, "1234567895", 10);
      MQAddTail(hMsgQueue, 0, "1234567896", 10);
      MQAddTail(hMsgQueue, 0, "1234567897", 10);
      MQAddTail(hMsgQueue, 0, "1234567898", 10);
      MQAddTail(hMsgQueue, 0, "1234567899", 10);
      int nPriority;
      char pData[1024];
      int nLength;
      // 从消息队列中取出消息   
      MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0)
      // 把上面取到的消息再放回原来的队列
      MQAddHead(hMsgQueue, nPriority, pData, nLength);
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}
5.6 MQGetMsgCount()


1、功能说明
此函数用来获取消息队列中消息的总数。

2、函数原型

int MQGetMsgCount(HANDLE hQueue);


3、参数说明
   1）HANDLE hQueue
      消息队列句柄，此句柄由MQCreateQueue函数生成。

4、返回值
    返回消息的数量

5、相关函数   

MQAddTail()
MQAddHead()
MQRemoveHead()


示例1：
功能：获取消息队列中的消息总数。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      MQCreateQueue(&pInitInfo, hMsgQueue);
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      MQAddTail(hMsgQueue, 1, "1234567891", 10);
      MQAddTail(hMsgQueue, 2, "1234567892", 10);
      MQAddTail(hMsgQueue, 3, "1234567893", 10);
      MQAddTail(hMsgQueue, 4, "1234567894", 10);
      MQAddTail(hMsgQueue, 5, "1234567895", 10);
      MQAddTail(hMsgQueue, 6, "1234567896", 10);
      MQAddTail(hMsgQueue, 7, "1234567897", 10);
      MQAddTail(hMsgQueue, 8, "1234567898", 10);
      MQAddTail(hMsgQueue, 9, "1234567899", 10);
      int nMsgCount;
      nMsgCount = MQGetMsgCount(hMsgQueue);
      print(“Msg Count: %d\r\n”, nMsgCount);
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}
5.7 MQRemoveAll()


1、功能说明
此函数用来清空消息队列中消息以及缓存文件。

2、函数原型

void MQRemoveAll(HANDLE hQueue);


3、参数说明
   1）HANDLE hQueue
      消息队列句柄，此句柄由MQCreateQueue函数生成。

4、返回值
    返回消息的数量

5、相关函数   

MQAddTail()
MQAddHead()
MQRemoveHead()


示例1：
功能：获取消息队列中的消息总数。

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      MQCreateQueue(&pInitInfo, hMsgQueue);
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      MQAddTail(hMsgQueue, 1, "1234567891", 10);
      MQAddTail(hMsgQueue, 2, "1234567892", 10);
      MQAddTail(hMsgQueue, 3, "1234567893", 10);
      MQAddTail(hMsgQueue, 4, "1234567894", 10);
      MQAddTail(hMsgQueue, 5, "1234567895", 10);
      MQAddTail(hMsgQueue, 6, "1234567896", 10);
      MQAddTail(hMsgQueue, 7, "1234567897", 10);
      MQAddTail(hMsgQueue, 8, "1234567898", 10);
      MQAddTail(hMsgQueue, 9, "1234567899", 10);
      printf(“Msg Count: %d\r\n”, MQGetMsgCount(hMsgQueue));
      MQRemoveAll(hMsgQueue);
      printf(“Msg Count: %d\r\n”, MQGetMsgCount(hMsgQueue));
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}


6          消息队列demo演示
6.1     简单功能演示
实现功能：
1、  创建一个支持文件缓存的消息队列
2、  添加若干消息
3、  取出消息，进行屏幕打印
4、  删除消息队列

#include "MsgQueue.h"
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      pInitInfo.bEnableSyncObj = true;
      int nRet;
      HANDLE hMsgQueue;
      MQCreateQueue(&pInitInfo, hMsgQueue);
      MQAddTail(hMsgQueue, 0, "1234567890", 10);
      MQAddTail(hMsgQueue, 1, "1234567891", 10);
      MQAddTail(hMsgQueue, 2, "1234567892", 10);
      MQAddTail(hMsgQueue, 3, "1234567893", 10);
      MQAddTail(hMsgQueue, 4, "1234567894", 10);
      MQAddTail(hMsgQueue, 5, "1234567895", 10);
      MQAddTail(hMsgQueue, 6, "1234567896", 10);
      MQAddTail(hMsgQueue, 7, "1234567897", 10);
      MQAddTail(hMsgQueue, 8, "1234567898", 10);
      MQAddTail(hMsgQueue, 9, "1234567899", 10);
      int nPriority;
      char pBuf[1024];
      int nLength;
      if(MQRemoveHead(hMsgQueue, nPriority, pBuf, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      if(MQRemoveHead(hMsgQueue, nPriority, pBuf, nLength, 0) == 0)
      {
            pBuf[nLength] = ‘\0’;
            printf(“Data: %s Priority: %d\r\n”, pBuf, nPriority);
      }
      MQDeleteQueue(hMsgQueue, false);
      return 0;
}


6.2     多写单读模式
创建一个支持文件缓存的消息队列，多个线程追加消息，一个线程处理消息。该消息队列支持多个优先级。

#include "MsgQueue.h"
DWORD WINAPI WriteThread(void* pPara)
{
      HANDLE hMsgQueue;
      hMsgQueue = pPara;
      char pData[1024];
      int i;
      for (i = 0; ; i++)
      {
            sprintf(pData, "this is msg: %d", i);
            MQAddTail(hMsgQueue, 0, pData, strlen(pData));
            MQAddTail(hMsgQueue, 1, pData, strlen(pData));
            MQAddTail(hMsgQueue, 2, pData, strlen(pData));
            Sleep(1);
      }
      return 0;
}
DWORD WINAPI ReadThread(void* pPara)
{
      HANDLE hMsgQueue;
      hMsgQueue = pPara;
      char pData[1024];
      int nPriority;
      int nLength;
      int i;
      for (i = 0; ; i++)
      {
            MQRemoveHead(hMsgQueue, nPriority, pData, nLength, 0xFFFFFFFF);
            pData[nLength] = '\0';
            printf("nPriority: %d %s\r\n", nPriority, pData);
            Sleep(0);
      }
      return 0;
}
int main(int argc, char* argv[])
{
      QUEUEINIT pInitInfo;
      pInitInfo.sRootDir = "e:\\cache";
      pInitInfo.sName = "sms";
      pInitInfo.nPriority = 3;
      pInitInfo.nFileMsgNum = 1024;
      pInitInfo.bPreDeleteFile = false;
      // 使用消息队列内部的同步对象
      pInitInfo.bEnableSyncObj = true;
      HANDLE hMsgQueue;
      MQCreateQueue(&pInitInfo, hMsgQueue);
      CreateThread(NULL, 0, WriteThread, hMsgQueue, 0, NULL);
      CreateThread(NULL, 0, WriteThread, hMsgQueue, 0, NULL);
      CreateThread(NULL, 0, WriteThread, hMsgQueue, 0, NULL);
      CreateThread(NULL, 0, ReadThread, hMsgQueue, 0, NULL);
      char sInput[1024];
      gets(sInput);
      return 0;
}


赞(0)    操作        顶端 
总帖数
1
每页帖数
101/1页1
返回列表
发新帖子
请输入验证码： 点击刷新验证码
您需要登录后才可以回帖 登录 | 注册
技术讨论

Weblogic中间件技术论坛

 |

Tuxedo中间件技术论坛

 |

数据库论坛

 |

Java论坛

 |

Linux/unix论坛

 |

网站地图


Powered by LandingBJ     @ 2008-2018     联动北方   
京ICP证号090597    京ICP备案09017624号    京公网安备110112000093号    
400-810-2327     Email：service@landingbj.com   

Powered by landingbj @ 2008-2015 联动北方
京ICP证号090597 京ICP备案09017624号 京公网安备110112000093号
86-27-87056338 Email：service@landingbj.com