Suivi des modifications avec SQL Service Broker Format JSON

SQL Service brocker est un outil qui permet de véhiculer des informations entre serveurs SQL via la couche http. Dans cet article, nous allons voir comment transiter des informations de modification de données d’un serveur à un autre à l’aide de Json. Pour ce faire, nous allons aussi nous attarder sur la fonctionnalité de suivi des modifications de données le “change tracking“. Cette fonctionnalité permet la capture de données modifiées. Tous les types de colonnes de base sont pris en charge par la capture de données modifiées excepté les colonnes calculées.

Installation de service broker

Service Broker est inclus dans chaque instance du Moteur de base de données. Par défaut, Service Broker est activé dans chaque instance et dans chaque nouvelle base de données.

Les données sont automatiquement nettoyées par un thread de nettoyage automatique en fonction de la période de rétention de la base de données.

Activer l’option TRUSTWORTHY sur la base de données :

ALTER DATABASE  AdventureWorks2012 SET TRUSTWORTHY ON

Activer l’option Broker :

ALTER DATABASE  AdventureWorks2012  SET ENABLE_BROKER with ROLLBACK IMMEDIATE

Configuration du message :

CREATE MESSAGE TYPE [//EmployeeMessage] VALIDATION = NONE

Création du contrat :

CREATE CONTRACT [//EmployeeContract] ([//EmployeeMessage] SENT BY INITIATOR)

Création de la Queue :

CREATE QUEUE [dbo].[EmployeeSendQueue] WITH STATUS = ON , RETENTION = ON
ON [PRIMARY]
GO
GRANT RECEIVE ON EmployeeSendQueue TO public

Déclaration du Service :

CREATE SERVICE [//Employee/Target_AdventureWorks2012_Service]
ON QUEUE [dbo].[EmployeeSendQueue]([//EmployeeContract])
GO
GRANT SEND ON SERVICE::[//Employee/Target_AdventureWorks2012_Service] TO public

Création de la route :

CREATE ROUTE [EmployeeRoute] WITH SERVICE_NAME = N'//Employee/AdventureWorks2012_Service' , ADDRESS = N'TCP://127.0.0.1:4022'

Mon service Broker est maintenant correctement configuré.

Configuration de Change tracking

Le suivi des modifications est une solution légère qui fournit un mécanisme efficace de suivi des modifications pour les applications.

L’outil nous permettra de savoir sur une table donnée, quelles lignes ont changé, le nombre de fois qu’une ligne a changé et les valeurs de modifications.

Création du compte :

CREATE USER [sa-scebroker] WITHOUT LOGIN

Définir le change tracking sur la base de données :

ALTER DATABASE [AdventureWorks2012] SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 5 DAYS, AUTO_CLEANUP = ON)

Définir la table que l’on souhaite suivre :

ALTER TABLE [HumanResources].[Employee] ENABLE CHANGE_TRACKING WITH (TRACK_COLUMNS_UPDATED = ON)

Comment ça marche ?

Pour voir comment fonctionne le module de suivi des modifications, je fais un update sur la table Employee.UPDATE [HumanResources].

[Employee]
SET JobTitle = JobTitle + '_0'
WHERE [BusinessEntityID] = 1Maintenant je veux observer les informations m ise à jour dans le disque de stockage.
SELECT
e.JobTitle,
e.MaritalStatus,
c.BusinessEntityID,
c.SYS_CHANGE_VERSION,
c.SYS_CHANGE_OPERATION,
c.SYS_CHANGE_COLUMNS,
c.SYS_CHANGE_CONTEXT
FROM CHANGETABLE (CHANGES [HumanResources].[Employee], 0) AS c
LEFT OUTER JOIN [HumanResources].[Employee] AS e ON e.BusinessEntityID = c.BusinessEntityID

Historisation

Pour faciliter l’accès aux dernières données récupérées, on archive la version des données enregistrées.

CREATE TABLE ChangeTrackingVersion (
lastVersion bigint,
lastsynchro datetime)INSERT INTO ChangeTrackingVersion (lastVersion,lastsynchro)
VALUES (0,getDate())

Procédure stockée d’envoi de données

Nous allons maintenant écrire la procédure stockée qui va remonter les données qui ont été modifiées et les envoyer via le service broker.

On récupère la dernière version des données récupérées par le batch :

DECALRE @MinVersion bigint, @currentVersion bigint
SELECT @currentVersion = lastVersion
FROM ChangeTrackingVersion

On récupère le dernier numéro de version de modification :

SELECT @MinVersion= ISNULL(MAX(sys_change_version),0)
FROM CHANGETABLE(CHANGES [HumanResources].[Employee],@currentVersion ) as c

On vérifie si la version courant est inférieure aux versions à insérer :

if(@currentVersion < @MinVersion)

On récupère les données mises à jour depuis la dernière version :

DECLARE @lastVersion int;
SELECT @lastVersion = ISNULL(lastVersion,0)
FROM ChangeTrackingVersion;

SELECT e.BusinessEntityID, e.NationalIDNumber, e.OrganizationLevel, e.JobTitle, c.SYS_CHANGE_OPERATION
FROM [HumanResources].[Employee] as e
INNER JOIN CHANGETABLE (CHANGES [HumanResources].[Employee], @lastVersion) AS c ON c.[BusinessEntityID] = e.[BusinessEntityID]
FOR JSON AUTO

Envoi du message :

print 'start msg : ' + cast(@BrokerMessage as nvarchar(max))

BEGIN
DIALOG CONVERSATION @conversationHandle
FROM SERVICE [//Employee/Target_AdventureWorks2012_Service]
TO SERVICE N'//Employee/Target_AdventureWorks2012_Service'
ON CONTRACT[//EmployeeContract]
WITH ENCRYPTION = OFF;
SET @requestMsg = cast(@BrokerMessage as varchar(max));
SEND
ON CONVERSATION @conversationHandle
MESSAGE TYPE [//EmployeeMessage] (@requestMsg);
END
CONVERSATION @conversationHandle

print 'end send message'

Mise à jour de la table d’historisation :

UPDATE ChangeTrackingVersion
SET lastVersion = isnull(@currentVersion,0)
,lastsynchro = getDate()

Réception des données

Nous allons maintenant configurer le serveur SQL cible pour réceptionner les données.

Configuration du serveur cible :

Activer l’option TRUSTWORTHY sur la base de données.

ALTER DATABASE  AdventureWorks2012 SET TRUSTWORTHY ON

Activer l’option Broker :

ALTER DATABASE  AdventureWorks2012  SET ENABLE_BROKER with ROLLBACK IMMEDIATE

Configuration du message :

CREATE MESSAGE TYPE [//EmployeeMessage] VALIDATION = NONE

Création du contrat :

CREATE CONTRACT [//EmployeeContract] ([//EmployeeMessage] SENT BY INITIATOR)

Création de la Queue :

ALTER QUEUE [dbo].[FullOrderQueue]
WITH STATUS = ON ,
RETENTION = OFF ,
ACTIVATION (STATUS = ON ,
PROCEDURE_NAME = [dbo].[PI_AMI_analyse_message_insertion_commande],
MAX_QUEUE_READERS = 15,
EXECUTE AS N'dbo'),
POISON_MESSAGE_HANDLING (STATUS = OFF)
GO

Déclaration du Service :

CREATE SERVICE [//Employee/Target_AdventureWorks2012_Service]
ON QUEUE [dbo].[EmployeeSendQueue]([//EmployeeContract])
GO
GRANT SEND ON SERVICE::[//Employee/Target_AdventureWorks2012_Service] TO public

Création de la route :

CREATE ROUTE [EmployeeRoute] WITH SERVICE_NAME = N'//Employee/AdventureWorks2012_Service' , ADDRESS = N'TCP://127.0.0.1:4022'

Et voilà !