如何利用MassTransit Saga实现所有HTTP请求完成后发布消息？

Absolutely! Sagas in MassTransit are exactly what you need here. They’re built for orchestration scenarios where you have to track multiple asynchronous operations, wait for specific conditions to align, and then trigger a final action—perfect for your upload workflow.

Let’s break down how to implement this step by step:

1. Define the Saga State Class #

First, create a saga class that tracks the progress of your upload process. It needs to keep count of completed uploads and whether the UploadCompleteCommand has been received:

public class UploadProcessingSaga :
    ISaga,
    InitiatedBy<UploadDataCommand>,
    Orchestrates<UploadDataCompletedEvent>,
    Consumer<UploadCompleteCommand>
{
    // Required by MassTransit to correlate all messages to the same saga instance
    public Guid CorrelationId { get; set; }
    
    // Track how many uploads have finished processing
    public int CompletedUploads { get; set; }
    
    // Flag to mark if the user has sent the upload complete signal
    public bool UploadCompleteReceived { get; set; }

    // Handle the first UploadDataCommand to initialize the saga
    public async Task Consume(ConsumeContext<UploadDataCommand> context)
    {
        // Trigger the actual upload processing (e.g., publish an event for a worker consumer)
        await context.Publish(new StartUploadProcessingEvent 
        { 
            CorrelationId = context.CorrelationId,
            UploadData = context.Message.Data // Pass along the uploaded data
        });
    }

    // Update the count when an upload finishes processing
    public async Task Consume(ConsumeContext<UploadDataCompletedEvent> context)
    {
        CompletedUploads++;
        // Check if we're ready to trigger the final workflow step
        await CheckIfProcessingIsFinished(context);
    }

    // Mark that the user has signaled all uploads are submitted
    public async Task Consume(ConsumeContext<UploadCompleteCommand> context)
    {
        UploadCompleteReceived = true;
        // Check if we're ready to trigger the final workflow step
        await CheckIfProcessingIsFinished(context);
    }

    // Core logic: Only publish the finished message when both conditions are met
    private async Task CheckIfProcessingIsFinished(ConsumeContext context)
    {
        if (CompletedUploads == 3 && UploadCompleteReceived)
        {
            await context.Publish(new UploadProcessingFinished 
            { 
                CorrelationId = CorrelationId 
            });
            
            // Optional: Mark the saga as complete if you don't need to track state anymore
            await context.Publish(new SagaCompleted { CorrelationId = CorrelationId });
        }
    }
}

2. Define Supporting Message Contracts #

You’ll need these message types to tie the workflow together:

• UploadDataCommand: Sent when the user posts data to api/upload
• StartUploadProcessingEvent: Triggers the actual file processing (handled by a separate consumer)
• UploadDataCompletedEvent: Published when an individual upload finishes processing
• UploadCompleteCommand: Sent when the user calls api/uploadcomplete
• UploadProcessingFinished: The final message to kick off your next workflow steps
• SagaCompleted: Optional, to mark the saga as resolved

3. Configure the Saga in MassTransit #

Register your saga and set up its endpoint. For production, use a persistent repository (like Entity Framework or Redis) instead of in-memory to preserve state across service restarts:

services.AddMassTransit(x =>
{
    // Register the saga with a persistent repository (adjust for your database)
    x.AddSaga<UploadProcessingSaga>()
        .EntityFrameworkRepository(r =>
        {
            r.ExistingDbContext<YourApplicationDbContext>();
            r.LockStatementProvider = new SqlServerLockStatementProvider(); // Use Postgres/MySQL variant as needed
        });
    // For development testing: .InMemoryRepository();

    // Register the consumer that handles actual upload processing
    x.AddConsumer<UploadProcessorConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);

        // Configure the dedicated receive endpoint for the saga
        cfg.ReceiveEndpoint("upload-processing-saga", e =>
        {
            e.ConfigureSaga<UploadProcessingSaga>(context);
        });
    });
});

Key Implementation Notes #

• Correlation ID: Ensure all 3 UploadDataCommand messages and the UploadCompleteCommand share the same CorrelationId. You can pass this via an HTTP header like Correlation-Id—MassTransit automatically detects this header and sets it as the message's correlation ID, tying all requests to the same saga instance.
• Error Handling: Add logic to handle failed uploads (e.g., retries, marking uploads as failed, or notifying the user). Extend the saga state to track failed uploads and adjust the completion condition if needed (e.g., require 3 successful uploads instead of just 3 initiated).
• Idempotency: Make sure your upload processing logic is idempotent—if a message is redelivered, it shouldn’t cause duplicate processing or incorrect state updates.
• Persistence: Never use in-memory storage for sagas in production. Persistent storage ensures your saga state survives service restarts and crashes.

This setup will reliably wait for all 3 uploads to finish processing AND for the UploadCompleteCommand to be received before publishing the UploadProcessingFinished message, exactly as you need to advance your workflow.

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