有时我需要将我的一些消息组合在一起,以便我的工作进程可以按顺序处理它们。
但是我发现的每一个例子都只显示了如何在每次对话中发送一条消息。
我正在寻找的是一个在对话中发送多个消息的示例,以及一个显示如何处理它们的示例。(我可能会同时发送它们,但我似乎只能将其中一个取出。)
有时我需要将我的一些消息组合在一起,以便我的工作进程可以按顺序处理它们。
但是我发现的每一个例子都只显示了如何在每次对话中发送一条消息。
我正在寻找的是一个在对话中发送多个消息的示例,以及一个显示如何处理它们的示例。(我可能会同时发送它们,但我似乎只能将其中一个取出。)
对话提供了按顺序处理消息的边界 - 因此您需要发送组中具有相同 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;