SQL Server 2008中Service Broker基础应用(上)

概述  

SQL Server Service Broker 用来创建用于交换消息的会话。消息在目标和发起方这两个端点之间进行交换。消息用于传输数据和触发消息收到时的处理过程。目标和发起方既可以在同一数据库引擎实例的同一数据库或不同数据库中,也可以在不同数据库引擎实例的同一数据库或不同数据库中。

 

每个 Service Broker 会话都有两个端点:会话发起方和目标。您将执行下列任务:

  • 为目标创建一个服务和队列,并为发起方创建一个服务和队列。
  • 创建请求消息类型和答复消息类型。
  • 创建约定,指定请求消息从发起方传递到目标并且答复消息从目标传递到发起方。

然后执行一个简单会话:

  • 启动会话。
  • 从发起方向目标发送一个请求。
  • 在目标处接收请求并将答复发送到发起方。
  • 在发起方处接收答复。
  • 结束会话。

对于其两端在同一 数据库引擎 实例中的会话,其消息不通过网络传输。数据库引擎 安全性和权限将限制对授权主体的访问。此方案不需要网络加密。

SQL Server 2008中SQL应用系列--目录索引

一、创建会话对象

导读:本文主要涉及Service Broker的基本概念及建立一个Service Broker应用程序的基本步骤。

1.启用Service Broker

----创建数据库
IF NOT EXISTS(SELECT * FROM SYS.DATABASES WHERE name='Dsend')
BEGIN
     CREATE DATABASE Dsend;

END

USE master;
GO
---开启数据库BROKER
ALTER DATABASE Dsend  SET ENABLE_BROKER;
GO

SELECT is_broker_enabled FROM SYS.DATABASES WHERE NAME='Dsend'

USE Dsend;
GO

一、前言

2.创建消息类型

由于经常在多个数据库引擎实例间引用 Service Broker 对象,因而大多数 Service Broker 对象的名称都是 URI 格式的。这有助于确保它们在多台计算机上是唯一的。这两种消息类型都指定 Service Broker 将只验证消息是否是格式正确的 XML 文档,并且指定 Service Broker 将不按照特定架构验证 XML。

