Service Broker对话池性能优化问询：高并发卡顿问题求解

• 多年前部署Service Broker（简称SB），约1年后引入对话池提升性能。
• 当前消息量激增：100+数据源调用存储过程，峰值可达100万条/小时，系统逐渐卡顿，需重启SQL服务才能恢复；正常状态可处理80万条/小时，卡顿期间仅能处理5万条/小时。
• 问题定位在DialogPool的访问阻塞或等待环节，但因多年未接触SB，遗忘相关细节，不知从何入手修复。
• 曾考虑“150技巧”（仅使用每第150个对话以跨页存储），但未实现过，也不清楚在当前架构中如何应用。
• 当前激活存储过程在32核服务器上运行30个线程。
• 需求：找到无需每日多次重启SQL服务的解决方案；微软暂未给出有效建议，下一步计划将DialogPool改为内存优化表。

相关对象定义 #

初始存储过程：BundleParse_SvcBroker_INS #

SET ANSI_NULLS ON
GO
CREATE  PROCEDURE [svcBroker].[BundleParse_SvcBroker_INS] (@Message XML)
AS
SET NOCOUNT ON
DECLARE @fromService sysname,
        @toService sysname,
        @onContract sysname,
        @messageType NVARCHAR(128),
        @messageBody NVARCHAR(MAX)
SET @fromService = 'DynamicParser\\my_InitiatorService'
SET @toService = 'DynamicParser\\my_TargetService'
SET @onContract = 'svcBroker_ods_claim_contract'
SET @messageType = 'svcBroker_ods_claim_request'
SET @messageBody = CONVERT(NVARCHAR(MAX), @message)
EXEC svcBroker.usp_send @fromService, @toService, @onContract, @messageType, @messageBody
GO

DialogPool表结构 #

(
[FromService] [sys].[sysname] NOT NULL,
[ToService] [sys].[sysname] NOT NULL,
[OnContract] [sys].[sysname] NOT NULL,
[Handle] [uniqueidentifier] NOT NULL,
[OwnerSPID] [int] NOT NULL,
[CreationTime] [datetime] NOT NULL,
[SendCount] [bigint] NOT NULL
) ON [PRIMARY]
GO
ALTER TABLE [svcBroker].[DialogPool] ADD CONSTRAINT [UQ__DialogPo__FE5BB31A4CC05EF3] UNIQUE NONCLUSTERED ([Handle]) ON [PRIMARY]
GO

发送消息存储过程：usp_send #

SET QUOTED_IDENTIFIER ON
SET ANSI_NULLS ON
GO
CREATE PROCEDURE [svcBroker].[usp_send] (
      @fromService SYSNAME,
      @toService SYSNAME,
      @onContract SYSNAME,
      @messageType SYSNAME,
      @messageBody NVARCHAR(MAX))
AS
BEGIN
    SET NOCOUNT ON;
    DECLARE @dialogHandle UNIQUEIDENTIFIER;
    DECLARE @sendCount BIGINT;     
    DECLARE @counter INT;
    DECLARE @error INT;
    SELECT @counter = 1;
    BEGIN TRANSACTION;
        WHILE (1=1)
            BEGIN
                EXEC svcBroker.usp_get_dialog @fromService, @toService, @onContract, @dialogHandle OUTPUT, @sendCount OUTPUT;
                IF (@messageBody IS NOT NULL)
                BEGIN
                    SEND ON CONVERSATION @dialogHandle MESSAGE TYPE @messageType (@messageBody);
                END
                ELSE
                BEGIN
                    SEND ON CONVERSATION @dialogHandle MESSAGE TYPE @messageType;
                END
                SELECT @error = @@ERROR;
                IF @error = 0
                BEGIN
                    SET @sendCount = @sendCount + 1;
                    BREAK;
                END
                SELECT @counter = @counter+1;
                IF @counter > 10
                BEGIN
                    RAISERROR('Failed to SEND on a conversation for more than 10 times. Error %i.', 16, 1, @error) WITH LOG;
                    BREAK;
                END
                EXEC svcBroker.usp_delete_dialog @dialogHandle;
                SELECT @dialogHandle = NULL;
        END
        IF @sendCount > 1000
        BEGIN
            EXEC svcBroker.usp_delete_dialog @dialogHandle ;
            SEND ON CONVERSATION @dialogHandle MESSAGE TYPE [svcBroker_EndOfStream];
        END
        ELSE
        BEGIN
            EXEC svcBroker.usp_free_dialog @dialogHandle, @sendCount;
        END
    COMMIT
