无需 Service Broker 的异步 T-SQL 执行

Oleg Vorkunov

5.00/5 (13投票s)

2008年9月16日

CPOL

6分钟阅读

154291

4767

一组 SQL CLR 存储过程，用于在不使用 Service Broker 的情况下异步执行 T-SQL。

引言

SQL Server 开发人员经常需要执行长时间运行的 SQL，例如更新、插入、批量插入、索引创建和其他非查询类型的 T-SQL。最终，总执行时间将是每个 SQL 执行时间的总和。尽管拥有强大的多处理器，SQL Server 开发人员并没有充分利用这些能力。这就是多任务处理的用武之地。SQL Server 2005 及更高版本为数据库开发人员提供了一种异步执行 SQL 的工具——Service Broker。它有其用途，但有时并不能解决问题。Service Broker 是一个消息传递框架，如果开发人员需要在所有并行进程完成后才能在 T-SQL 应用程序中执行下一步，则需要付出额外的努力来实现这一点：等待所有消息被处理，将下一步放入 Service Broker 队列，以及其他技术。我将不详细介绍使用 Service Broker 来完成此任务的多种方法，而是向您展示一种同时执行多个 T-SQL 的不同方法。

背景

假设您正在开发一个 ETL 进程，该进程需要批量加载多个表、创建多个索引、将数据从一个数据库/服务器移动到另一个数据库/服务器，并且您的进程需要移动大量行。最终，即使您对所有索引进行了优化并尽最大可能优化了基础架构，您的进程仍会运行数小时。但是，如果说，您需要复制 10 个表，并且您将同时复制所有这些表。如果您并行执行此操作，将节省大量时间。这是解决方案——一个非常简单的解决方案，而且不涉及 Service Broker 的麻烦和 SQL 序列的同步。复制完所有 10 个表后，您需要为每个复制的表创建索引。所以，在这个简单的项目中，您有两个主要步骤：

步骤 1 - 复制表
步骤 2 - 创建索引

在下面的示例中，我将展示如何使用附加部署脚本中的存储过程来完成此操作。

使用代码

附加的部署脚本包含 SQL CLR 存储过程和用户定义函数，它们实现了并行 SQL 执行。

Parallel_Declare - 声明一个并行块。此过程期望一个参数 @name。该参数仅用于信息目的，以便在需要获取错误消息时使用。
Parallel_UseTransaction - 指示并行块中的所有 SQL 将在单个分布式事务下运行。此过程期望一个参数 @mode。该参数可以具有 varchar(20) 值：Serializable、RepeatableRead、ReadCommitted、ReadUncommitted、Snapshot、Chaos。
Parallel_SetOption_CommandTimeout - 设置 SQL 执行超时。该参数为 int 类型的 @commandTimeout。默认值为 120 秒。如果设置为 0，则并行块中的任何 SQL 执行都不会超时。
Parallel_SetOption_MaxThreads - 设置要同时执行的最大线程数。即使您可以向并行块添加无限数量的 SQL 以并行运行，但一次只执行此选项确定的 SQL 数量。其他 SQL 将在线程池队列中等待。该参数为 int 类型的 @maxThreads。默认值为 10。有效值为 1 到 64。
Parallel_AddSql - 此过程将 SQL 添加到要并行执行的并行块中。此过程不执行 SQL。它只是将其添加到列表中。参数为：nvarchar(128) @key，以及 nvarchar(MAX) @sql，其中 @key 是标识要执行的 SQL 数组中 SQL 的唯一键，@sql 是要执行的实际 SQL。SQL 可以是任何 T-SQL，但不能包含 GO 语句。
Parallel_Execute - 此过程实际开始执行添加到并行块中的所有 SQL。该过程没有参数，但它返回以下值：0 - 执行成功，或失败线程计数的非负值。如果任何 SQL 执行失败，该过程不会崩溃；相反，它将返回一个非零值。如果您为当前块执行调用了 Parallel_UseTransaction，则在成功的情况下，所有 SQL 的结果将 **提交**，在任何 SQL 失败的情况下，所有 SQL 将 **回滚**。 **这是所有这些过程的一个主要要点：** Parallel_Execute 过程将在所有 SQL 完成或任何 SQL 失败之前结束。
Parallel_GetExecutionResult() - 一个用户定义的表函数，它将返回有关所有线程执行的信息。该表将包含 SQL 键、成功标志、线程运行时间、错误消息和错误堆栈。
Parallel_GetErrorMessage() - 如果并行块中的任何 SQL 失败，此函数将返回 nvarchar(MAX) 类型的错误消息；否则，将返回 NULL。

