sql-server - 使用 SSB 在对话中发送和接收多条消息

Question

有时我需要将我的一些消息组合在一起，以便我的工作进程可以按顺序处理它们。

但是我发现的每一个例子都只显示了如何在每次对话中发送一条消息。

我正在寻找的是一个在对话中发送多个消息的示例，以及一个显示如何处理它们的示例。（我可能会同时发送它们，但我似乎只能将其中一个取出。）

Answer 1

对话提供了按顺序处理消息的边界 - 因此您需要发送组中具有相同 ConversationId 的所有消息。我这样做的方法是使用一个实用程序表，该表在创建时存储 ConversationId，以便每次发送消息时，它都会查找适当的会话 ID 以将其发送。

SELECT @conversationHandle = ConversationHandle FROM Qproc.SessionConversation
WHERE 
FromService = @fromService
AND ToService = @toService
AND OnContract = @onContract
AND Terminated IS NULL

IF @conversationHandle IS NULL
BEGIN
    BEGIN DIALOG CONVERSATION @conversationHandle
        FROM SERVICE @fromService
        TO SERVICE @toService
        ON CONTRACT @onContract
        WITH ENCRYPTION = OFF; --, LIFETIME = 60*60*24*100;

    -- Store the ongoing conversation for further use
    INSERT INTO QProc.SessionConversation ( FromService, ToService, OnContract,ConversationHandle)
    VALUES(     @fromService,   @toService, @onContract,    @conversationHandle )
END


    -- Create the dialog timer, timeout is seconds; this will notify the ClientQueue if nothing has happened on the conversation
    --in the timeout period
    BEGIN CONVERSATION TIMER (@conversationHandle) TIMEOUT = 60*8;
    SEND ON CONVERSATION @conversationHandle
    MESSAGE TYPE [http://COMPANYNAME/AsyncTriggerRequestMesssage]
        (@messageBody);

您在另一端只看到一条消息的原因与会话组锁定有关 - 您应该阅读此内容以了解发生了什么，但基本上一旦消息处理过程看到一条消息，它在消息队列上的视图仅限于一次对话。一旦您重复使用相同的对话 ID，这将不是问题。这是一个接收示例：

DECLARE @RecvReqDlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvReqMsg VARCHAR(8000);
DECLARE @RecvReqMsgName sysname;


WHILE (1=1)
    BEGIN
        BEGIN TRANSACTION;

        WAITFOR
        ( RECEIVE TOP(1)
            @RecvReqDlgHandle = conversation_handle,
            @RecvReqMsg = message_body,
            @RecvReqMsgName = message_type_name
          FROM  QProc.AsyncTaskServiceQueue
        ), TIMEOUT 500;

        IF @@ROWCOUNT=0
        BEGIN
            ROLLBACK TRANSACTION;
            BREAK
        END


        IF @RecvReqMsgName = 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog' 
            BEGIN
            END CONVERSATION @RecvReqDlgHandle;
            END

        IF @RecvReqMsgName='http://COMPANYNAME/AsyncTriggerRequestMesssage'
            BEGIN
                DECLARE @BodyDoc XML;
                SET @BodyDoc=CONVERT(XML, @RecvReqMsg)  ;   
                EXEC QProc.AsyncTaskRunTask @BodyDoc;

            END 

        COMMIT TRANSACTION;

    END

最后，一旦不再使用这些对话，您就需要清理它们，如下所示：

DECLARE @conversationHandle UNIQUEIDENTIFIER;
DECLARE @messageTypeName SYSNAME;

BEGIN TRANSACTION;

RECEIVE TOP(1) 
    @conversationHandle = conversation_handle,
    @messageTypeName = message_type_name
FROM QProc.AsyncTaskClientQueue;

IF @conversationHandle IS NOT NULL
BEGIN
    --If the DialogTimer message arrives, then there has been no activity on this conversation for a while (see timeout setting in [QProc].[DispatchAsyncTaskMessage])
    --so we terminate gracefully and go home.


    IF @messageTypeName = 'http://schemas.microsoft.com/SQL/ServiceBroker/DialogTimer'
    OR @messageTypeName = 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog'
    BEGIN
        END CONVERSATION @conversationHandle;
        UPDATE Qproc.SessionConversation SET TERMINATED = getUtcDate() WHERE ConversationHandle = @conversationHandle;

    END
    END

COMMIT TRANSACTION;