20. SQL -- 队列服务
service broker(队列服务)[*] 概述：
Service Broker 为SQL Server 提供消息队列，这提供了从数据库中发送异步事务性消息队列的方法。Service Broker 消息可以保证以适当的顺序或原始的发送顺序不重复地一次性接收。并且因为内建在SQL Server 中，这些消息在数据库发生故障时是可以恢复的，也可以随数据库一起备份。在SQL Server 2008 中，还引入了使用Create Broker Priority
命令对会话设定优先级，可以对重要的或不重要的会话进行优先级设定，以保证消息合理地处理。
Service Broker 使用称为“对话协议”的可靠消息传送协议来确保发送到远程队列的消息按顺序到达并且仅到达一次。正如对话是双向会话一样，对话协议同时支持双向的消息传送。对话消息具有标题，可确保消息按正确顺序安全地传送到正确的目标位置。它包含序列号、消息所在对话的标识符、要发送到的服务的名称、安全性信息以及用于控制消息传
送的一些其他信息。当目标位置成功接收到消息后，它将发出确认，以便源位置知道已成功传送了消息。如果可能，确认信息将包含在另一条消息的标题中发送回源位置，以尽可能减小消息的数量。如果源位置在某一时限内没有收到确认，将重新发送该消息直到成功传送。
消息传送系统在传送较大的消息时经常会遇到问题。发送以 GB 为单位的消息会花几分钟时间，从而会占用一段时间的网络连接。如果发生网络错误导致重新发送消息多次，将会严重影响网络性能。为了解决此问题，Service Broker 对话协议将大型消息拆分成多个较小的片段，然后再单独发送每个片段。如果发生网络错误导致重新发送，则只重新发
送传送失败的消息片断。因此，ServiceBroker 最大可以支持2GB 的消息，而许多可靠的消息传送系统只能发送 100MB 或更小的消息。
[*] 事务性消息传送
“仅一次”消息处理需要使用事务性消息。为了说明此问题，假定某个应用程序在处理消息的过程中发生了错误。当应用程序重新启动时，它如何才能知道是否要处理发生错误时正在处理的消息呢？数据库中可能已经更新了消息处理的结果，因此重新处理消息可能产生重复的数据。唯一安全的处理方式是使接收消息成为更新数据库的同一事务的一部分。
在系统崩溃时数据库更新和消息接收都回滚，因此数据库和消息队列的状态与系统崩溃前的状态相同。因为所有 Service Broker 操作都在数据库事务环境中发生，所以保证了消息传送操作的事务完整性。典型的Service Broker 消息处理事务包括以下步骤：
○1 、开始事务。
○2 、从会话组中接收一条或多条消息。
○3 、从状态表中检索会话的状态。
○4 、处理消息并根据消息内容对应用程序数据进行一项或多项更新。
○5 、发送一些 Service Broker 消息：即，将响应发送到传入的消息或将消息发送到处理传入消息所需的其他服务。
○6 、如果此会话组包含其他消息，则读取和处理此会话组中的其他消息。
○7 、使用新的会话状态更新会话状态表。
○8 、提交事务
Service Broker 事务性消息传送的一项重要功能是在任何时间如果崩溃或应用程序发生错误，事务将回滚并且一切都返回到事务开始时的状态，即状态不变、应用程序数据不变、消息未发送并且接收的消息返回到队列中。这使此类应用程序中的错误处理非常简单。
[*] 队列读取器
在 Service Broker 应用程序的消息处理过程中，队列读取器首先从队列接收消息。由于消息始终从队列中拉出，因此在消息到达队列时接收应用程序必须正在运行。许多异步消息应用程序都面临着一个问题，即如何确保队列读取器在需要时处于运行状态？有两种传统的方法：让队列读取器成为连续运行的服务，或者使用触发器，在每条消息到达时，
都将触发消息传送系统。Windows NT 服务方法是指即使在不处理消息时也运行应用程序。由于队列读取器频繁开始和停止，触发器方法可能会遇到性能问题。
Service Broker 采用称为激活的中间状态方法来管理队列读取器。为设置激活，DBA将存储过程与 Service Broker 队列相关联。当第一条消息到达队列时，激活逻辑将启动指定的存储过程。存储过程负责接收和处理消息，直到队列为空。队列为空后，存储过程
可以终止以节省资源。
如果 Service Broker 确定消息添加到队列的速度比存储过程能够处理的速度快，激活逻辑将开始其他的存储过程，直到存储过程能够处理传入的消息或者达到了为队列配置的存储过程的最大数量。由于为队列提供服务的队列读取器数量随着传入消息速率的更改增大或减小，因此所有时间都将运行适当数量的队列读取器。
[*] 在数据库中使用消息传送的原因：
消息和数据的单客户端连接。除了上一部分中提到的统一编程模型，还具有一些其他显著的优势：
· 当在任何可以连接到数据库的客户端上运行时，应用程序可以通过事务接收消息。当接收程序与队列在同一计算机上运行时，许多消息传送系统只允许事务性接收。
· 与在数据库中不存储消息的消息系统不同，事务性消息传送不需要分布式事务或两阶段提交。
数据和消息之间集成的管理、部署和操作。用于保护和管理数据库数据的所有工具和技术同样适用于消息：
· 备份和恢复数据库还可以备份和恢复消息队列。
· 如果使用群集或数据库镜像来保护数据库以避免出现故障，则您的消息享有相同的保护。
· 由于队列具有相关的视图，因此确定队列中的操作非常简单。要了解队列中有多少条消息吗？请从队列中选择 count(*)。要了解有多少条消息尚未传送吗？请从 sys.transmission_queue中选择*。如果消息中包含 XML 数据，则可以使用XQuery 进行搜索。可以使队列中的消息与状态信息甚至数据表相结合，以确定订单输入系统中特定订单的完成状态。
将消息传送功能内置在数据库中还有一些显著的性能优势。
· 正如前面所述，事务性消息不要求两阶段提交。
· 消息更新、状态更改和数据更新都记录在同一事务日志中，因此在提交事务时只需写入单个日志。
· 可靠的消息传送通常将消息从传送队列传送到接收队列。如果 Service Broker检测到接收队列与传送队列位于同一数据库实例中，则消息直接放入接收队列中，从而节省了额外的输入/输出并减少了事务提交的次数。
[*] Service Broker 编程使用的对象：
Service Broker 功能通过 SQL Server 中的新对象启用，这些新对象可以由一组T-SQL 扩展来创建和操作。为了数据库程序员的方便，使用了他们熟悉的用于配置其他数据库对象的 CREATE、ALTER 和 DROP DDL 语句来配置 Service Broker 应用程序。用于创建 Service Broker 对话以及在对话中发送和接收消息的命令是 Transact SQL 语言的 DML 扩展。接收命令的语法与选择命令相似，它返回包含消息的行集，就像选择命令返回包含行的行集一样。用于为Service Broker 编程的客户端 API 与用于为所有数据库编程的 API 相同，例如 OLEDB、ODBC、ADO、ADO.NET 等等。下面是 Service Broker 使用的对象：
队列
Service Broker 使用队列在消息发送程序和消息接收程序之间提供松散耦合。发送程序可以使用 SEND 命令将消息放到队列中，然后应用程序继续操作并依靠 ServiceBroker 来确保消息到达其目标位置。
队列允许较大的计划灵活性。例如，发送程序可以发送多条消息以供多个接收程序并行处理。接收程序可能在消息发送很长时间后才处理消息，但由于传入消息进行了排队，因此接收程序可以按其自己的速率处理消息，而且发送程序无须等待接收程序完成处理便可继续操作。
对话
Service Broker 还实现了对话，对话是两个端点之间的双向消息流。对话中的所有消息都进行了排序，而且对话消息总是按照发送的顺序传送。该顺序在事务、输入线程、输出线程以及系统崩溃和重新启动过程中都保持不变。有些消息系统可以确保单个事务中发送或接收的消息顺序，但不能确保多个事务中的顺序，因此Service Broker 对话具有
独特的优势。
对话是一种会话类型。Service Broker 会话是持久可靠的通信通道。在以后的SQL Server 版本中，Service Broker 将包括一对多的单向会话，也称为可靠的“发布-订阅”。在 SQL Server 2005 中，对话是唯一的会话类型，因此这两个术语同义。每条消息都包括唯一标识与它相关的对话的会话句柄。例如，某个订单输入应用程序可能同时使用发货应用程序、库存应用程序和帐单应用程序打开了对话。因为每个应用程序中的消息都具有唯一的会话句柄，所以可以轻松地确定发送每条消息的应用程序。
消息类型
当前，所有 Service Broker 消息都与特定的消息类型相关联。消息类型是与消息一起传送的标签，因此接收消息的应用程序可以确定所接收消息的类型。此外，如果消息包含XML 文档，则消息类型可以与 XML 架构集合相关联。如果为某个消息类型指定了架构集合，则所接收的该类型消息将在收到时根据架构集合进行验证，而没有通过架构验证的消息将会被拒绝。
规范
Service Broker 规范是一个消息类型集合。一个对话总是与一个规范相关联，而规范定义哪些消息类型可以通过对话发送。
服务
Service Broker 服务将一个或多个规范与一个队列相关联。规范定义可以将哪些消息类型发送到队列。服务名称用于建立对话的端点。服务名称用作实际队列的别名，因此您可以编写引用服务名称的Service Broker 程序，然后在部署应用程序时将它与实际队列相关联。
[*] 创建Service Broker
创建Service Broker 应用程序大体步骤如下：
○1 、定义希望应用程序执行的异步任务。
学习是不断积累及重复学习的过程，当学习变成一种习惯的时候，就真正进入了学习的殿堂！
第 475 页 共 588 页
○2 、确定Service Broker 的发起方服务和目标服务是否创建在同一个SQL Server 实例中。如果是两个实例，实例间的通信还需要创建经过证书认证或NT 安全的身份认证，并且要创建端点、路由以及对话安全模式。
○3 、如果没有启用，则在多方参与的数据库中使用Alter Database命令设置Enable_broker以及Truseworthy 数据库选项。
○4 、为所有多方参与的数据库创建数据库主密钥。
○5 、创建希望在服务之间发送的消息类型。
○6 、创建契约（Contract）来定义可以由发起方发送的各种消息以及由目标发送的消息类型的种类。
○7 、同时在两方参与的数据库中创建用于保存消息的队列。
○8 、同时在绑定特定约定到特定队列的多方参与的数据库中创建服务。
DEMO1：
--1、启用数据库的Service Broker 活动
--Enabling Databases for Service Broker Activity
USEmaster
GO
IF NOT EXISTS (SELECT name
FROM sys.databases
WHEREname = 'BookStore')
CREATEDATABASE BookStore
GO
IF NOT EXISTS (SELECT name
FROM sys.databases
WHEREname = 'BookDistribution')
CREATEDATABASE BookDistribution
GO
ALTERDATABASE BookStore SET ENABLE_BROKER
GO
ALTERDATABASE BookStore SET TRUSTWORTHY ON
GO
ALTERDATABASE BookDistributionSET ENABLE_BROKER
GO
ALTERDATABASE BookDistributionSET TRUSTWORTHY ON
--2、创建数据库主密钥
--Creating the DatabaseMaster Key for Encryption
USE BookStore
GO
CREATEMASTER KEY
ENCRYPTIONBY PASSWORD = 'I5Q7w1d3'
GO
USE BookDistribution
GO
CREATEMASTER KEY
ENCRYPTIONBY PASSWORD = 'D1J3q5z8X6y4'
GO
--3、管理消息类型
使用CREATE MESSAGE TYPE
--Managing Message Types
Use BookStore
GO
--发送图书订单的消息类型
CREATEMESSAGE TYPE [//SackConsulting/SendBookOrder]
VALIDATION= WELL_FORMED_XML
GO
--目标数据库发送的消息类型
CREATEMESSAGE TYPE [//SackConsulting/BookOrderReceived]
VALIDATION= WELL_FORMED_XML
GO
--执行同样的定义
Use BookDistribution
GO
--发送图书订单的消息类型
CREATEMESSAGE TYPE [//SackConsulting/SendBookOrder]
VALIDATION= WELL_FORMED_XML
GO
--目标数据库发送的消息类型
CREATEMESSAGE TYPE [//SackConsulting/BookOrderReceived]
VALIDATION= WELL_FORMED_XML
GO
--4、创建契约（Contract）
使用Create Contract
--Creating Contracts
Use BookStore
GO
CREATECONTRACT
[//SackConsulting/BookOrderContract]
( [//SackConsulting/SendBookOrder]
SENTBY INITIATOR,
[//SackConsulting/BookOrderReceived]
SENTBY TARGET
)
GO
USE BookDistribution
GO
CREATECONTRACT
[//SackConsulting/BookOrderContract]
( [//SackConsulting/SendBookOrder]
SENTBY INITIATOR,
[//SackConsulting/BookOrderReceived]
SENTBY TARGET
)
GO
--发起方和目标的定义必须相同
--5、创建队列
队列用来保存数据。使用命令Create queue
--Creating Queues
Use BookStore
GO
--保存BookDistribution过来的消息
CREATEQUEUE BookStoreQueue
WITHSTATUS=ON
GO
USE BookDistribution
GO
--保存BookStore过来的消息
CREATEQUEUE BookDistributionQueue
WITHSTATUS=ON
GO
--6、创建服务
服务定义端点，然后使用它来将消息队列绑定到一个或多个契约上。服务使用队列和契约来定义一个或一组任务。服务是消息的发起方和接收方强制约定的规则，并将消息路由到正确的序列。使用CreateService
--Creating Services
Use BookStore
GO
CREATESERVICE
[//SackConsulting/BookOrderService]
ONQUEUE dbo.BookStoreQueue --指定的队列绑定到契约
([//SackConsulting/BookOrderContract])
GO
USE BookDistribution
GO
CREATESERVICE
[//SackConsulting/BookDistributionService]
ONQUEUE dbo.BookDistributionQueue --指定的队列绑定到契约
([//SackConsulting/BookOrderContract]
)
GO
--7、启动对话
对话会话（dialog conservation）是在服务之间进行消息交换的操作。
使用Begin Dialog Conversation 命令创建新的会话。使用Send 来发送消息。使用EndConversation命令结束会话
--Initiating a Dialog
Use BookStore
GO
--保存会话句柄和订单信息
DECLARE@Conv_Handler uniqueidentifier
DECLARE@OrderMsg xml
BEGINDIALOG CONVERSATION @Conv_Handler --创建会话
FROMSERVICE [//SackConsulting/BookOrderService]
TOSERVICE '//SackConsulting/BookDistributionService'
ONCONTRACT [//SackConsulting/BookOrderContract]
SET @OrderMsg =
'
';
SENDON CONVERSATION @Conv_Handler--发送到BookDistribution数据库的队列中
MESSAGETYPE [//SackConsulting/SendBookOrder]
(@OrderMsg)
--8、查询队列中传入的消息
--Querying the Queue for IncomingMessages
USE BookDistribution
GO
SELECTmessage_type_name, CAST(message_body as xml) message,
queuing_order,
conversation_handle,
conversation_group_id
FROM dbo.BookDistributionQueue
返回：
--9、检索并响应消息
使用Receive 语句从队列中读取行（消息），也可以删除已经读取的消息。Receive 的结果可以填充到常规表中，也可以在局部变量中执行其他操作，或发送到其他serviceBroker 消息。如果消息是XML 数据类型的消息，则可以直接借助TSQL 的XQuery 来操作。
--Receiving and Responding to aMessage
USE BookDistribution
GO
--创建一个表存放接收到的订单信息
CREATETABLE dbo.BookOrderReceived
(
BookOrderReceivedIDint> conversation_handleuniqueidentifier NOTNULL,
conversation_group_iduniqueidentifier NOTNULL,
message_bodyxml NOT NULL
)
GO
--声明变量
DECLARE@Conv_Handler uniqueidentifier
DECLARE@Conv_Group uniqueidentifier
DECLARE@OrderMsg xml
DECLARE@TextResponseMsgvarchar(8000)
DECLARE@ResponseMsg xml
DECLARE@OrderID int;
--从队列中获取消息，将接收值赋于局部变量
RECEIVETOP(1) @OrderMsg = message_body,
--TOP指定最多一条消息
@Conv_Handler= conversation_handle,
@Conv_Group= conversation_group_id
FROM dbo.BookDistributionQueue;
--将变量值插入表中
INSERTdbo.BookOrderReceived
(conversation_handle,
conversation_group_id,
message_body)
VALUES(@Conv_Handler,@Conv_Group, @OrderMsg )
--使用XQuery进行抽取以响应消息订单
SELECT@OrderID = @OrderMsg.value('(/order/@id)', 'int' )
SELECT@TextResponseMsg=
'';
SELECT@ResponseMsg = CAST(@TextResponseMsg as xml);
--使用既有的会话句柄，发送响应消息到发起方
SENDON CONVERSATION @Conv_Handler
MESSAGETYPE [//SackConsulting/BookOrderReceived]
--10、结束会话
--Ending a Conversation
USE BookStore
GO
--创建订单确认表
CREATETABLE dbo.BookOrderConfirmation
(BookOrderConfirmationID int> conversation_handleuniqueidentifier NOTNULL,
DateReceiveddatetime NOT NULL DEFAULT GETDATE(),
message_bodyxml NOT NULL)
DECLARE@Conv_Handler uniqueidentifier
DECLARE@Conv_Group uniqueidentifier
DECLARE@OrderMsg xml
DECLARE@TextResponseMsgvarchar(8000);
RECEIVETOP(1) @Conv_Handler = conversation_handle,
@OrderMsg= message_body
FROM dbo.BookStoreQueue
INSERTdbo.BookOrderConfirmation
(conversation_handle, message_body)
VALUES(@Conv_Handler,@OrderMsg );
ENDCONVERSATION @Conv_Handler;
GO
USE BookDistribution
GO
DECLARE@Conv_Handler uniqueidentifier
DECLARE@Conv_Group uniqueidentifier
DECLARE@OrderMsg xml
DECLARE@message_type_namenvarchar(256);
RECEIVETOP(1) @Conv_Handler = conversation_handle,
@OrderMsg= message_body,
@message_type_name= message_type_name
FROM dbo.BookDistributionQueue
--双方必须都结束会话
IF
@message_type_name= 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog'
BEGIN
ENDCONVERSATION @Conv_Handler;
END
--查询会话状态
SELECTstate_desc, conversation_handle
FROM sys.conversation_endpoints
返回：
DEMO2：
--1、创建测试数据库
usemaster
go
createdatabase brokerdemo
go
use brokerdemo
go
--2、检查Server Service Broker 的状态：=没有启用,1=已经启用
selectis_broker_enabled
from sys.databases
wheredatabase_id=db_id('brokerdemo')
--若没有启用，则用以下命令启用Service Broker
alterdatabase brokerdemo
setenable_broker
--3、定义消息类型（Message Type）,创建一个名为msgpostcard的消息类型,并定义它的验证方式,如果
要收发消息，必须要去定义消息类型
createMessage Type msgpostcard
validation= WELL_FORMED_XML
go
--4、定义合约（Contracts）
CREATECONTRACT ctrmail
( msgpostcard SENT BY INITIATOR)
go
--5、定义队列（Qeueus）,需要定义两个队列，一个用来发消息，另一个用来收消息
CREATEQUEUE qpostoffice
CreateQUEUE qmailbox
go
--6、定义服务（Service），在QUEUE之上定义service
createservice svcsender on queue qpostoffice
createservice svcreceiver
onqueue qmailbox (ctrmail)--需要一个约定进行通知
--7、定义服务的程序逻辑性（Service program logic）
begintran--定义一个事务，在事务中要定义一个对话，两个之间的沟通
declare@dialog uniqueidentifier--声明一个对话的变量,收发消息的时候必须要打开这个dialog
--接下来定义对话的相关属性
begindialog conversation @dialog--begin 一个dialog 它是一个会话
fromservice svcsender--从svcsende这个service里面去调用
toservice 'svcreceiver'--
oncontract ctrmail--基于ctrmail约定
withencryption=off--不加密
declare@message xml--定义message类型为XML并为它赋值寄送一张明信片
set @message =N'你好';
sendon conversation @dialog
--定义消息在收发的时候通过以前的会话@dialog去告知serviceBroker谁发给谁
messagetype msgpostcard (@message)--引用前面定义的消息类型
select@dialog=conversation_handle from sys.conversation_endpoints
wherefar_service='svcreceiver'--进行查询调用系统表中的一个handle
;send on conversation @dialog--再写一个收到message之后回复的消息
messagetype msgpostcard
(convert(varbinary(max),N'很高兴看到你'))
endconversation @dialog
--结束会话,在这个会话中，我们所做的就是一个对话，在两者之间传送消息
committran--提交事务
--下面可以查看到消息的内容，message_body需要转换成XML类型
select*,convert(xml,message_body) as body内容 from qmailbox
--可以查看所发的消息的内容,应该是看不到内容的，已经发送出去了。
select* from qpostoffice
--查看endpoints,会话端点代表Service Broker 会话的每一端
select* from sys.conversation_endpoints
返回：
页:
[1]