AI 助理
备案控制台
开发者社区 云原生 云消息队列 文章 正文

SQL Server Service Broker创建单个数据库会话（消息队列）

2014-12-04 1181
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介：
2018-01-08 18:27 by pursuer.chen, 287 阅读, 1 评论, 收藏, 编辑

概述  

SQL Server 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 不要求会话双方使用相同的验证，但通常会话双方还是会为消息类型指定相同的验证。消息类型不能是临时对象。允许使用以 # 开头的消息类型名称，但它们是永久对象。

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消息类型
      );
复制代码

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] ;
复制代码

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

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
复制代码

 

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
复制代码

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

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

 

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
复制代码

三、查询

复制代码 
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';
复制代码

四、删除会话对象

复制代码 
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

总结

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

 

 





本文转自pursuer.chen(陈敏华)博客园博客，原文链接：http://www.cnblogs.com/chenmh/p/8204765.html，如需转载请自行联系原作者

文章标签：
云数据库 RDS SQL Server 版
数据库
SQL
Go
存储
数据格式
消息中间件
数据安全/隐私保护
XML
关键词：
SQL数据库
数据库sql
service数据库
sqlserver数据库
云数据库RDS SQLServer版数据库
相关实践学习
使用SQL语句管理索引
本次实验主要介绍如何在RDS-SQLServer数据库中，使用SQL语句管理索引。
SQL Server on Linux入门教程
SQL Server数据库一直只提供Windows下的版本。2016年微软宣布推出可运行在Linux系统下的SQL Server数据库，该版本目前还是早期预览版本。本课程主要介绍SQLServer On Linux的基本知识。 相关的阿里云产品：云数据库RDS&nbsp;SQL Server版 RDS SQL Server不仅拥有高可用架构和任意时间点的数据恢复功能，强力支撑各种企业应用，同时也包含了微软的License费用，减少额外支出。 了解产品详情:&nbsp;https://www.aliyun.com/product/rds/sqlserver
长征6号
目录
相关文章
hju6meadphitw
|
28天前
|
SQL 开发框架 .NET
ASP.NET连接SQL数据库：详细步骤与最佳实践指南ali01n.xinmi1009fan.com
随着Web开发技术的不断进步，ASP.NET已成为一种非常流行的Web应用程序开发框架。在ASP.NET项目中，我们经常需要与数据库进行交互，特别是SQL数据库。本文将详细介绍如何在ASP.NET项目中连接SQL数据库，并提供最佳实践指南以确保开发过程的稳定性和效率。一、准备工作在开始之前，请确保您
hju6meadphitw
132 3
陌陌谣
|
11天前
|
SQL 数据采集 监控
局域网监控电脑屏幕软件：PL/SQL 实现的数据库关联监控
在当今网络环境中，基于PL/SQL的局域网监控系统对于企业和机构的信息安全至关重要。该系统包括屏幕数据采集、数据处理与分析、数据库关联与存储三个核心模块，能够提供全面而准确的监控信息，帮助管理者有效监督局域网内的电脑使用情况。
陌陌谣
12 2
1436047922066202
|
14天前
|
存储 数据挖掘 数据库
数据库数据恢复—SQLserver数据库ndf文件大小变为0KB的数据恢复案例
一个运行在存储上的SQLServer数据库，有1000多个文件，大小几十TB。数据库每10天生成一个NDF文件，每个NDF几百GB大小。数据库包含两个LDF文件。 存储损坏，数据库不可用。管理员试图恢复数据库，发现有数个ndf文件大小变为0KB。 虽然NDF文件大小变为0KB，但是NDF文件在磁盘上还可能存在。可以尝试通过扫描&拼接数据库碎片来恢复NDF文件，然后修复数据库。
1436047922066202
28 3
1100237741946300
|
30天前
|
SQL 关系型数据库 MySQL
Go语言项目高效对接SQL数据库：实践技巧与方法
在Go语言项目中，与SQL数据库进行对接是一项基础且重要的任务
1100237741946300
50 11
1100237741946300
|
29天前
|
SQL 监控 数据库
慢SQL对数据库写入性能的影响及优化技巧
在数据库管理系统中，慢SQL（即执行缓慢的SQL语句）不仅会影响查询性能，还可能对数据库的写入性能产生显著的不利影响
1100237741946300
81 5
1100237741946300
|
29天前
|
SQL 数据库 数据库管理
数据库SQL函数应用技巧与方法
在数据库管理中，SQL函数是处理和分析数据的强大工具
1100237741946300
37 5
yeye向阳而生
|
1月前
|
SQL 存储 数据可视化
SQL 数据库大揭秘：连接数字世界的魔法桥梁
在数字化时代，数据如繁星般璀璨，而 SQL 数据库则像强大的引力场，有序汇聚、整理和分析这些数据。SQL 数据库是一个巨大的数字宝库，装满各行各业的“宝藏”。本文将带你探索 SQL 数据库在电商、金融、医疗和教育等领域的应用。例如，在电商中，它能精准推荐商品；在金融中，它是安全卫士，防范欺诈；在医疗中，它是健康管家，管理病历；在教育中，则是智慧导师，个性化教学。此外，还将介绍如何利用板栗看板等工具实现数据可视化，提升决策效率。
yeye向阳而生
34 4
hju6meadphitw
|
1月前
|
SQL 监控 数据处理
SQL数据库数据修改操作详解
数据库是现代信息系统的重要组成部分，其中SQL（StructuredQueryLanguage）是管理和处理数据库的重要工具之一。在日常的业务运营过程中，数据的准确性和及时性对企业来说至关重要，这就需要掌握如何在数据库中正确地进行数据修改操作。本文将详细介绍在SQL数据库中如何修改数据，帮助读者更好
hju6meadphitw
148 4
hju6meadphitw
|
1月前
|
SQL 存储 Oracle
Oracle数据库SQL语句详解与应用指南
在数字化时代，数据库已成为各类企业和组织不可或缺的核心组件。Oracle数据库作为业界领先的数据库管理系统之一，广泛应用于各种业务场景。掌握Oracle数据库的SQL语句是数据库管理员、开发人员及运维人员的基本技能。本文将详细介绍Oracle数据库SQL语句的基本概念、语法、应用及最佳实践。一、Or
hju6meadphitw
46 3
hju6meadphitw
|
1月前
|
SQL 数据管理 数据库
文章初学者指南：SQL新建数据库详细步骤与最佳实践
引言：在当今数字化的世界，数据库管理已经成为信息技术领域中不可或缺的一部分。作为广泛使用的数据库管理系统，SQL已经成为数据管理和信息检索的标准语言。本文将详细介绍如何使用SQL新建数据库，包括准备工作、具体步骤和最佳实践，帮助初学者快速上手。一、准备工作在开始新建数据库之前，你需要做好以下准备工作
hju6meadphitw
109 3
云原生

