基于Service Broker的异步消息传递

这里演示同一个SQL Server中不同数据库之间的基于Service Broker的异步消息传递，其中Stored Procedure充当Service Program。HelloWorldDB为目标数据库，DotNetFun2则为消息发送发的数据库。

 

同时，假设Server Broker的基本对象类型已经创建，如MessageType(XMLMessage), Contract(XMLContract), Queue(SendingQueue and ReceivingQueue)等等，具体操作可以参考《A simple tutorial on SQL Server 2005 Beta 2 Service Broker》。另外，因为在不同的Databases之间进行消息传递，因此需要创建Route，具体操作可以参考《SQL Server 2005 Beta 2 Service Broker: Create Route》。

 

1.创建Stored Procedure作为Internal Service Program.

 

USE HelloWorldDB
            GO
            Alter Procedure HelloWorldResponder
            As
            Begin
            Declare @conversationHandle UNIQUEIDENTIFIER
            Declare @message_body nvarchar(MAX)
            Declare @message_type_name SYSNAME
            WHILE (1=1)
            BEGIN
            BEGIN TRANSACTION
            -- Wait for 1 seconds for messages to arrive
            WAITFOR (
            -- For simplicity we process one message at a time
            RECEIVE TOP(1)
            @message_type_name=message_type_name,
            @conversationHandle=conversation_handle,
            @message_body=message_body
            FROM [ReceivingQueue]), TIMEOUT 1000
            -- If a message was received, process it, else skip
            IF (@@rowcount <= 0)
            BREAK;
            -- If this is a XML message,
            -- respond with an appropriate greeting
            IF @message_type_name = 'XMLMessage'
            BEGIN
            SEND ON CONVERSATION @conversationHandle
            MESSAGE TYPE XMLMessage
            ('<hello>Hello From Rickie</hello>')
            END CONVERSATION @conversationHandle
            END
            COMMIT
            END
            COMMIT
            END
            GO

 

 

 

该Stored Procedure负责从ReceivingQueue中检索消息，并根据Queue的Retention设置，来确定从Queue中移除消息或更新Queue中消息的状态。

 

2.设置目标队列（Target Queue）的激活机制

 

Use HelloWorldDB
            go
            ALTER QUEUE [ReceivingQueue] WITH
            ACTIVATION (
            STATUS = ON, -- Turn on internal activation
            PROCEDURE_NAME = [HelloWorldResponder], -- Our stored proc
            MAX_QUEUE_READERS = 4, -- Up to 4 concurrent readers
            EXECUTE AS SELF)  -- Execute as user of incoming dialog

 

设置上述创建的Stored Procedure，该Stored Procedure将被激活并处理Queue中的消息。

 

3.在Initiator端发送消息

 

 

Use DotNetFun2
            go
            DECLARE @conversationHandle uniqueidentifier
            BEGIN TRANSACTION
            -- Begin a dialog to the Hello World Service
            BEGIN DIALOG  @conversationHandle
            FROM SERVICE    [SendingService]
            TO SERVICE      'ReceivingService','a727462b-52e7-4405-9eee-d19923729790'
            ON CONTRACT     [XMLContract]
            WITH ENCRYPTION = OFF, LIFETIME = 600;
            -- Send message
            SEND ON CONVERSATION @conversationHandle
            MESSAGE TYPE [XMLMessage]
            ('<hello>Welcome to Rickie Lee's blog, www.cnblogs.com/rickie</hello>');
            Select * From sys.conversation_endpoints
            COMMIT

 

 

其中，TO SERVICE 'ReceivingService','a727462b-52e7-4405-9eee-d19923729790'，’ReceivingSerice’表示目标Service名称，'a727462b-52e7-4405-9eee-d19923729790'则指定目标Service所在的数据库，可以通过如下SQL Script获取：

 

 

-- Retrieve remote broker instance guid
            SELECT service_broker_guid
            FROM sys.databases
            WHERE database_id = DB_ID('HelloWorldDB')

 

 

另外，可以通过如下的SQL script来检测Initiator端收到的Reply消息：

 

Select cast(message_body as XML) From SendingQueue
            Receive message_type_name,
            cast(message_body as XML)
            From SendingQueue

 

4.查询对话端点状态（State of Conversation Endpoints）

 

最后，可以通过在Target/Initiator端查询sys.conversation_endpoints表，获取Dialog对话状态：

Select * From sys.conversation_endpoints