Использование SSB для отправки и получения нескольких сообщений в разговоре

Question

Есть случаи, когда мне нужно сгруппировать некоторые из моих сообщений, чтобы они могли обрабатываться в порядке моих рабочих процессов.

Но каждый отдельный пример, который я нахожу, показывает только, как сделать одно сообщение для разговора.

Что я ищу, это пример, который отправляет несколько сообщений в цепочке и пример, который показывает, как их обрабатывать. (Я могу отправить их обоим, но мне кажется, что я могу их вернуть).

Answer 1

Беседа обеспечивает границу, в пределах которой обрабатываются сообщения в порядке - поэтому вам нужно ОТПРАВИТЬ все сообщения в группе с тем же диалоговым интерфейсом. То, как я это делаю, - это таблица служебных программ, в которой хранится ConversationId при создании, так что каждый раз, когда отправляется сообщение, он просматривает соответствующий разговорник, чтобы отправить его.

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); 

Причины вы видите только одно сообщение на другом конце делать с разговором групповой блокировкой - вы должны прочитать об этом, чтобы понять, что происходит, но в основном, как только процедура обработки сообщений видели сообщение , его представление о очереди сообщений ограничивается одной беседой. Это не будет проблемой после повторного использования одного и того же идентификатора беседы. Вот пример получим:

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;