云消息队列

热门文章

最新文章

  • 1
    利用消息队列MQTT，打造一款属于自己的IM社交软件
  • 2
    什么是消息队列？
  • 3
    SpringBoot开发案例之整合Kafka实现消息队列
  • 4
    为什么需要消息队列，及使用消息队列的好处？
  • 5
    消息队列服务Kafka揭秘：痛点、优势以及适用场景
  • 6
    Redis笔记（七）Java实现Redis消息队列
  • 7
    MSMQ消息队列安装
  • 8
    阿里云正式推出消息队列Kafka：兼容开源，数据可靠性99.999999%
  • 9
    Linux进程间通信——使用消息队列
  • 10
    RabbitMq 集成 spring boot 消息队列 入门Demo
  • 1
    云消息队列 RabbitMQ 版实践解决方案评测
    61
  • 2
    RocketMQ体验测评
    62
  • 3
    关键成果与技术应用
    70
  • 4
    消息队列 MQ使用问题之如何使用DefaultMQPushConsumer来消费消息
    140
  • 5
    消息队列 MQ使用问题之遇到消费速度是固定的并且导致了堆积，该怎么办
    172
  • 6
    消息队列 MQ使用问题之如何在grpc客户端中设置负载均衡器
    46
  • 7
    消息队列 MQ使用问题之如何配置生产环境
    105
  • 8
    消息队列 MQ使用问题之在使用C++客户端和GBase的ESQL进行编译时出现core dump，该怎么办
    122
  • 9
    消息队列 MQ使用问题之如何将RocketMQ中某个集群的topic迁移到另一个集群
    86
  • 10
    消息队列 MQ操作报错合集之建立连接时发生了超时错误，该如何解决
    468

    • 相关课程

    更多
  • SQL入门与实践
  • 数据库的前世今生
  • 数据库核心概念
  • 从传统数据库到云数据库演进
  • 数据库及SQL/MySQL基础
  • SQL进阶及查询

    • 相关电子书

    更多
  • SQL Server在电子商务中的应用与实践
  • GeoMesa on Spark SQL
  • 原生SQL on Hadoop引擎- Apache HAWQ 2.x最新技术解密malili

    • 相关实验场景

    更多
  • 使用DAS实现数据库SQL优化
  • 自建数据库迁移到云数据库
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 云原生HTAP数据库，让你的交易和分析一库搞定
  • 玩转MaxCompute SQL! 30分钟搞定数据分析挖掘
  • 快速体验PolarDB开源数据库
    • 下一篇
    无影云桌面