END;
GO

获取对话存储过程：usp_get_dialog #

SET ANSI_NULLS ON
GO
CREATE PROCEDURE [svcBroker].[usp_get_dialog] (
      @fromService SYSNAME,
      @toService SYSNAME,
      @onContract SYSNAME,
      @dialogHandle UNIQUEIDENTIFIER OUTPUT,
      @sendCount BIGINT OUTPUT)
AS
BEGIN
    SET NOCOUNT ON;
    DECLARE @dialog TABLE (
            FromService SYSNAME NOT NULL,
            ToService SYSNAME NOT NULL,
            OnContract SYSNAME NOT NULL,
            Handle UNIQUEIDENTIFIER NOT NULL,
            OwnerSPID INT NOT NULL,
            CreationTime DATETIME NOT NULL,
            SendCount BIGINT NOT NULL);
    SET TRANSACTION ISOLATION LEVEL READ COMMITTED
    BEGIN TRANSACTION;
        DELETE @dialog;
        UPDATE  TOP(1) svcBroker.DialogPool WITH(READPAST)
        SET     OwnerSPID = @@SPID
        OUTPUT INSERTED.* INTO @dialog 
        WHERE   FromService = @fromService
        AND     ToService = @toService
        AND     OnContract = @OnContract
        AND     OwnerSPID = -1;
        IF @@ROWCOUNT > 0
        BEGIN
            SET @dialogHandle = (SELECT Handle FROM @dialog);
            SET @sendCount = (SELECT SendCount FROM @dialog);          
        END
        ELSE
        BEGIN
            BEGIN DIALOG CONVERSATION @dialogHandle
            FROM SERVICE @fromService
            TO SERVICE @toService
            ON CONTRACT @onContract
            WITH ENCRYPTION = OFF;
            INSERT  INTO svcBroker.DialogPool (
                    FromService, ToService, OnContract, Handle, OwnerSPID, CreationTime, SendCount)
            VALUES (@fromService, @toService, @onContract, @dialogHandle, @@SPID, GETDATE(), 0);
            SET @sendCount = 0;
        END
    COMMIT
END;
GO

删除对话存储过程：usp_delete_dialog #

SET ANSI_NULLS ON
GO
CREATE PROCEDURE [svcBroker].[usp_delete_dialog] (@dialogHandle UNIQUEIDENTIFIER)
AS
BEGIN
      SET NOCOUNT ON;
      BEGIN TRANSACTION;
      DELETE svcBroker.DialogPool WHERE Handle = @dialogHandle;
      COMMIT
END;
GO

激活存储过程：ODS_TargetQueue_Receive #

