在SQL Server 2005中实现异步触发器架构

在SQL Server 2005数据库中，通过新增的Service Broker可以实现异步触发器的处理功能。本文提供一种使用Service Broker实现的通用异步触发器方法。

在本这个方法中，通过Service Broker构造异步触发器处理架构，对于要使用这种架构的表，只需要创建相应的触发器及处理触发器中数据的存储过程，并且在异步触发器架构中登记触发器和处理的存储过程即可。如果一个触发器中的数据要被多个表使用，只需要在dbo.tb_async_trigger_subscribtion中登记相应处理数据的存储过程即可，即一个表的数据变更可以被多个表订阅（使用）。

架构的步骤如下：

1． 数据库配置

需要配置数据库以允许使用Service Broker。本文以tempdb库为例，故配置均在tempdb上下文中进行。

USE tempdbGO -- 允许Service BrokerALTER DATABASE tempdb SETENABLE_BROKERGO

2． 构建异步触发器相关的对象

下面的T-SQL创建异步触发器处理架构相关的对象。

-- =======================================-- 异步触发器对象-- 1. service broker 对象-- =======================================-- a. message type, 要求使用xml 传递数据CREATE MESSAGE TYPE MSGT_async_triggerVALIDATION = WELL_FORMED_XMLGO -- b. 只需要发送消息CREATE CONTRACT CNT_async_trigger( MSGT_async_trigger SENT BY INITIATOR)GO -- c. 存储消息的队列CREATE QUEUE dbo.Q_async_triggerGO -- d. 用于消息处理的服务CREATE SERVICE SRV_async_trigger ON QUEUE dbo.Q_async_trigger(CNT_async_trigger)GO -- =======================================-- 异步触发器对象-- 2. 异步触发器处理的对象-- =======================================-- a. 登记异步触发器的表CREATE TABLE dbo.tb_async_trigger( ID int IDENTITYPRIMARY KEY, table_name sysname, trigger_name sysname) -- b. 登记订阅异步触发器的存储过程CREATE TABLE dbo.tb_async_trigger_subscriber( ID int IDENTITYPRIMARY KEY, procedure_name sysname) -- c. 异步触发器和存储过程之间的订阅关系CREATE TABLE dbo.tb_async_trigger_subscribtion( trigger_id intREFERENCES dbo.tb_async_trigger( ID), procedure_id intREFERENCES dbo.tb_async_trigger_subscriber( ID), PRIMARY KEY(trigger_id, procedure_id))GO -- d. 发送消息的存储过程CREATE PROC dbo.p_async_trigger_send @message xmlASSET NOCOUNT ONDECLARE @handle uniqueidentifierBEGIN DIALOG CONVERSATION @handle FROM SERVICE [SRV_async_trigger] TO SERVICE N'SRV_async_trigger' ON CONTRACT CNT_async_trigger WITHENCRYPTION = OFF;SEND ON CONVERSATION @handle MESSAGE TYPE MSGT_async_trigger(@message);-- 消息发出即可, 不需要回复, 因此发出后即可结束会话END CONVERSATION @handleGO -- e. 处理异步触发器发送的消息CREATE PROC dbo.p_async_trigger_processASSET NOCOUNT ONDECLARE @handle uniqueidentifier, @message xml, @rows intSET @rows = 1WHILE @rows > 0BEGIN -- 处理已经收到的消息 WAITFOR(RECEIVE TOP(1) @handle = conversation_handle, @message = CASEWHEN message_type_name = N'MSGT_async_trigger' THEN CONVERT(xml, message_body)ELSE NULL ENDFROM dbo.Q_async_trigger ), TIMEOUT 10 SET @rows = @@ROWCOUNT IF @rows > 0 BEGIN-- 结束会话END CONVERSATION @handle; -- 处理消息-- a. 取发送者信息DECLARE @table_name sysname, @trigger_name sysname, @sql nvarchar(max)SELECT @table_name = @message.value('(/root/table_name)[1]', 'sysname'), @trigger_name = @message.value('(/root/trigger_name)[1]', 'sysname') -- b. 调用异步触发器订阅的存储过程;WITHSUB AS( SELECT TR.table_name, TR.trigger_name, SUB.procedure_name FROM dbo.tb_async_trigger TR, dbo.tb_async_trigger_subscriber SUB, dbo.tb_async_trigger_subscribtion TRSUB WHERE TRSUB.trigger_id = TR.ID AND TRSUB.procedure_id = SUB.ID)SELECT @sql = (SELECT N'EXEC ' + procedure_name + N' @message'FROM SUBWHERE table_name = @table_name AND trigger_name = @trigger_nameFOR XML PATH(''), ROOT('r'), TYPE ).value('(/r)[1]', 'nvarchar(max)')EXEC sp_executesql @sql, N'@message xml', @message ENDENDGO -- f. 绑定处理的存储过程到队列ALTER QUEUE dbo.Q_async_trigger WITH ACTIVATION(STATUS = ON,PROCEDURE_NAME = dbo.p_async_trigger_process,MAX_QUEUE_READERS = 10,EXECUTE AS OWNER)GO

3． 使用示例

下面的T-SQL演示使用异步触发器构架。示例中创建了三个表：

Dbo.t1 这个是源表，此表的数据变化将用于其他表

Dbo.t2 这个表要求保持与dbo.t1同步

Dbo.tb_log 这个表记录dbo.t1中的数据变化情况

触发器 TR_async_trigger 用于将表Dbo.t1中的数据变化发送到异步触发器构架中。dbo.p_Sync_t1_t2和dbo.p_Record_log用于处理dbo.t1于中变化的数据。

在处理时，需要把相关的信息登记到异步触发器架构的表中。

