当前位置：首页 > article >正文

SQL Server Service Broker完整示例

article 2024/11/16 8:36:04

准备

创建Message，Contract，Queue和Service

创建调用存储过程

启用SQL Agent并创建Job执行存储过程

调用demo

常见故障排除

准备

判断你的数据库YourDatabaseName是否启用了Service Broker

SELECT is_broker_enabled FROM sys.databases WHERE name = 'YourDatabaseName';

如果未启用，可以通过以下命令启用。

ALTER DATABASE YourDatabaseName SET ENABLE_BROKER;

创建Message，Contract，Queue和Service

然后按照以下步骤分别创建Message，Contract，Queue和Service。

--1. 创建Message，Contract，Queue和Service
CREATE MESSAGE TYPE [DBTestSync] VALIDATION = NONE;

CREATE CONTRACT [DBTestmessages] ([DBTestSync] SENT BY ANY)

CREATE QUEUE [dbo].[DBTestSyncQueue] WITH STATUS = ON , RETENTION = OFF , POISON_MESSAGE_HANDLING (STATUS = ON) ;

CREATE SERVICE [DBTestSyncService]  ON QUEUE [dbo].[DBTestSyncQueue] ;

创建调用存储过程

The dequeue stored procedure (Service Program) is core of service broker implementation.

Processes the messages in the Queue

Handles the known errors – Inserts back the message into the queue and reprocesses it.

Send email notifications for unhandled exceptions.

Inserts the unknown errors into the ServiceBrokerException table.

--2. 创建调用的存储过程
--DEQUEUE SP/ SERVICE PROGRAM - This code is responsible for picking the message from the queue and processing data
CREATE proc [dbo].[SP_TEST_SERVICE_BROKER_IN_SP]
as
begin
	set nocount on;
	DECLARE @Handle UNIQUEIDENTIFIER ;
	DECLARE @MessageType SYSNAME ;
	DECLARE @Message XML
	DECLARE @dt DATEtime =GETDATE()
	DECLARE @ID INT 
	DECLARE @Name VARCHAR(50)
	declare @spname varchar(500)
	declare @ERROR VARCHAR(500)
 
	SET XACT_ABORT ON
	BEGIN TRY
		--BEGIN TRAN
		WAITFOR( RECEIVE TOP (1)  
		@Handle = conversation_handle,
		@MessageType = message_type_name,
		@Message = message_body FROM dbo.[DBTestSyncQueue]),TIMEOUT 1000--[DBTestSyncQueue]就是上面创建的Queue

		--SELECT cast(@Message  as xml)
		set @spname =CAST(CAST(@Message.query('/mydata/SPName/text()') AS NVARCHAR(MAX)) AS VARCHAR(500))

		IF @spname='SP_TEST_SERVICE_BROKER'
		BEGIN

			Declare @Id int,@Name int

			SET @Id = convert(int, CAST(CAST(@Message.query('/mydata/Id/text()') AS NVARCHAR(MAX)) AS VARCHAR(50)))
			SET @Name = convert(int, CAST(CAST(@Message.query('/mydata/Name/text()') AS NVARCHAR(MAX)) AS VARCHAR(50)))
			
			update [Users] set Name=@Name where Id=@Id
		END

	--COMMIT TRAN
	END TRY
	BEGIN CATCH
		--ROLLBACK
		DECLARE @ErrorHandle UNIQUEIDENTIFIER;
		SET @ERROR =ERROR_MESSAGE()
		IF (ERROR_NUMBER() = 1205 OR ERROR_NUMBER() = 1222 OR ERROR_NUMBER()=18452)
		BEGIN
		  BEGIN DIALOG CONVERSATION @ErrorHandle
			FROM SERVICE DBTestSyncService 
			TO SERVICE 'DBTestSyncService'
			ON CONTRACT DBTestmessages WITH ENCRYPTION = OFF;

			SEND ON CONVERSATION @ErrorHandle MESSAGE TYPE DBTestSync(@MESSAGE);
		  
		END
		ELSE
		BEGIN
		  INSERT INTO dbo.ServiceBrokerException
			VALUES (
			@Message,
			@ERROR ,
			@dt
			);
				declare @messagebody varchar(5000)
			=concat('<b>Error in processing Service Broker Queue</b><BR><b>SPName:</b>',isnull(@spname,'SP Cant be Determined'),'<BR> <b>ERROR:</b>',isnull(@ERROR, 'Error cant be Determined')) exec msdb.dbo.sp_send_dbmail @profile_name='DBAMail', @recipients= 'group-agency360@alterdomusgroup.onmicrosoft.com',
			@subject='Service Broker: Error in processing Service Broker Queue',
			@body=@messagebody, @body_format='HTML'

		END
	END CATCH;
	
	SET XACT_ABORT OFF
END

GO

启用SQL Agent并创建Job执行存储过程

Job需要创建两个Steps,两个Steps内容都是exec SP_TEST_SERVICE_BROKER_IN_SP。

设置Job的Steps：需要创建两个steps。

Step 1：

1. 在General里面输入Step1-SP_TEST_SERVICE_BROKER_IN_SP

2. Database选择YourDatabaseName

3. Advanced选择Go to the next step

Step 2：

1. 在General里面输入Step2-SP_TEST_SERVICE_BROKER_IN_S

2. Database选择YourDatabaseName

3. Advanced选择Go to step: Step1-SP_TEST_SERVICE_BROKER_IN_SP

设置Job的Schedules:

Occurs every day every 10 second(s) between 12:00:00 AM and 11:59:59 PM. Schedule will be used starting on 11/14/2024.

Name: ServiceBrokerJob--自己随意命名

Schedule type: Recurring

Frequence

        Occurs: Daily

        Recurs every: 1 days(s)

Daily frequence

Occures every: 10 seconds--根据自己需要设置

Startint at: 12:00:00 AM

Ending at: 11:59:59 PM

Duration:

Start date: 11/14/2024--默认是你创建的日期

调用demo

--3. 调用demo。你可以在你调用的地方这样子写
create proc SP_TEST
(
	@Id int,
	@Name nvarchar(25)
)
AS
BEGIN
	SELECT * 
	INTO #TmpTest 
	--Forming a Message-- 
	FROM ( SELECT @Id AS Id ,@Name AS Name ,'SP_TEST_SERVICE_BROKER' AS SPName )a     


	DECLARE @XMLMESSAGE XML;   

	SELECT @XMLMESSAGE = (SELECT * FROM #TmpTest FOR XML PATH ('mydata'), TYPE); 

	----Sending Message to the Queue---- 

	DECLARE @Handle UNIQUEIDENTIFIER; 

	BEGIN  

	DIALOG CONVERSATION @Handle 

	FROM SERVICE DBTestSyncService  

	TO SERVICE 'DBTestSyncService' 

	ON CONTRACT DBTestmessages WITH ENCRYPTION = OFF; 

	SEND ON CONVERSATION @Handle MESSAGE TYPE DBTestSync(@XMLMESSAGE); 
END