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

-- =======================================
-- 3. 使用示例
-- =======================================
-- ===============================
-- 测试对象
-- a. 源表
CREATE TABLE dbo.t1(
    id int IDENTITY
        PRIMARY KEY,
    col int
)
-- b. 同步的目的表
CREATE TABLE dbo.t2(
    id int IDENTITY
        PRIMARY KEY,
    col int
)
-- c. 记录操作的日志表
CREATE TABLE dbo.tb_log(
    id int IDENTITY
        PRIMARY KEY,
    user_name sysname,
    operate_type varchar(10),
    inserted XML,
    deleted xml
)
GO
 
-- a. 异步发送处理消息的触发器
CREATE TRIGGER TR_async_trigger
ON dbo.t1
FOR INSERT, UPDATE, DELETE
AS
IF @@ROWCOUNT = 0
    RETURN
 
SET NOCOUNT ON
 
-- 将要发送的数据生成xml 数据
DECLARE
    @message xml
SELECT
    @message = (
            SELECT
                table_name = (
                        SELECT TOP 1
                            OBJECT_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 = @message
GO
 
-- b. 处理异步触发器的存储过程
-- b.1 同步到t2 的存储过程
CREATE PROC dbo.p_Sync_t1_t2
    @message xml
AS
SET NOCOUNT ON
DECLARE
    @inserted bit,
    @deleted bit
SELECT
    @inserted = @message.exist('/root/inserted'),
    @deleted = @message.exist('/root/deleted')
IF @inserted = 1
    IF @deleted = 1 -- 更新
    BEGIN
        ;WITH
        I 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.col
        FROM dbo.t2 A, I, D
        WHERE A.ID = I.ID
            AND I.ID = D.ID
    END
    ELSE            -- 插入
    BEGIN
        SET IDENTITY_INSERT dbo.t2 ON
        ;WITH
        I 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, col
        FROM I
        SET IDENTITY_INSERT dbo.t2 OFF
    END
ELSE                -- 删除
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.ID
END
GO
 
-- b.2 记录操作记录到dbo.tb_log 的存储过程
CREATE PROC dbo.p_Record_log
    @message xml
AS
SET NOCOUNT ON
DECLARE
    @inserted bit,
    @deleted bit
SELECT
    @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 = CASE
                    WHEN @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 ALL
SELECT N'dbo.p_Record_log'
 
INSERT dbo.tb_async_trigger_subscribtion(
    trigger_id, procedure_id)
SELECT 1, 1 UNION ALL
SELECT 1, 2
GO

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

-- =======================================
-- 5. 删除相关的对象
-- =======================================
-- a. 删除service broker 对象
DROP SERVICE SRV_async_trigger
DROP QUEUE dbo.Q_async_trigger
DROP CONTRACT CNT_async_trigger
DROP MESSAGE TYPE MSGT_async_trigger
GO
 
-- b. 删除异步触发器处理的相关对象
DROP PROC dbo.p_async_trigger_process
DROP PROC dbo.p_async_trigger_send
DROP TABLE dbo.tb_async_trigger_subscribtion
DROP TABLE dbo.tb_async_trigger_subscriber
DROP TABLE dbo.tb_async_trigger
GO
 
-- c. 删除测试的对象
DROP TABLE dbo.tb_log, dbo.t1, dbo.t2
DROP PROC dbo.p_Sync_t1_t2, dbo.p_Record_log