-- =======================================-- 3. 使用示例-- =======================================-- ===============================-- 测试对象-- a. 源表CREATE TABLE dbo.t1( id int IDENTITYPRIMARY KEY, col int)-- b. 同步的目的表CREATE TABLE dbo.t2( id int IDENTITYPRIMARY KEY, col int)-- c. 记录操作的日志表CREATE TABLE dbo.tb_log( id int IDENTITYPRIMARY KEY, user_name sysname, operate_type varchar(10), inserted xml, deleted xml)GO -- a. 异步发送处理消息的触发器CREATE TRIGGER TR_async_triggerON dbo.t1FOR INSERT, UPDATE, DELETEASIF @@ROWCOUNT = 0 RETURN SET NOCOUNT ON -- 将要发送的数据生成xml 数据DECLARE @message xmlSELECT @message = ( SELECT table_name = ( SELECT TOP 1OBJECT_NAME(parent_object_id) FROM sys.objects WHERE object_id = @@PROCID), trigger_name = OBJECT_NAME(@@PROCID), user_name = SUSER_SNAME(), inserted = ( SELECT * FROM inserted FOR XML AUTO, TYPE), deleted = ( SELECT * FROM deleted FOR XML AUTO, TYPE) FOR XML PATH(''), ROOT('root'), TYPE)-- 发送消息EXEC dbo.p_async_trigger_send @message = @messageGO -- b. 处理异步触发器的存储过程-- b.1 同步到t2 的存储过程CREATE PROC dbo.p_Sync_t1_t2 @message xmlASSET NOCOUNT ONDECLARE @inserted bit, @deleted bitSELECT @inserted = @message.exist('/root/inserted'), @deleted = @message.exist('/root/deleted')IF @inserted = 1 IF @deleted = 1 -- 更新 BEGIN;WITHI AS( SELECT id = T.c.value('@id[1]', 'int'), col = T.c.value('@col[1]', 'int') FROM @message.nodes('/root/inserted/inserted') T(c)),D AS( SELECT id = T.c.value('@id[1]', 'int'), col = T.c.value('@col[1]', 'int') FROM @message.nodes('/root/deleted/deleted') T(c))UPDATE A SET col = I.colFROM dbo.t2 A, I, DWHERE A.ID = I.ID AND I.ID = D.ID END ELSE -- 插入 BEGINSET IDENTITY_INSERT dbo.t2 ON;WITHI AS( SELECT id = T.c.value('@id[1]', 'int'), col = T.c.value('@col[1]', 'int') FROM @message.nodes('/root/inserted/inserted') T(c))INSERT dbo.t2( id, col)SELECT id, colFROM ISET IDENTITY_INSERT dbo.t2 OFF ENDELSE -- 删除BEGIN ;WITH D AS(SELECT id = T.c.value('@id[1]', 'int'), col = T.c.value('@col[1]', 'int')FROM @message.nodes('/root/deleted/deleted') T(c) ) DELETE A FROM dbo.t2 A, D WHERE A.ID = D.IDENDGO -- b.2 记录操作记录到dbo.tb_log 的存储过程CREATE PROC dbo.p_Record_log @message xmlASSET NOCOUNT ONDECLARE @inserted bit, @deleted bitSELECT @inserted = @message.exist('/root/inserted'), @deleted = @message.exist('/root/deleted')INSERT dbo.tb_log( user_name, operate_type, inserted, deleted)SELECT @message.value('(/root/user_name)[1]', 'sysname'), operate_type = CASEWHEN @inserted = 1 AND @deleted = 1 THEN 'update'WHEN @inserted = 1 THEN 'insert'WHEN @deleted = 1 THEN 'delete' END, @message.query('/root/inserted'), @message.query('/root/deleted')GO -- ===============================-- 在异步触发器处理系统中登记对象INSERT dbo.tb_async_trigger( table_name, trigger_name)VALUES( N't1', N'TR_async_trigger') INSERT dbo.tb_async_trigger_subscriber( procedure_name)SELECT N'dbo.p_Sync_t1_t2' UNION ALLSELECT N'dbo.p_Record_log' INSERT dbo.tb_async_trigger_subscribtion( trigger_id, procedure_id)SELECT 1, 1 UNION ALLSELECT 1, 2GO

4.使用测试

下面的T-SQL修改表dbo.t1中的数据，并检查dbo.t2、dbo.tb_log中的数据，以确定异步触发器架构的工作是否成功。

执行完成后可以看到dbo.t2、dbo.tb_log中有相关的记录。

-- ===============================-- 测试INSERT dbo.t1SELECT 1 UNION ALLSELECT 2 UPDATE dbo.t1 SET col = 2WHERE id = 1 DELETE dbo.t1WHERE id = 2 -- 显示结果WAITFOR DELAY '00:00:05' -- 延迟5 分钟, 以便有时间处理消息(因为是异步的)SELECT * FROM dbo.t2SELECT * FROM dbo.tb_logGO

5.使用测试

下面的T-SQL删除本文中建立的所有对象。

-- =======================================-- 5. 删除相关的对象-- =======================================-- a. 删除service broker 对象DROP SERVICE SRV_async_triggerDROP QUEUE dbo.Q_async_triggerDROP CONTRACT CNT_async_triggerDROP MESSAGE TYPE MSGT_async_triggerGO -- b. 删除异步触发器处理的相关对象DROP PROC dbo.p_async_trigger_processDROP PROC dbo.p_async_trigger_sendDROP TABLE dbo.tb_async_trigger_subscribtionDROP TABLE dbo.tb_async_trigger_subscriberDROP TABLE dbo.tb_async_triggerGO -- c. 删除测试的对象DROP TABLE dbo.tb_log, dbo.t1, dbo.t2DROP PROC dbo.p_Sync_t1_t2, dbo.p_Record_log