MSMQ - C++ 还是 COM？

Question

我需要开始编写一些与其他机器上的 WCF 代码交互的 MSMQ 代码。有 MSMQ 经验的人是否会就使用直接 C++ 与使用 COM 的 MSMQ 的优缺点提出建议？

Answer 1

其实你不必决定。可以组合。

这是一个完整的发送/接收实现的代码示例。

如果你改进了就告诉我...

h 文件：

#pragma once

#include <tchar.h>

// ==========================================================================
// MSMQWrapper - wrappes the COM object used to send and receive messages through the MSMQ
class CMSMQWrapper
{
    HANDLE m_hQ;

public:
    CMSMQWrapper()
    {
        m_hQ= INVALID_HANDLE_VALUE;
        ::CoInitializeEx(NULL, COINIT_MULTITHREADED);
    }

    ~CMSMQWrapper()
    {
        ::CoUninitialize();
    }

    bool InitLocalQueue (const WCHAR* wczQueueName     ); // [i] .\private$\queue_name
    bool InitDestQueue  (const WCHAR* wczDestQueueName ); // [i] comp_name\private$\queue_name

    bool ReadQueue      (const WCHAR* wczQueueName    ,   // [i]
                               BYTE*  pBuf            ,   // [i]
                               size_t nBufLen         ,   // [i]
                               int&   nBytesRead       ); // [o]

    bool SendToDestQueue(const BYTE*  pBuf            ,
                               size_t nBufLen          );
};

cpp 文件：

#include "stdafx.h"
#include "msmqwrap.h"

#include <windows.h>
#include <AtlBase.h>
#import "mqoa.dll" named_guids // no_namespace

#pragma comment (lib, "Mqrt.lib")
#include "mq.h"

using namespace MSMQ;

// ==========================================================================
// CMSMQWrapper
// ==========================================================================

bool CMSMQWrapper::InitLocalQueue(const WCHAR* wczQueueName)
{
    CComQIPtr<IMSMQQueueInfo, &IID_IMSMQQueueInfo> ipQueueInfo;

    HRESULT hr= ::CoCreateInstance(CLSID_MSMQQueueInfo     ,
                                   NULL                    ,
                                   CLSCTX_SERVER           ,
                                   IID_IMSMQQueueInfo      ,
                                   (void**)(&ipQueueInfo.p) );
    if (S_OK != hr)
        return false;

    hr= ipQueueInfo->put_PathName(_bstr_t(wczQueueName));
    if (S_OK != hr)
        return false;

    hr= ipQueueInfo->put_Label(_bstr_t(wczQueueName));
    if (S_OK != hr)
        return false;

    VARIANT vtFalse;
    VariantInit(&vtFalse);
    vtFalse.vt     = VT_BOOL;
    vtFalse.boolVal= FALSE  ;

    try
    {
        hr= ipQueueInfo->Create(&vtFalse, &vtFalse);
    }
    catch (_com_error& er)
    { 
        if (MQ_ERROR_QUEUE_EXISTS == er.Error()) // queue already exists
            hr= S_OK;
        else
        {
            // report error - Failed receiving, (WCHAR*)er.Description()
            return false;
        }
    }

    return true;
}

// --------------------------------------------------------------------------
bool CMSMQWrapper::ReadQueue(const WCHAR* wczQueueName,  // [i]
                             BYTE*        pBuf        ,  // [i]
                             size_t       nBufLen     ,  // [i]
                             int&         nBytesRead   ) // [o]
{
    // set value of ReceiveTimout parameter
    _variant_t vtReceiveTimeout;
    vtReceiveTimeout= (long)INFINITE;

    try
    {
        IMSMQQueueInfoPtr qinfo("MSMQ.MSMQQueueInfo");
        qinfo->PathName= wczQueueName;

        IMSMQQueuePtr qRec;
        qRec= qinfo->Open(MQ_RECEIVE_ACCESS, MQ_DENY_NONE); // open queue to retrieve message

        // retrieve messages from queue
        IMSMQMessagePtr msgRec("MSMQ.MSMQMessage");
        msgRec= qRec->Receive(&vtMissing, &vtMissing, &vtMissing, &vtReceiveTimeout);
        if (NULL == msgRec)
        {
            nBytesRead= 0; // there are no messages in the queue
            return true;
        }

        nBytesRead           = msgRec->BodyLength;
        _variant_t recVariant= msgRec->Body      ;

        // close queue
        qRec->Close();

        SAFEARRAY* psa= recVariant.parray;
        nBytesRead    = __min(psa->rgsabound->cElements, nBufLen);

        for (LONG ind= 0; ind< nBytesRead; ind++)
            SafeArrayGetElement(psa, &ind, &pBuf[ind]);

        return true;
    }
    catch (_com_error comerr)
    {
        // report error - failed receiving, (WCHAR*)comerr.Description());
        return false;
    }
}

// --------------------------------------------------------------------------
bool CMSMQWrapper::InitDestQueue(const WCHAR* wczDestQueueName) // comp_name\private$\queue_name
{
    // validate the input strings
    if (NULL == wczDestQueueName)
        return false;

    // create a direct format name for the queue
    WCHAR wczFormatName[1000];
    str_cat(wczFormatName, 1000, L"DIRECT=OS:", wczDestQueueName);

    HRESULT hr;
    hr = ::MQOpenQueue(wczFormatName, MQ_SEND_ACCESS, MQ_DENY_NONE, &m_hQ);
    if (MQ_OK != hr) //MQ_ERROR_QUEUE_NOT_FOUND
        return false;

    return true;
}

// --------------------------------------------------------------------------
bool CMSMQWrapper::SendToDestQueue(const BYTE* pBuf, size_t nBufLen)
{
    MQMSGPROPS  MsgProps         ;
    const UINT  _nProps= 1       ;
    MSGPROPID   aPropId [_nProps];
    PROPVARIANT aVariant[_nProps];

    aPropId [0]            = PROPID_M_BODY   ; // msg to send
    aVariant[0].vt         = VT_VECTOR|VT_UI1;
    aVariant[0].caub.pElems= (BYTE*)pBuf     ;
    aVariant[0].caub.cElems= nBufLen         ;

    MsgProps.cProp         = _nProps         ; // number of props to set
    MsgProps.aPropID       = aPropId         ;
    MsgProps.aPropVar      = aVariant        ;
    MsgProps.aStatus       = 0               ;

    if (MQ_OK != ::MQSendMessage(m_hQ, &MsgProps, MQ_NO_TRANSACTION))
        return false;

    return true;
}