要使用这些存储过程，您需要将提供的脚本应用于您的数据库。例如，我已经设置了一个 ClrLibDb 数据库。确保服务器实例已配置为启用 CLR 支持。如何执行此操作的详细信息此处省略。

将此数据库选项 TRUSTWORTHY 设置为 **on**。
键入 USE [your database name];。
复制/粘贴 Deployment.sql 并执行它。

此外，我创建了一个 [test] 数据库，在该数据库中创建了一个名为 [Table_1] 的表，该表有两个字段，类型分别为 int 和 varchar(50)。

以下是使用示例

-- First you have to declare parallel block.
-- You can name it anyway you like. THIS IS A MUST
exec ClrLibDb.dbo.Parallel_Declare 'test block'

-- optionally you can setup options for your block
exec ClrLibDb.dbo.Parallel_SetOption_MaxThreads 9
exec ClrLibDb.dbo.Parallel_SetOption_CommandTimeout 30

-- if you need all sqls run under single
-- transaction you can call this procedure
-- if any sql failed, everything will be rollback,
-- otherwise everything will be committed
-- after Parallel_Execute procedure completes
exec ClrLibDb.dbo.Parallel_UseTransaction 'ReadUncommitted'

-- and then you can add sqls.
exec ClrLibDb.dbo.Parallel_AddSql 'sql1', 
     'insert into test.dbo.table_1 values (1, ''test1'')'
exec ClrLibDb.dbo.Parallel_AddSql 'sql2', 
     'insert into test.dbo.table_1 values (2, ''test2'')'
exec ClrLibDb.dbo.Parallel_AddSql 'sql3', 
     'insert into test.dbo.table_1 values (3, ''test3'')'
exec ClrLibDb.dbo.Parallel_AddSql 'sql4', 
     'insert into test.dbo.table_1 values (4, ''test4'')'

-- after all sqls are added you call Parallel_Execute
-- to start execution of all sqls in parallel
DECLARE @RC int
exec @RC = ClrLibDb.dbo.Parallel_execute
-- you can analyze error from return code from procedure
IF @RC != 0
BEGIN
    DECLARE @ErrorMessage varchar(MAX)
    SET @ErrorMessage = ClrLibDb.dbo.Parallel_GetErrorMessage()
    print @ErrorMessage
END

-- or you can view execution results
-- using Parallel_GetExecutionResult() function
select * from ClrLibDb.dbo.parallel_GetExecutionResult()

这是 10 个表复制和索引的伪示例（为简化起见，只有三个表 :))

exec ClrLibDb.dbo.Parallel_Declare 'Copy tables'
exec ClrLibDb.dbo.Parallel_UseTransaction 'ReadUncommitted'

exec ClrLibDb.dbo.Parallel_AddSql 'sql1', 
     'SELECT * INTO targetdb.dbo.Table1  FROM sourcedb.dbo.table1'
exec ClrLibDb.dbo.Parallel_AddSql 'sql2', 
     'SELECT * INTO targetdb.dbo.Table2  FROM sourcedb.dbo.table2'
exec ClrLibDb.dbo.Parallel_AddSql 'sql3', 
     'SELECT * INTO targetdb.dbo.Table3  FROM sourcedb.dbo.table3'