CREATE MESSAGE TYPE
       [//Dsend/test/RequestMessage]
       VALIDATION = WELL_FORMED_XML;
CREATE MESSAGE TYPE
       [//Dsend/test/ReplyMessage]
       VALIDATION = WELL_FORMED_XML;

创建请求消息和答复消息,并且消息的格式为XML格式。

注意:Service Broker 验证传入消息。如果消息包含的消息正文与指定的验证类型不符,则 Service Broker 将放弃此无效消息,并向发送此消息的服务返回一条错误消息。会话双方必须定义相同的消息类型名称。为便于排除故障,尽管 Service Broker 不要求会话双方使用相同的验证,但通常会话双方还是会为消息类型指定相同的验证。消息类型不能是临时对象。允许使用以 # 开头的消息类型名称,但它们是永久对象。

金沙官网线上,     Service Broker为SQL Server提供消息队列,这提供了从数据库中发送异步事务性消息队列的方法。Service Broker消息可以保证以适当的顺序或原始的发送顺序不重复地一次性接收。并且因为内建在SQL Server中,这些消息在数据库发生故障时是可以恢复的,也可以随数据库一起备份。在SQL Server 2008中,还引入了使用Create Broker Priority命令对会话设定优先级,可以对重要的或不重要的会话进行优先级设定,以保证消息合理地处理。

3.创建约定

约定用于定义在 Service Broker 会话中所使用的消息类型,还用于确定会话的哪一端可以发送该类型的消息。每个会话都要遵循一个约定。当会话开始时,启动服务为会话指定约定。目标服务指定该目标服务将接受其会话的约定。 

/*
SENT BY INITIATOR    ----指示只有会话的发起方才能发送指定消息类型的消息。启动会话的服务称为会话的“发起方”
SENT BY TARGET       ----指示只有会话的目标才能发送指定消息类型的消息。接受由另一个服务启动的会话的服务称为会话的目标。
SENT BY ANY          ----指示发起方和目标都可以发送此类型的消息。
*/

CREATE CONTRACT [//Dsend/test/RequestContract]
      ([//Dsend/test/RequestMessage]    
       SENT BY INITIATOR,             ---约定只有发起方才能使用//Dsend/test/RequestMessage消息类型
       [//Dsend/test/ReplyMessage]
       SENT BY TARGET                 ---约定只有答复方才能使用//Dsend/test/ReplyMessage消息类型
      );

     本文假定一个在线数据库BookStore中存储了一些业务订单。我们使用Service Broker应用程序将消息发送到另一个数据库BookDistribution,该数据库是分离的应用程序调用,该应用程序控制仓库入库和出库交付, 并返回消息给BookStore。

4.创建队列

队列可以存储消息。当一条针对某项服务的消息到达时,Service Broker 会将该消息放入与该服务关联的队列中。

创建发起方和答复方的队列。

CREATE QUEUE RequestQueue WITH STATUS=ON;
CREATE QUEUE ReplyQueue WITH STATUS=ON;

注意:

1.队列可以通过SELECT 语句查询,但是不能使用INSERT、UPDATE、DELETE 或 TRUNCATE 语句来操作。只能使用在 Service Broker 会话中运行的语句(如 SEND、RECEIVE 和 END CONVERSATION)来修改队列的内容。

2.队列可能不是临时对象。因此,以 # 开头的队列名称无效。

3.通过以不可用状态创建队列,可以先准备好服务的基础结构,然后再允许在队列中接收消息。

4.如果队列中没有消息,则 Service Broker 不会停止激活存储过程。如果队列中在短时间内没有可用消息,应退出激活存储过程。

5.在 Service Broker 启动存储过程时将检查激活存储过程的权限,而不是在创建队列时检查。CREATE QUEUE 语句不验证 EXECUTE AS 子句中指定的用户是否有权限执行 PROCEDURE NAME 子句中指定的存储过程。

6.队列不可用时,Service Broker 将在数据库的传输队列中保存使用该队列的服务的消息。sys.transmission_queue 目录视图提供传输队列的视图。

 

例:创建具有多个参数的队列

以下示例在 DEFAULT 文件组中创建一个队列。该队列不可用。消息被保留在队列中,直到消息所属的会话结束。通过 ALTER QUEUE 启用队列后,该队列将启动存储过程 2008R2.dbo.expense_procedure 来处理消息。此存储过程以运行 CREATE QUEUE 语句的用户的身份执行。该队列最多启动存储过程的 10 个实例。

CREATE QUEUE ExpenseQueue
    WITH STATUS = OFF,
      RETENTION = ON,
      ACTIVATION (
          PROCEDURE_NAME = AdventureWorks2008R2.dbo.expense_procedure,
          MAX_QUEUE_READERS = 10,
          EXECUTE AS SELF )
    ON [DEFAULT] ;

    创建Service Broker应用程序大体步骤如下:

5.创建服务

Service Broker 使用服务的名称路由消息、将消息传递到数据库中的正确队列,以及强制执行会话的约定。一个服务可以同时绑定多个约束。

--1.创建要用于发起方的队列和服务。由于未指定约定名称,因而其他服务不可将此服务用作目标服务,此服务只能启动会话。
CREATE SERVICE
       [//Dsend/test/RequestService]
       ON QUEUE RequestQueue     
GO
---2.创建接答复服务
CREATE SERVICE
       [//Dsend/test/ReplyService]
       ON QUEUE ReplyQueue
       ([//Dsend/test/RequestContract]
       )
       ;
GO

注意:服务公开与其关联的约定提供的功能,以便其他服务可使用该功能。CREATE SERVICE 语句指定针对此服务的约定。一个服务只能是使用该服务指定的约定会话的目标。未指定约定的服务不会向其他服务公开任何功能。

从此服务启动的会话可使用任何约定。如果服务仅启动会话,则创建服务时可不指定约定。Service Broker 从远程服务接受新会话时,目标服务的名称决定了 Broker 在会话中放入消息的队列。

 

例:创建具有多个约定的服务

CREATE SERVICE [//Adventure-Works.com/Expenses] ON QUEUE ExpenseQueue
    ([//Adventure-Works.com/Expenses/ExpenseSubmission],
     [//Adventure-Works.com/Expenses/ExpenseProcessing]) ;

1、定义希望应用程序执行的异步任务。

5.开启回话

DECLARE @InitDlgHandle UNIQUEIDENTIFIER;
DECLARE @RequestMsg NVARCHAR(100);

BEGIN TRANSACTION;

BEGIN DIALOG @InitDlgHandle
     FROM SERVICE
      [//Dsend/test/RequestService] ---指定的服务是用于答复消息的返回地址
     TO SERVICE
      N'//Dsend/test/ReplyService'  ---指定的服务是消息发送到的地址。
     ON CONTRACT
      [//Dsend/test/RequestContract]
     WITH
         ENCRYPTION = OFF;

SELECT @InitDlgHandle;

SELECT @RequestMsg =
       N'<RequestMsg>3</RequestMsg>';

SEND ON CONVERSATION @InitDlgHandle
     MESSAGE TYPE 
     [//Dsend/test/RequestMessage]
     (@RequestMsg);

SELECT @RequestMsg AS SentRequestMsg;

COMMIT TRANSACTION;
GO

 

2、确定Service Broker的发起方服务和目标服务是否创建在同一个SQL Server实例中。如果是两个实例,实例间的通信还需要创建经过证书认证或NT安全的身份认证,并且要创建端点、路由以及对话安全模式。

6.接收方接收消息并返回消息给发送方

DECLARE @RecvReqDlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvReqMsg NVARCHAR(100);
DECLARE @RecvReqMsgName sysname;
DECLARE @Message NVARCHAR(100)

BEGIN TRANSACTION;

WAITFOR
( RECEIVE TOP(1)
    @RecvReqDlgHandle = conversation_handle,
    @RecvReqMsg = message_body,
    @RecvReqMsgName = message_type_name
  FROM ReplyQueue
), TIMEOUT 1000;

SELECT @RecvReqMsg AS ReceivedRequestMsg;
SELECT @RecvReqMsgName;


BEGIN

     -----返回接收消息确认结果到发起方

     DECLARE @ReplyMsg NVARCHAR(100);
     SELECT @ReplyMsg =
     N'<ReplyMsg>Reply Message</ReplyMsg>';

     SEND ON CONVERSATION @RecvReqDlgHandle
          MESSAGE TYPE 
          [//Dsend/test/ReplyMessage]
          (@ReplyMsg);

     SELECT @RecvReqDlgHandle;

     ---正常的流程不是在这里结束会话
     END CONVERSATION @RecvReqDlgHandle;
END

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

COMMIT TRANSACTION;
GO

例:接收数据同时插入表变量

金沙官网线上 1金沙官网线上 2

DECLARE @conversation_group_id UNIQUEIDENTIFIER ;

DECLARE @procTable TABLE(
     service_instance_id UNIQUEIDENTIFIER,
     handle UNIQUEIDENTIFIER,
     message_sequence_number BIGINT,
     service_name NVARCHAR(512),
     service_contract_name NVARCHAR(256),
     message_type_name NVARCHAR(256),
     validation NCHAR,
     message_body VARBINARY(MAX)) ;

SET @conversation_group_id = <retrieve conversation group ID from database> ;

RECEIVE TOP (1)
    conversation_group_id,
    conversation_handle,
    message_sequence_number,
    service_name,
    service_contract_name,
    message_type_name,
    validation,
    message_body
FROM ExpenseQueue
INTO @procTable
WHERE conversation_group_id = @conversation_group_id ;

View Code

 

3、如果没有启用,则在多方参与的数据库中使用Alter Database命令设置Enable_broker以及Truseworthy数据库选项。

7.发送方收到消息终止会话

DECLARE @RecvReplyMsg NVARCHAR(100);
DECLARE @RecvReplyDlgHandle UNIQUEIDENTIFIER;
DECLARE @RecvReqMsgName sysname;

BEGIN TRANSACTION;

WAITFOR
( RECEIVE TOP(1)
    @RecvReplyDlgHandle = conversation_handle,
    @RecvReplyMsg = message_body,
    @RecvReqMsgName=message_type_name
  FROM RequestQueue
), TIMEOUT 1000;

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

SELECT @RecvReplyMsg AS ReceivedReplyMsg;

COMMIT TRANSACTION;

GO

4、为所有多方参与的数据库创建数据库主密钥。

三、查询

SELECT state_desc,*
FROM sys.conversation_endpoints

SELECT message_type_name, CAST(message_body as xml) message,*
FROM dbo.RequestQueue

SELECT message_type_name, CAST(message_body as xml) message,*
FROM dbo.ReplyQueue

SELECT CAST(message_body as xml) message,* FROM sys.transmission_queue

--END CONVERSATION '236AF2C5-57F4-E711-A9E6-005056C00008';

5、创建希望在服务之间发送的消息类型。

四、删除会话对象

IF EXISTS (SELECT * FROM sys.services
           WHERE name =
           N'//AWDB/1DBSample/TargetService')
     DROP SERVICE
     [//AWDB/1DBSample/TargetService];

IF EXISTS (SELECT * FROM sys.service_queues
           WHERE name = N'TargetQueue1DB')
     DROP QUEUE TargetQueue1DB;

-- Drop the intitator queue and service if they already exist.
IF EXISTS (SELECT * FROM sys.services
           WHERE name =
           N'//AWDB/1DBSample/InitiatorService')
     DROP SERVICE
     [//AWDB/1DBSample/InitiatorService];

IF EXISTS (SELECT * FROM sys.service_queues
           WHERE name = N'InitiatorQueue1DB')
     DROP QUEUE InitiatorQueue1DB;

IF EXISTS (SELECT * FROM sys.service_contracts
           WHERE name =
           N'//AWDB/1DBSample/SampleContract')
     DROP CONTRACT
     [//AWDB/1DBSample/SampleContract];

IF EXISTS (SELECT * FROM sys.service_message_types
           WHERE name =
           N'//AWDB/1DBSample/RequestMessage')
     DROP MESSAGE TYPE
     [//AWDB/1DBSample/RequestMessage];

IF EXISTS (SELECT * FROM sys.service_message_types
           WHERE name =
           N'//AWDB/1DBSample/ReplyMessage')
     DROP MESSAGE TYPE
     [//AWDB/1DBSample/ReplyMessage];
GO

 

参考:http://www.cnblogs.com/downmoon/archive/2011/04/05/2005900.html

参考:https://docs.microsoft.com/zh-cn/sql/t-sql/statements/create-message-type-transact-sql

6、创建契约(Contract)来定义可以由发起方发送的各种消息以及由目标发送的消息类型的种类。

总结

使用单数据库会话可以处理一般的队列发送和读写消息的场景,其中sys.conversation_endpoints系统视图需要重点关注。

 

 

 

 

 

备注:

    作者:pursuer.chen

    博客:http://www.cnblogs.com/chenmh

本站点所有随笔都是原创,欢迎大家转载;但转载时必须注明文章来源,且在文章开头明显处给明链接,否则保留追究责任的权利。

《欢迎交流讨论》

 

7、同时在两方参与的数据库中创建用于保存消息的队列。

8、同时在绑定特定约定到特定队列的多方参与的数据库中创建服务。

二、实例

    下面我们通过一个示例来实现以上步骤:

(一)、启用数据库的Service Broker活动

-- Enabling Databases for Service Broker Activity

USE master
GO
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name ='BookStore')
CREATE DATABASE BookStore
GO
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name ='BookDistribution')
CREATE DATABASE BookDistribution
GO
ALTER DATABASE BookStore SET ENABLE_BROKER
GO
ALTER DATABASE BookStore SET TRUSTWORTHY ON
GO
ALTER DATABASE BookDistribution SET ENABLE_BROKER
GO
ALTER DATABASE BookDistribution SET TRUSTWORTHY ON

 

(二)、创建数据库主密钥

-- Creating the DatabaseMaster Key for Encryption

USE BookStore
GO
CREATE MASTER KEY
ENCRYPTION BY PASSWORD ='I5Q7w1d3'
GO

USE BookDistribution
GO
CREATE MASTER KEY
ENCRYPTION BY PASSWORD ='D1J3q5z8X6y4'
GO

 

(三)、管理消息类型

使用CREATE MESSAGE TYPE(http://msdn.microsoft.com/en-us/library/ms187744.aspx)命令,

-- Managing Message Types

Use BookStore
GO
-- 发送图书订单的消息类型
CREATE MESSAGE TYPE [//SackConsulting/SendBookOrder]
VALIDATION = WELL_FORMED_XML
GO

--目标数据库发送的消息类型
CREATE MESSAGE TYPE [//SackConsulting/BookOrderReceived]
VALIDATION = WELL_FORMED_XML
GO

--执行同样的定义
Use BookDistribution
GO
-- 发送图书订单的消息类型
CREATE MESSAGE TYPE [//SackConsulting/SendBookOrder]
VALIDATION = WELL_FORMED_XML
GO

--目标数据库发送的消息类型
CREATE MESSAGE TYPE [//SackConsulting/BookOrderReceived]
VALIDATION = WELL_FORMED_XML
GO

 

--注意,此处没有定义消息的内容。实际的消息是消息类型的实例。

(四)、创建契约(Contract)

使用Create Contract(http://msdn.microsoft.com/en-us/library/ms178528.aspx

-- Creating Contracts

Use BookStore
GO
CREATE CONTRACT
[//SackConsulting/BookOrderContract]
( [//SackConsulting/SendBookOrder]
SENT BY INITIATOR,
[//SackConsulting/BookOrderReceived]
SENT BY TARGET
)
GO

USE BookDistribution
GO
CREATE CONTRACT
[//SackConsulting/BookOrderContract]
( [//SackConsulting/SendBookOrder]
SENT BY INITIATOR,
[//SackConsulting/BookOrderReceived]
SENT BY TARGET
)
GO

 

--发起方和目标的定义必须相同

(五)、创建队列

队列用来保存数据。使用命令Create queue(http://msdn.microsoft.com/en-us/library/ms190495.aspx

-- Creating Queues

Use BookStore
GO
--保存BookDistribution过来的消息
CREATE QUEUE BookStoreQueue
WITH STATUS=ON
GO

USE BookDistribution
GO
--保存BookStore过来的消息
CREATE QUEUE BookDistributionQueue
WITH STATUS=ON
GO

 

(六)、创建服务

服务定义端点,然后使用它来将消息队列绑定到一个或多个契约上。服务使用队列和契约来定义一个或一组任务。有点拗口,是不是?

服务是消息的发起方和接收方强制约定的规则,并将消息路由到正确的序列。

使用Create Service(http://msdn.microsoft.com/en-us/library/ms190332.aspx)命令。

本文由金沙官网线上发布于数据库,转载请注明出处:SQL Server 2008中Service Broker基础应用(上)

您可能还会对下面的文章感兴趣: