Konfiguration af Service Broker til asynkron behandling

I min sidste artikel talte jeg om fordelene ved at implementere asynkron behandling ved hjælp af Service Broker i SQL Server i forhold til de andre metoder, der findes til afkoblet behandling af lange opgaver. I denne artikel gennemgår vi alle de komponenter, der skal konfigureres til en grundlæggende Service Broker-konfiguration i en enkelt database, og de vigtige overvejelser for samtalestyring mellem broker-tjenester. For at komme i gang skal vi oprette en database og aktivere databasen til brug af Service Broker:

CREATE DATABASE AsyncProcessingDemo;GO IF (SELECT is_broker_enabled FROM sys.databases WHERE name =N'AsyncProcessingDemo') =0BEGIN ÆNDRE DATABASE AsyncProcessingDemo SET ENABLE_BROKER;ENDGO USE AsyncProcessingDemo; 
Konfiguration af mæglerkomponenter
 

 De grundlæggende objekter, der skal oprettes i databasen, er meddelelsestyperne for meddelelserne, en kontrakt, der definerer, hvordan meddelelserne sendes mellem tjenesterne, en kø og initiatortjenesten, og en kø og måltjenesten. Mange eksempler online for servicemægler viser kompleks objektnavngivning for meddelelsestyper, kontrakter og tjenester for Service Broker. Der er dog ikke et krav om, at navnene skal være komplekse, og simple objektnavne kan bruges til alle objekterne.
 

 For meddelelserne skal vi oprette en meddelelsestype for anmodningen, som vil blive kaldt AsyncRequest , og en meddelelsestype for resultatet, som vil blive kaldt AsyncResult . Begge vil bruge XML, der vil blive valideret som korrekt dannet af mæglertjenesterne til at sende og modtage de data, der kræves af tjenesterne.
 
-- Opret meddelelsestyperCREATE MESSAGE TYPE [AsyncRequest] VALIDATION =WELL_FORMED_XML;CREATE MEDDELELSESTYPE [AsyncResult] VALIDATION =WELL_FORMED_XML;
 

 Kontrakten specificerer, at AsyncRequest vil blive sendt af den initierende tjeneste til måltjenesten, og at måltjenesten returnerer et AsyncResult besked tilbage til den initierende tjeneste. Kontrakten kan også specificere flere meddelelsestyper for initiativtageren og målet, eller at en specifik meddelelsestype kan sendes af enhver tjeneste, hvis den specifikke behandling kræver det.
 
-- Opret kontrakten CREATE CONTRACT [AsyncContract] ( [AsyncRequest] SENDT AF INITIATOR, [AsyncResult] SENDT AF MÅL);
 

 For hver af tjenesterne skal der oprettes en kø for at give lagring af de beskeder, som tjenesten modtager. Måltjenesten, som anmodningen vil blive sendt til, skal oprettes med angivelse af AsyncContract for at tillade, at beskeder sendes til tjenesten. I dette tilfælde hedder tjenesten ProcessingService og vil blive oprettet i ProcessingQueue i databasen. Den initierende tjeneste kræver ikke, at en kontrakt specificeres, hvilket gør den kun i stand til at modtage meddelelser som svar på en samtale, der blev indledt fra den.
 
-- Opret behandlingskøen og servicen - angiv kontrakten for at tillade afsendelse til tjenestenCREATE QUEUE ProcessingQueue;CREATE SERVICE [ProcessingService] ON QUEUE ProcessingQueue ([AsyncContract]); -- Opret anmodningskøen og tjenesten CREATE QUEUE RequestQueue; CREATE SERVICE [RequestService] ON QUEUE RequestQueue;
 
Send en meddelelse til behandling
 

 Som jeg forklarede i den forrige artikel, foretrækker jeg at implementere en indpaknings-lagret procedure til at sende en ny besked til en mæglertjeneste, så den kan ændres én gang for at skalere ydeevne, hvis det kræves. Denne procedure er en simpel indpakning omkring oprettelse af en ny samtale og afsendelse af beskeden til ProcessingService .
 
-- Opret indpakningsproceduren for at sende beskederCREATE PROCEDURE dbo.SendBrokerMessage @FromService SYSNAME, @ToService SYSNAME, @Contract SYSNAME, @MessageType SYSNAME, @MessageBody XMLASBEGIN SET NOCOUNT ON; DECLARE @conversation_handle UNIQUEIDENTIFIER; START TRANSAKTIONEN; BEGYND DIALOGSAMTALE @conversation_handle FRA SERVICE @FromService TIL SERVICE @ToService PÅ KONTRAKT @Kontrakt MED KRYPTERING =FRA; SEND PÅ SAMTALE @conversation_handle MESSAGE TYPE @MessageType(@MessageBody); FORGIFT TRANSAKTION;ENDGO
 

 Ved at bruge den lagrede indpakningsprocedure kan vi nu sende en testmeddelelse til ProcessingService for at validere, at vi har sat mæglertjenesterne korrekt op.
 
-- Send en requestEXECUTE dbo.SendBrokerMessage @FromService =N'RequestService', @ToService =N'ProcessingService', @Contract =N'AsyncContract', @MessageType =N'AsyncRequest', @MessageBody =N'12345'; -- Tjek for besked om behandlingskøSELECT CAST(message_body AS XML) FROM ProcessingQueue;GO
 
Behandler meddelelser
 

 Mens vi manuelt kunne behandle beskederne fra ProcessingQueue , vil vi sandsynligvis have, at meddelelserne behandles automatisk, når de sendes til ProcessingService . For at gøre dette skal der oprettes en lagret aktiveringsprocedure, som vi tester og derefter binder til køen for at automatisere behandlingen ved køaktivering. For at behandle en besked skal vi RECEIVE beskeden fra køen i en transaktion sammen med beskedtypen og samtalehåndtaget for beskeden. Meddelelsestypen sikrer, at den passende logik anvendes på meddelelsen, der behandles, og samtalehåndtaget tillader, at et svar sendes tilbage til den initierende tjeneste, når meddelelsen er blevet behandlet.
 

 RECEIVE kommandoen tillader, at en enkelt besked eller flere beskeder inden for det samme samtalehåndtag eller -gruppe behandles i en enkelt transaktion. For at behandle flere meddelelser skal der bruges en tabelvariabel, eller for at udføre enkelt meddelelsesbehandling kan der bruges en lokal variabel. Aktiveringsproceduren nedenfor henter en enkelt meddelelse fra køen, kontrollerer meddelelsestypen for at afgøre, om det er en AsyncRequest besked, og udfører derefter den langvarige proces baseret på de modtagne beskedoplysninger. Hvis den ikke modtager en besked i løkken, vil den vente op til 5000ms eller 5 sekunder på, at en anden meddelelse kommer ind i køen, før den forlader løkken og afslutter dens eksekvering. Efter at have behandlet en meddelelse, bygger den et AsyncResult besked og sender den tilbage til initiativtageren på samme samtalehåndtag, som beskeden blev modtaget fra. Proceduren kontrollerer også meddelelsestypen for at bestemme, om en EndDialog eller Error besked er modtaget for at rydde op i samtalen ved at afslutte den.
 
-- Opret behandlingsprocedure for behandling af køCREATE PROCEDURE dbo.ProcessingQueueActivationASBEGIN SET NOCOUNT ON; DECLARE @conversation_handle UNIQUEIDENTIFIER; DECLARE @message_body XML; DECLARE @message_type_name systemnavn; MENS (1=1) BEGYND TRANSAKTIONEN; WAITFOR ( MODTAG TOP (1) @conversation_handle =conversation_handle, @message_body =CAST(message_body AS XML), @message_type_name =message_type_name FRA ProcessingQueue), TIMEOUT 5000; HVIS (@@ROWCOUNT =0) START TILBAGETRANSAKTIONEN; PAUSE; SLUT HVIS @message_type_name =N'AsyncRequest' BEGIN -- Håndter kompleks lang behandling her -- Til demonstration trækker vi kontonummeret og sender kun et svar tilbage DECLARE @AccountNumber INT =@message_body.value('(AsyncRequest/AccountNumber) [1]', 'INT'); -- Byg svarmeddelelse og send tilbage DECLARE @reply_message_body XML =N' ' + CAST(@AccountNumber AS NVARCHAR(11)) + ' '; SEND PÅ SAMTALE @conversation_handle MEDDELELSESTYPE [AsyncResult] (@reply_message_body); END -- Hvis afslut dialogmeddelelsen, afslut dialogen ELSE IF @message_type_name =N'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog' BEGIN END CONVERSATION @conversation_handle; END -- Hvis fejlmeddelelsen, log og afslut samtalen ELSE IF @message_type_name =N'http://schemas.microsoft.com/SQL/ServiceBroker/Error' BEGIN -- Log fejlkoden og udfør den nødvendige håndtering her -- End samtalen for fejlen AFSLUT KONVERSATION @conversation_handle; AFSLUT FORBINDELSESTRANSAKTIONEN; ENDENDGO
 

 RequestQueue skal også behandle de beskeder, der sendes til den, så en yderligere procedure til behandling af AsyncResult meddelelser, der returneres af ProcessingQueueActivation-proceduren, skal oprettes. Da vi ved, at AsnycResult-meddelelsen betyder, at alt behandlingsarbejde er afsluttet, kan samtalen afsluttes, når vi behandler den besked, som sender en EndDialog-meddelelse til ProcessingService, som derefter vil blive behandlet af dens aktiveringsprocedure for at afslutte samtale rydde op i alt og undgå ilden og glemme problemer, der opstår, når samtaler afsluttes korrekt.
 
