关于数据同步的几种实现

USE master
GO
--如果数据库DBFrom、DBTo不存在，则创建相应的数据库
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name ='DBFrom')
CREATE DATABASE DBFrom
GO
IF NOT EXISTS (SELECT name FROM sys.databases WHERE name ='DBTo')
CREATE DATABASE DBTo
GO
--分别为该数据库启用Service Broker活动并且授权信任
ALTER DATABASE DBFrom SET ENABLE_BROKER
GO
ALTER DATABASE DBFrom SET TRUSTWORTHY ON
GO
ALTER AUTHORIZATION ON DATABASE::DBFrom To sa
GO
ALTER DATABASE DBTo SET ENABLE_BROKER
GO
ALTER DATABASE DBTo SET TRUSTWORTHY ON
GO
ALTER AUTHORIZATION ON DATABASE::DBTo TO sa
GO

Use DBFrom
go
create master key
encryption by password='pass@word1'
goUse DBTo
go
create master key
encryption by password='pass@word1'
go

Use DBFrom
go
--数据同步—消息类型
create message type [http://oa.founder.com/Data/Sync]
validation=well_formed_xml
go
--数据同步--错误反馈消息类型
create message type [http://oa.founder.com/Data/Sync/Error]
validation=well_formed_xml
go
--数据同步协议
create contract[http://oa.founder.com/Data/SyncContract]
(
[http://oa.founder.com/Data/Sync]
sent by initiator,
[http://oa.founder.com/Data/Sync/Error]
sent by target
)
goUse DBTo
go
--数据同步—消息类型
create message type [http://oa.founder.com/Data/Sync]
validation=well_formed_xml
go
--数据同步--错误反馈消息类型
create message type [http://oa.founder.com/Data/Sync/Error]
validation=well_formed_xml
go
--数据同步协议create contract[http://oa.founder.com/Data/SyncContract]
(
[http://oa.founder.com/Data/Sync]
sent by initiator,
[http://oa.founder.com/Data/Sync/Error]
sent by target
)
Go

use DBFrom
go
--同步数据--目标服务配置
create table SyncDataFarServices
(
ServiceID uniqueidentifier,
ServiceName nvarchar(256)
)
go
--将上边创建的服务名，插入此表中
insert into SyncDataFarServices (ServiceID,ServiceName)
values
(NEWID(),'http://oa.founder.com/DBTo/Data/SyncService')
go

 Use DBFrom
go
--发送同步数据消息
Create procedure UP_SyncDataSendMsg
(
@PrimaryKeyField nvarchar(128),
@TableName nvarchar(128),
@DMLType char(1),
@XMLData xml
)
as
beginSET @XMLData.modify('insert <DMLType>{sql:variable("@DMLType")}</DMLType>  as first into /');SET @XMLData.modify('insert <PrimaryKeyField>{sql:variable("@PrimaryKeyField")}</PrimaryKeyField>  as first into /');SET @XMLData.modify('insert <Table>{sql:variable("@TableName")}</Table> as first into /');DECLARE FarServices CURSOR FOR SELECT ServiceName FROM SyncDataFarServices;  open FarServicesdeclare @FarServiceName nvarchar(256);fetch FarServices into @FarServiceName;while @@FETCH_STATUS=0beginbegin Transactiondeclare @Conv_Handler uniqueidentifierbegin DIALOG conversation @Conv_Handler --开始一个会话from service [http://oa.founder.com/DBFrom/Data/SyncService]to service @FarServiceNameon contract [http://oa.founder.com/Data/SyncContract];          send on conversation @Conv_HandlerMessage type [http://oa.founder.com/Data/Sync](@XMLData);fetch FarServices into @FarServiceName;commit;endclose FarServices;deallocate FarServices;
end
go

use DBFrom
go
create Table dbo.SyncException
(
ErrorID uniqueidentifier,
ConversationHandleID uniqueidentifier,
ErrorNumber int,
ErrorSeverity int,
ErrorState int,
ErrorProcedure nvarchar(126),
ErrorLine int,
ErrorMessage nvarchar(2048),
MessageContent nvarchar(max),
CreateDate DateTime
)
go--修改异常信息记录表
alter table dbo.SyncException
add
PrimaryKeyField nvarchar(128),
TableName nvarchar(128),
DMLType char(1),
DBName nvarchar(128)
Go

use DBFrom
go
create procedure UP_SyncDataFeedback
as
begin
set nocount on
--会话变量声明
declare @ConversationHandle uniqueidentifier;--会话句柄
declare @Msg_Body nvarchar(max);
declare @Msg_Type_Name sysname;
--变量赋值
while(1=1)
beginbegin transaction--从队列中接收消息waitfor(receive top(1)@Msg_Type_Name=message_type_name,@ConversationHandle=[conversation_handle],@Msg_Body=message_bodyfrom dbo.[DBFrom_DataSyncQueue]),timeout 1000--如果接收到消息处理，否则跳过if(@@ROWCOUNT<=0)break;if @Msg_Type_Name='http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog'end conversation @ConversationHandle;else if @Msg_Type_Name='http://oa.founder.com/Data/Sync/Error'begindeclare @DataSource xml;set @DataSource=Convert(xml,@Msg_Body);insert into dbo.SyncException(ErrorID,ConversationHandleID,ErrorNumber,ErrorSeverity,ErrorState,ErrorProcedure,ErrorLine,ErrorMessage,
PrimaryKeyField,TableName,DMLType,MessageContent,DBName,CreateDate)selectNEWID(),@ConversationHandle,T.c.value('./@ErrNumber','INT'),T.c.value('./@ErrSeverity','INT'),T.c.value('./@ErrState','INT'),T.c.value('./@ErrProcedure','Nvarchar(126)'),T.c.value('./@ErrLine','INT'),T.c.value('./@ErrMessage','nvarchar(2048)'),T.c.value('./@PrimaryKeyField','nvarchar(128)'),T.c.value('./@TableName','nvarchar(128)'),T.c.value('./@DMLType','char(1)'),T.c.value('./@MessageContent','nvarchar(max)'),T.c.value('./@DBName','nvarchar(128)'),GETDATE()from @DataSource.nodes('/row') as T(c);endelse if @Msg_Type_Name='http://schemas.microsoft.com/SQL/ServiceBroker/Error'end conversation @ConversationHandle;commit Transaction;
end
end
commit;
go

--对Service Broker队列使用内部激活，并指定将调用的存储过程
use DBFrom
go
alter queue dbo.DBFrom_DataSyncQueue with activation
(status=on,max_queue_Readers=1,procedure_name=UP_SyncDataFeedback,execute as owner
);
Go

use DBFrom
Go
--用户信息同步
Create Trigger UT_DataSync_Users
on dbo.Org_Users
after insert,update,delete
as
set nocount on ;
--变量声明
declare @PrimaryKeyField nvarchar(128),@TableName nvarchar(128),@DMLType char(1);
declare @InsertCount int ,@DeleteCount int ;
declare @XMLData xml;
--变量赋值
set @PrimaryKeyField='ID' --组合主键，多个主键使用","隔开
set @TableName='Org_Users'
set @InsertCount=(select COUNT(*) from inserted)
set @DeleteCount=(select COUNT(*) from deleted)
if @InsertCount=@DeleteCount and @InsertCount<>0  ----Updatebeginselect @XMLData=(select * from inserted For xml raw,binary base64,ELEMENTS XSINIL);set @DMLType='U';end
else if(@InsertCount<>0 and @DeleteCount=0) ----Insertbeginselect @XMLData=(select * from inserted for xml raw ,Binary base64,ELEMENTS XSINIL)set @DMLType='A';end
else----Deletebeginselect @XMLData=(select *from deleted for xml raw,binary base64,ELEMENTS XSINIL)set @DMLType='D';endif(@XMLData is not null)beginexec UP_SyncDataSendMsg @PrimaryKeyField,@TableName,@DMLType,@XMLData;endgo

use DBTo
go
--转换用‘,'分割的字符串@str
create Function dbo.uf_SplitString
(
@str nvarchar(max),
@Separator nchar(1)=','
)
returns nvarchar(2000)
as
begindeclare @Fields xml;--结果字段列表declare @Num int;-----记录循环次数declare @Pos int;-----记录开始搜索位置declare @NextPos int;--搜索位置临时变量declare @FieldValue nvarchar(256);--搜索结果set @Num=0;set @Pos=1;set @Fields=CONVERT(xml,'<Fields></Fields>');while (@Pos<=LEN(@Str))  beginselect @NextPos=CHARINDEX(@Separator,@Str,@Pos)if(@NextPos=0 OR @NextPos is null)select @NextPos=LEN(@Str)+1;select @FieldValue=RTRIM(ltrim(substring(@Str,@Pos,@NextPos-@Pos)))select @Pos=@NextPos+1set @Num=@Num+1;if @FieldValue<> ''beginset @Fields.modify('insert <Field>{sql:variable("@FieldValue")}</Field> as last into /Fields[1]');           end   endreturn Convert(nvarchar(2000),@Fields);
end
go

--将XML数据源中的数据同步到数据表中(包括增删改)Use DBTo
go
create function dbo.UF_XMLDataSourceToSQL
(@DataSource XML,--数据源@TableName varchar(128),--同步数据表名称@PrimaryKeyField varchar(128),--需要同步的表的主键，主键为多个时用‘,'隔开@DMLType char(1) --A:新建；U:编辑；D：删除
)
returns nvarchar(4000)
as
begin--变量声明及数据初始化--声明数据表@TableName列Column相关信息变量declare @ColumnName nvarchar(128),@DataType nvarchar(128),@MaxLength int;--声明用于拼接SQL的变量declare @FieldsList nvarchar(4000),@QueryStatement nvarchar(4000);declare @Sql nvarchar(4000);declare @StrLength int;--变量初始化set @FieldsList='  ';--初始化变量不为null，否则对变量使用'+='操作符无效set @QueryStatement='  ';--主键信息，根据参数求解如：<Fields><Field>ID1</Field><Field>ID2</Field></Fields>declare @PKs xml;--当前字段是否主键-在‘更新’，‘删除’同步数据时使用declare @IsPK nvarchar(128);--初始化游标--游标内容包括目标数据表TableName列信息DECLARE ColumnNameList CURSOR FOR SELECT COLUMN_NAME,DATA_TYPE,CHARACTER_MAXIMUM_LENGTH FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME=@TableName AND 
DATA_TYPE<>'xml';--数据处理if @DMLType='A'--插入数据beginopen ColumnNameListfetch ColumnNameList into @ColumnName,@DataType,@MaxLength;while @@FETCH_STATUS=0begin--判断数据源列中是否存在属性：@ColumnName--判断数据源列中是否存在--元素：@ColumnNameIf @DataSource.exist('/row/*[local-name()=sql:variable("@ColumnName")]')=1begin--拼接SQLset @FieldsList+=(@ColumnName+',');set @QueryStatement+=('T.c.value(''(./'+@ColumnName+'[not(@xsi:nil)])[1]'','''+@DataType);--元素读取(包含空值情况)if @MaxLength is not null and @MaxLength<>-1beginset @QueryStatement+='('+CONVERT(nvarchar,@MaxLength)+')';endelse if @MaxLength=-1 and @DataType<>'xml'--已调整beginset @QueryStatement+='(MAX)';endset @QueryStatement+=(''') as '+@ColumnName+',');endfetch ColumnNameList into @ColumnName,@DataType,@MaxLengthendclose ColumnNameList;deallocate ColumnNameList;set @StrLength=LEN(@FieldsList);--去掉@FieldsList结尾的’,'set @FieldsList=SUBSTRING(@FieldsList,1,@StrLength-1);       set @StrLength=LEN(@QueryStatement);--去掉@QueryStatement结尾的’,'set @QueryStatement=SUBSTRING(@QueryStatement,1,@StrLength-1);set @Sql=N'insert into '+@TableName+'('+@FieldsList+') select '+@QueryStatement+' from @DataSource.nodes(''/row/'') as T(c)';endelse if @DMLType='U'--更新数据begin--更新语句where 后的条件表达式declare @Condition nvarchar(1000);set @Condition='  ';set @PKs=CONVERT(xml,dbo.uf_SplitString(@PrimaryKeyField,','));Open ColumnNameListfetch ColumnNameList into @ColumnName,@DataType,@MaxLength;while @@FETCH_STATUS=0begin--判断数据源列中是否存在元素：@ColumnNameif @DataSource.exist('/row/*[local-name()=sql:variable("@ColumnName")]')=1beginset @IsPK=null;SELECT @IsPk=Fs.F FROM (SELECT T.c.value('.[text()]','Nvarchar(128)') AS F FROM @PKs.nodes('/Fields/Field') AS T(c))Fs Where Fs.F=@ColumnNameif @IsPK is null or @IsPK=''begin--非主键，更新字段值set @FieldsList+=(@ColumnName+'=Source.'+@ColumnName+',');endelsebegin--主键，作为要更新条件set @Condition+=@TableName+'.'+@ColumnName+'=Source.'+@ColumnName+' And ';end--XML查询set @QueryStatement+=('T.c.value(''(./'+@ColumnName+'[not(@xsi:nil)])[1]'','''+@DataType);--元素读取（包含空值情况)if @MaxLength is not null and @MaxLength<>-1beginset @QueryStatement+='('+CONVERT(nvarchar,@MaxLength)+')';endelse if @MaxLength=-1 and @DataType<>'xml'beginset @QueryStatement+='(max)';endset @QueryStatement+=(''') as '+@ColumnName+',');endfetch ColumnNameList Into @ColumnName,@DataType,@MaxLengthendclose ColumnNameList;Deallocate ColumnNameList;          --去掉@FieldsList结尾的','set @StrLength=LEN(@FieldsList);set @FieldsList=SUBSTRING(@FieldsList,1,@StrLength-1);        --去掉@QueryStatement结尾的','set @StrLength=LEN(@QueryStatement);set @QueryStatement=SUBSTRING(@QueryStatement,1,@StrLength-1);--去掉@Condition结尾的‘and'set @StrLength=LEN(rtrim(@Condition));set @Condition=SUBSTRING(rtrim(@Condition),1,@StrLength-3);           set @Sql=N'USE DBTo ; update '+@TableName+' set '+@FieldsList+' from (select '+@QueryStatement+' from @DataSource.nodes(''/row'') as T(c)) Source where '+@Condition;end  else if @DMLType='D' --删除数据begin--更新语句where后的条件表达式declare @LinkField nvarchar(1000);set @LinkField='  ';set @PKs=CONVERT(xml,dbo.uf_SplistString(@PrimaryKeyField,','));open ColumnNameListfetch ColumnNameList into @ColumnName,@DataType,@MaxLength;while @@FETCH_STATUS=0beginif @DataSource.exist('row/*[local-name()=sql:variable("@ColumnName")]')=1beginset @IsPK=null;--初始化--当前字段是否为主键select @IsPK=Fs.F from (select T.c.value('.[text()]','nvarchar(128)') as F from @PKs.nodes('/Fields/Field') as T(c))Fs where Fs.F=@ColumnName--主键if @IsPK is not null and @IsPK<>''begin--主键删除条件set @LinkField+='Target.'+@ColumnName+'=Source.'+@ColumnName+' And ';--XML 查询set @QueryStatement+=('T.c.value(''(./'+@ColumnName+'[not(@xsi:nil)])[1]'','''+@DataType);--元素读取（包含空值情况）if(@MaxLength is not null and @MaxLength<>-1)beginset @QueryStatement+='('+CONVERT(nvarchar,@MaxLength)+')';endelse if @MaxLength=-1 and @DataType<>'xml'beginset @QueryStatement+='(max)';endset @QueryStatement+=(''') as '+@ColumnName+',');end endfetch ColumnNameList into @ColumnName,@DataType,@MaxLengthendclose ColumnNameList;deallocate ColumnNameList;         --去除@QueryStateMent结尾的','set @StrLength=LEN(@QueryStatement);set @QueryStatement=SUBSTRING(@QueryStatement,1,@StrLength-1);          --去除@LinkField 结尾的’Add‘set @StrLength=LEN(rtrim(@LinkField));set @LinkField=SUBSTRING(rtrim(@LinkField),1,@StrLength-3);         set @Sql=N'Delete from '+@TableName+' from '+@TableName+' as Target inner join (select '+@QueryStatement+ ' from @DataSource.nodes(''/row'') as T(c))Source on '+@LinkField;end    Return @Sql--'hello'
end
go

--将数据同步到数据表中
create procedure UP_SyncDataToTable
as
begin
set nocount on
--会话变量声明
declare @ConversationHandle uniqueidentifier;--会话句柄
declare @Msg_Body nvarchar(max);
declare @Msg_Type_Name sysname;
declare @ErrorNumber int ;
--变量赋值
while(1=1)beginbegin transaction--从队列中接收消息waitfor(receive top(1)@Msg_Type_Name=message_type_name,@ConversationHandle=[conversation_handle],@Msg_Body=message_body
--         from dbo.[DBTo_DataSyncQueue]from dbo.[DBFrom_DataSyncQueue]),timeout 500--如果接收到消息-处理，否则跳过if @@ROWCOUNT<=0beginbreak;end           if @Msg_Type_Name='http://oa.founder.com/Data/Sync'begin--声明变量declare @DMLType char(1);declare @PrimaryKeyField nvarchar(128),@TableName nvarchar(128),@Sql nvarchar(4000);declare @DataSource xml--受影响的行数declare @EffectRowCount int;declare @ErrMsg xml;begin try--变量赋值set @DataSource=convert(xml,@Msg_Body);--数据源set @PrimaryKeyField=@DataSource.value('(/PrimaryKeyField)[1][text()]','nvarchar(128)');--主键列表set @TableName=@DataSource.value('(/Table)[1][text()]','nvarchar(128)');--操作数据表set @DMLType=@DataSource.value('/DMLType[1][text()]','char(1)');--操作类型set @Sql=dbo.UF_XMLDataSourceToSQL(@DataSource,@TableName,@PrimaryKeyField,@DMLType);exec sp_executesql @Sql,N'@DataSource XML',@DataSource;end trybegin catchdeclare @DBName nvarchar(128)select @DBName=Name from master..SysDataBases where dbid=(select dbid from master..sysprocesses where spid=@@SPID)set @ErrorNumber=ERROR_NUMBER();set @ErrMsg=(select ERROR_NUMBER() as ErrNumber,ERROR_SEVERITY() as ErrSeverity,ERROR_STATE() as ErrState,ERROR_PROCEDURE() as ErrProcedure,ERROR_LINE() as ErrLine,ERROR_MESSAGE() as ErrMessage,@PrimaryKeyField as PrimaryKeyField,@TableName as TableName,@DMLType as DMLType,@Msg_Body as MessageContent,@DBName as DBNamefor XML raw);--GOTO 错误处理标签goto Err_Handle;   end catch--结束会话End Conversation @ConversationHandleif @ErrorNumber is not nullbegin        --错误处理区域Err_Handle:if @ErrMsg is not nullbegindeclare @test nvarchar(128);--发送失败消息send on conversation @ConversationHandlemessage type [http://oa.founder.com/Data/Sync/Error](@ErrMsg)end--结束会话end conversation @ConversationHandle--break;--回滚--不可回滚，否则将无法发送失败消息--GoTO  Err_Lab；end end commit transactionend
end
go

--对Service Broker队列使用内部激活,并指定将要调用的存储过程
use DBTo
go
--alter Queue dbo.[DBTo_DataSyncQueue] with activation
alter Queue dbo.[DBFrom_DataSyncQueue] with activation
(status=on,max_queue_readers=1,Procedure_name=UP_SyncDataToTable,Execute as self
)
Go