SET QUOTED_IDENTIFIER ON
SET ANSI_NULLS ON
GO
CREATE PROCEDURE [svcBroker].[ODS_TargetQueue_Receive]
AS
BEGIN
    set nocount on
    DECLARE @receive_table TABLE(
            queuing_order BIGINT,
            conversation_handle UNIQUEIDENTIFIER,
            message_type_name SYSNAME,
            message_body xml);
    DECLARE message_cursor CURSOR LOCAL FORWARD_ONLY READ_ONLY
            FOR SELECT
            conversation_handle,
            message_type_name,
            message_body
            FROM @receive_table ORDER BY queuing_order;
     DECLARE @conversation_handle UNIQUEIDENTIFIER;
     DECLARE @message_type SYSNAME;
     DECLARE @message_body xml;
     DECLARE @error_number INT;
     DECLARE @error_message VARCHAR(4000);
     DECLARE @error_severity INT;
     DECLARE @error_state INT;
     DECLARE @error_procedure SYSNAME;
     DECLARE @error_line INT;
     DECLARE @error_dialog VARCHAR(50);
     BEGIN TRY
       WHILE (1 = 1)
       BEGIN
         BEGIN TRANSACTION;
         WAITFOR (
            RECEIVE TOP (1000)
               [queuing_order],
               [conversation_handle],
               [message_type_name],
               convert(xml, [message_body])
            FROM svcBroker.ODS_TargetQueue
            INTO @receive_table
         ), TIMEOUT 2000;
         IF @@ROWCOUNT = 0
         BEGIN
              COMMIT;
              BREAK;
         END
         ELSE
         BEGIN
              OPEN message_cursor;
              WHILE (1=1)
              BEGIN
                  FETCH NEXT FROM message_cursor
                            INTO @conversation_handle,
                                 @message_type,
                                 @message_body;
                  IF (@@FETCH_STATUS != 0) BREAK;
                  BEGIN TRY
                      IF @message_type = 'svcBroker_ods_claim_request'
                      BEGIN
                          exec ParseMessages @message_body
                      END
                      ELSE IF @message_type in ('svcBroker_EndOfStream', 'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog')
                      BEGIN
                          END CONVERSATION @conversation_handle;
                      END
                      ELSE IF @message_type = 'http://schemas.microsoft.com/SQL/ServiceBroker/Error'
                      BEGIN
                           WITH XMLNAMESPACES ('http://schemas.microsoft.com/SQL/ServiceBroker/Error' AS ssb)
                           SELECT
                           @error_number = CAST(@message_body AS XML).value('(//ssb:Error/ssb:Code)[1]', 'INT'),
                           @error_message = CAST(@message_body AS XML).value('(//ssb:Error/ssb:Description)[1]', 'VARCHAR(4000)');
                           SET @error_dialog = CAST(@conversation_handle AS VARCHAR(50));
                           RAISERROR('Error in dialog %s: %s (%i)', 16, 1, @error_dialog, @error_message, @error_number);
                           END CONVERSATION @conversation_handle;
                      END
                  END TRY
                  BEGIN CATCH
                     SET @error_number = ERROR_NUMBER();
                     SET @error_message = ERROR_MESSAGE();
                     SET @error_severity = ERROR_SEVERITY();
                     SET @error_state = ERROR_STATE();
                     SET @error_procedure = ERROR_PROCEDURE();
                     SET @error_line = ERROR_LINE();
                     IF XACT_STATE() = -1
                     BEGIN
                          ROLLBACK TRANSACTION;
                          BEGIN TRANSACTION;
                            INSERT  INTO svcBroker.target_processing_errors (
                                    error_conversation,[error_number],[error_message],[error_severity],
                                    [error_state],[error_procedure],[error_line],[doomed_transaction],
                                    [message_body])
                            VALUES  (NULL, @error_number, @error_message,@error_severity,
                                    @error_state, @error_procedure, @error_line, 1, @message_body);
                          COMMIT;
                          RAISERROR ('Message processing error', 16, 1);
                     END
                     ELSE IF XACT_STATE() = 1
                     BEGIN
                            INSERT  INTO svcBroker.target_processing_errors (
                                    error_conversation,[error_number],[error_message],[error_severity],
                                    [error_state],[error_procedure],[error_line],[doomed_transaction],
                                    [message_body])
                            VALUES  (NULL, @error_number, @error_message,@error_severity,
                                    @error_state, @error_procedure, @error_line, 0, @message_body);
                     END
                  END CATCH
              END
              CLOSE message_cursor;
              DELETE @receive_table;
         END
         COMMIT;
       END
     END TRY
     BEGIN CATCH
         SET @error_number = ERROR_NUMBER();
         SET @error_message = ERROR_MESSAGE();
         SET @error_severity = ERROR_SEVERITY();
         SET @error_state = ERROR_STATE();
         SET @error_procedure = ERROR_PROCEDURE();
         SET @error_line = ERROR_LINE();
         IF XACT_STATE() = -1
         BEGIN
              ROLLBACK TRANSACTION;
              BEGIN TRANSACTION;
              INSERT INTO svcBroker.target_processing_errors (
                    error_conversation,[error_number],[error_message],[error_severity],
                    [error_state],[error_procedure],[error_line],[doomed_transaction],
                    [message_body])
              VALUES(NULL, @error_number, @error_message,@error_severity, @error_state, @error_procedure, @error_line, 1, @message_body);
              COMMIT;
         END
         ELSE IF XACT_STATE() = 1
         BEGIN
              INSERT INTO svcBroker.target_processing_errors (
                    error_conversation,[error_number],[error_message],[error_severity],
                    [error_state],[error_procedure],[error_line],[doomed_transaction],
                    [message_body])
              VALUES(NULL, @error_number, @error_message, @error_severity, @error_state, @error_procedure, @error_line, 0, @message_body);
              COMMIT;
         END
    END CATCH
END;
GO

注：ParseMessages存储过程仅用于处理消息内容，不含任何Service Broker相关代码。

内容的提问来源于stack exchange，提问作者mbourgon