基于SQL Server Service Broker实现指定架构存储过程并行执行

Hey Chris, let's walk through how to build this solution with SQL Server Service Broker—no broken third-party files required. Here's a practical, step-by-step approach to automatically enqueue and run all stored procedures in your Build schema in parallel:

Step 1: Set Up the Core Service Broker Objects #

First, we need to create the foundational Service Broker components to handle messaging between your enqueuing process and execution workers:

-- Enable Service Broker on your database (if not already enabled)
ALTER DATABASE YourDatabaseName SET ENABLE_BROKER;
GO

-- Create a message type to carry the stored procedure name
CREATE MESSAGE TYPE [//YourApp/SPExecutionRequest]
VALIDATION = WELL_FORMED_XML;
GO

-- Create a contract defining the message exchange rules
CREATE CONTRACT [//YourApp/SPExecutionContract]
([//YourApp/SPExecutionRequest] SENT BY INITIATOR);
GO

-- Create a queue to hold execution requests (with auto-activation)
CREATE QUEUE [BuildSPExecutionQueue]
WITH ACTIVATION (
    STATUS = ON,
    PROCEDURE_NAME = [dbo].[ProcessBuildSPQueue], -- We'll build this next
    MAX_QUEUE_READERS = 5, -- Adjust based on your server's capacity (controls parallelism)
    EXECUTE AS OWNER
);
GO

-- Create a service tied to the queue and contract
CREATE SERVICE [//YourApp/SPExecutionService]
ON QUEUE [BuildSPExecutionQueue] ([//YourApp/SPExecutionContract]);
GO

Step 2: Build the Activation Procedure to Execute SPs #

This procedure triggers automatically when messages arrive in the queue, executing target stored procedures in parallel (controlled by MAX_QUEUE_READERS):

CREATE PROCEDURE [dbo].[ProcessBuildSPQueue]
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @conversation_handle UNIQUEIDENTIFIER;
    DECLARE @message_body XML;
    DECLARE @sp_name NVARCHAR(512);

    -- Process messages until the queue is empty
    WHILE (1 = 1)
    BEGIN
        BEGIN TRANSACTION;

        -- Wait for the next message (timeout after 1 second if none arrive)
        WAITFOR (
            RECEIVE TOP(1)
                @conversation_handle = conversation_handle,
                @message_body = message_body
            FROM [BuildSPExecutionQueue]
        ), TIMEOUT 1000;

        -- Exit loop if no messages were received
        IF @@ROWCOUNT = 0
        BEGIN
            ROLLBACK TRANSACTION;
            BREAK;
        END

        -- Extract the stored procedure name from the message
        SET @sp_name = @message_body.value('(/SPName)[1]', 'NVARCHAR(512)');

        BEGIN TRY
            -- Execute the stored procedure (QUOTENAME prevents SQL injection)
            EXEC sp_executesql N'EXEC ' + QUOTENAME(@sp_name);

            -- End the conversation since execution succeeded
            END CONVERSATION @conversation_handle;
        END TRY
        BEGIN CATCH
            -- Handle errors: log them, then end the conversation with an error
            DECLARE @error_msg NVARCHAR(4000) = ERROR_MESSAGE();
            END CONVERSATION @conversation_handle WITH ERROR = 1 DESCRIPTION = @error_msg;
            
            -- Optional: Log errors to a dedicated table
            -- INSERT INTO [dbo].[SPExecutionErrors] (SPName, ErrorMessage, ErrorTime)
            -- VALUES (@sp_name, @error_msg, GETDATE());
        END CATCH

        COMMIT TRANSACTION;
    END
END
GO

Step 3: Create a Procedure to Enqueue All Build Schema SPs #

This procedure scans the Build schema for all stored procedures and sends each one to the execution queue. You can run it manually or schedule it for timed execution:

CREATE PROCEDURE [dbo].[EnqueueAllBuildSPs]
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @sp_name NVARCHAR(512);
    DECLARE @conversation_handle UNIQUEIDENTIFIER;

    -- Cursor to iterate over all SPs in the Build schema
    DECLARE sp_cursor CURSOR FOR
        SELECT QUOTENAME(s.name) + '.' + QUOTENAME(p.name)
        FROM sys.procedures p
        JOIN sys.schemas s ON p.schema_id = s.schema_id
        WHERE s.name = 'Build';

    OPEN sp_cursor;
    FETCH NEXT FROM sp_cursor INTO @sp_name;

    WHILE @@FETCH_STATUS = 0
    BEGIN
        -- Start a new conversation with the execution service
        BEGIN DIALOG @conversation_handle
            FROM SERVICE [//YourApp/SPExecutionService]
            TO SERVICE '//YourApp/SPExecutionService'
            ON CONTRACT [//YourApp/SPExecutionContract]
            WITH ENCRYPTION = OFF; -- Enable encryption if needed

        -- Send the SP name as an XML message
        SEND ON CONVERSATION @conversation_handle
            MESSAGE TYPE [//YourApp/SPExecutionRequest]
            ('<SPName>' + @sp_name + '</SPName>');

        -- End the initiator side of the conversation
        END CONVERSATION @conversation_handle;

        FETCH NEXT FROM sp_cursor INTO @sp_name;
    END

    CLOSE sp_cursor;
    DEALLOCATE sp_cursor;
END
GO

Step 4: Automate the Enqueue Process #

To run this on a schedule, create a SQL Server Agent Job that executes EXEC [dbo].[EnqueueAllBuildSPs]; at your desired interval. This will automatically pick up any new SPs added to the Build schema and enqueue them for parallel execution.

Key Tips for Smooth Operation #

• Parallelism Tuning: Adjust MAX_QUEUE_READERS in the queue creation to match your server's CPU/memory capacity—too high can cause resource contention.
• SP Safety: Ensure all SPs in the Build schema are safe for parallel execution (no shared state conflicts, proper transaction handling).
• Monitoring: Use these queries to check queue status and errors:

  -- Check number of pending execution requests
  SELECT COUNT(*) FROM [BuildSPExecutionQueue];
  
  -- Check failed messages in the transmission queue
  SELECT * FROM sys.transmission_queue;
  
  -- Verify queue activation status
  SELECT name, is_activation_enabled, max_queue_readers FROM sys.service_queues WHERE name = 'BuildSPExecutionQueue';
  

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