DECLARE @RC int
exec @RC = ClrLibDb.dbo.Parallel_execute
IF @RC != 0
BEGIN
    DECLARE @ErrorMessage varchar(MAX)
    SET @ErrorMessage = ClrLibDb.dbo.Parallel_GetErrorMessage()
    RAISERROR(@ErrorMessage, 16, 1)
END

-- tables copy is complete, let's create indexes
exec ClrLibDb.dbo.Parallel_Declare 'Create indexes #1'
exec ClrLibDb.dbo.Parallel_UseTransaction 'ReadUncommitted'

exec ClrLibDb.dbo.Parallel_AddSql 'index1', 
     'CREATE INDEX IX01_Table1 ON targetdb.Table1 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index2', 
     'CREATE INDEX IX02_Table1 ON targetdb.Table1 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index3', 
     'CREATE INDEX IX03_Table1 ON targetdb.Table1 (<fields>)'

DECLARE @RC int
exec @RC = ClrLibDb.dbo.Parallel_execute
IF @RC != 0
BEGIN
    DECLARE @ErrorMessage varchar(MAX)
    SET @ErrorMessage = ClrLibDb.dbo.Parallel_GetErrorMessage()
    RAISERROR(@ErrorMessage, 16, 1)
END

exec ClrLibDb.dbo.Parallel_Declare 'Create indexes #2'
exec ClrLibDb.dbo.Parallel_UseTransaction 'ReadUncommitted'

exec ClrLibDb.dbo.Parallel_AddSql 'index1', 
     'CREATE INDEX IX01_Table2 ON targetdb.Table2 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index2', 
     'CREATE INDEX IX02_Table2 ON targetdb.Table2 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index3', 
     'CREATE INDEX IX03_Table2 ON targetdb.Table2 (<fields>)'

DECLARE @RC int
exec @RC = ClrLibDb.dbo.Parallel_execute
IF @RC != 0
BEGIN
    DECLARE @ErrorMessage varchar(MAX)
    SET @ErrorMessage = ClrLibDb.dbo.Parallel_GetErrorMessage()
    RAISERROR(@ErrorMessage, 16, 1)
END

exec ClrLibDb.dbo.Parallel_Declare 'Create indexes #3'
exec ClrLibDb.dbo.Parallel_UseTransaction 'ReadUncommitted'

exec ClrLibDb.dbo.Parallel_AddSql 'index1', 
     'CREATE INDEX IX01_Table3 ON targetdb.Table3 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index2', 
     'CREATE INDEX IX02_Table3 ON targetdb.Table3 (<fields>)'
exec ClrLibDb.dbo.Parallel_AddSql 'index3', 
     'CREATE INDEX IX03_Table3 ON targetdb.Table3 (<fields>)'

DECLARE @RC int
exec @RC = ClrLibDb.dbo.Parallel_execute
IF @RC != 0
BEGIN
    DECLARE @ErrorMessage varchar(MAX)
    SET @ErrorMessage = ClrLibDb.dbo.Parallel_GetErrorMessage()
    RAISERROR(@ErrorMessage, 16, 1)
END

您可以使用您的创意在其他并行块内运行并行块。是的。这些并行块可以通过将一些并行块放入存储过程来嵌套。但是，在一个并行块中开始的事务不会传递到嵌套的并行块。这是 SQL Server CLR 集成的一个限制。

请注意

关于它是如何工作的，您需要知道

每个 SQL 都在一个单独的线程中执行，并使用 ADO.NET 连接到服务器。它将使用 Windows 身份验证连接到 SQL Server，并使用运行 SQL Server 服务的 Windows 帐户进行身份验证。您可以将此帐户添加为任何数据库的 DBO，或为此帐户设置执行并行 SQL 中编程任务的权限。
由于每个 SQL 都在其自己的连接下运行，因此此连接的默认数据库设置为 [master]，您需要为 SQL 中使用的表、存储过程和其他对象提供数据库名称。您可以从示例中看到，我使用了 targetdb 和 sourcedb 数据库。

历史

2008 年 9 月 16 日 - 上传了 SqlClrLib 项目。