-- Opret procedure til behandling af svar på anmodningskøen CREATE PROCEDURE dbo.RequestQueueActivationASBEGIN SET NOCOUNT ON; DECLARE @conversation_handle UNIQUEIDENTIFIER; DECLARE @message_body XML; DECLARE @message_type_name systemnavn; MENS (1=1) BEGYND TRANSAKTIONEN; WAITFOR ( MODTAG TOP (1) @conversation_handle =conversation_handle, @message_body =CAST(message_body AS XML), @message_type_name =message_type_name FROM RequestQueue), TIMEOUT 5000; HVIS (@@ROWCOUNT =0) START TILBAGETRANSAKTIONEN; PAUSE; END IF @message_type_name =N'AsyncResult' BEGIN -- Håndter om nødvendigt svarmeddelelsen her DECLARE @AccountNumber INT =@message_body.value('(AsyncResult/AccountNumber)[1]', 'INT'); -- Da dette er alt det arbejde, der udføres, skal du afslutte samtalen for at sende EndDialog-meddelelsen END CONVERSATION @conversation_handle; END -- Hvis afslut dialogmeddelelsen, afslut dialogen ELSE IF @message_type_name =N'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog' BEGIN END CONVERSATION @conversation_handle; END -- Hvis fejlmeddelelse, log og afslut samtale ELSE IF @message_type_name =N'http://schemas.microsoft.com/SQL/ServiceBroker/Error' BEGIN END END CONVERSATION @conversation_handle; AFSLUT FORBINDELSESTRANSAKTIONEN; ENDENDGO
 
Test af procedurerne
 

 Inden købehandlingen for vores tjenester automatiseres, er det vigtigt at teste aktiveringsprocedurerne for at sikre, at de behandler meddelelserne korrekt, og for at forhindre, at en kø bliver deaktiveret, hvis der skulle opstå en fejl, der ikke håndteres korrekt. Da der allerede er en besked på ProcessingQueue ProcessingQueueActivation procedure kan udføres for at behandle denne meddelelse. Husk, at WAITFOR vil få proceduren til at tage 5 sekunder at afslutte, selvom beskeden behandles med det samme fra køen. Efter at have behandlet meddelelsen, kan vi bekræfte, at proceduren fungerede korrekt ved at forespørge på RequestQueue for at se, om et AsyncResult meddelelse findes, og så kan vi bekræfte, at RequestQueueActivation proceduren fungerer korrekt ved at udføre den.
 
-- Behandle meddelelsen fra behandlingskøenEXECUTE dbo.ProcessingQueueActivation;GO -- Check for svar meddelelse på anmodning queueSELECT CAST(message_body AS XML) FROM RequestQueue;GO -- Behandle meddelelsen fra anmodningskøenEXECUTE dbo.RequestQueueActivation;GO 
 
Automatisering af behandlingen
 

 På dette tidspunkt er alle komponenterne færdige for fuldt ud at automatisere vores behandling. Det eneste, der er tilbage, er at binde aktiveringsprocedurerne til deres passende køer og derefter sende endnu en testmeddelelse for at validere, at den bliver behandlet, og at der ikke er noget tilbage i køerne bagefter.
 
-- Ændre behandlingskøen for at angive intern aktiveringALTER QUEUE ProcessingQueue WITH ACTIVATION ( STATUS =ON, PROCEDURE_NAME =dbo.ProcessingQueueActivation, MAX_QUEUE_READERS =10, EXECUTE AS SELF );GO -- Ændre den interne anmodningskø for QUEUEALTERQuify. MED AKTIVERING ( STATUS =TIL, PROCEDURE_NAME =dbo.RequestQueueActivation, MAX_QUEUE_READERS =10, UDFØR SOM SELV );GO -- Test automatiseret aktivering-- Send en anmodning UDFØR dbo.SendBrokerMessage @FromService =', @ToServiceService N', @ToServiceService ProcessingService', @Contract =N'AsyncContract', @MessageType =N'AsyncRequest', @MessageBody =N'12345'; -- Tjek for besked på behandlingskø -- intet er der, fordi det blev automatisk behandletSELECT CAST(message_body AS XML) FROM ProcessingQueue;GO -- Check for svar besked på anmodningskø -- intet er der, fordi det blev automatisk behandletSELECT CAST(message_body AS XML) FRA RequestQueue;GO
 
Oversigt
 

 De grundlæggende komponenter til automatiseret asynkron behandling i SQL Server Service Broker kan konfigureres i en enkelt databaseopsætning for at tillade afkoblet behandling af langvarige opgaver. Dette kan være et effektivt værktøj til at forbedre applikationens ydeevne ud fra en slutbrugers oplevelse ved at afkoble behandlingen fra slutbrugerens interaktion med applikationen.