K8s环境下MassTransit Pod停启时取消消费者任务方案咨询

针对你遇到的问题，核心是要让MassTransit在Pod收到SIGTERM信号时，正确触发消费任务的取消，并阻止新消息的消费。以下是可行的实现步骤：

1. 调整MassTransit配置，关联应用生命周期与停止超时 #

在AddMassTransit配置中，需要明确设置接收端点的停止超时，并启用停止时的取消触发，同时确保应用关闭令牌能传递到MassTransit。

builder.Services.AddMassTransit(config =>
{
    config.AddConsumer<NotificationMessageConsumer>();

    config.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host(builder.Configuration["RabbitMQ:Host"], h =>
        {
            h.Username(builder.Configuration["RabbitMQ:Username"]);
            h.Password(builder.Configuration["RabbitMQ:Password"]);
        });

        cfg.ReceiveEndpoint("notification-queue", e =>
        {
            e.ConfigureConsumer<NotificationMessageConsumer>(ctx);
            // 设置停止时等待正在处理消息的超时（需小于K8s终止宽限期）
            e.StopTimeout = TimeSpan.FromSeconds(25);
            // 启用接收端点停止时触发消费任务取消
            e.UseCancelWhenStopping();
            // 可选：配置重试策略，忽略停止导致的取消异常，避免消息重复入队
            e.UseMessageRetry(r =>
            {
                r.Ignore<OperationCanceledException>();
                r.Interval(3, TimeSpan.FromSeconds(5));
            });
        });
    });
});

// 配置应用关闭超时，需与MassTransit停止超时匹配
builder.Services.Configure<HostOptions>(options =>
{
    options.ShutdownTimeout = TimeSpan.FromSeconds(30);
});

2. 改造消费者，合并应用停止令牌与消费上下文令牌 #

由于消费上下文的CancellationToken可能无法及时响应Pod的SIGTERM信号，我们可以注入IHostApplicationLifetime获取应用停止令牌，将其与上下文令牌合并，确保取消信号能及时传递到业务逻辑。

public class NotificationMessageConsumer : IConsumer<NotificationMessage>
{
    private readonly IHostApplicationLifetime _appLifetime;

    public NotificationMessageConsumer(IHostApplicationLifetime appLifetime)
    {
        _appLifetime = appLifetime;
    }

    public async Task Consume(ConsumeContext<NotificationMessage> context)
    {
        // 合并消费上下文令牌与应用停止令牌
        using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(
            context.CancellationToken, 
            _appLifetime.ApplicationStopping);
        var combinedToken = linkedTokenSource.Token;

        try
        {
            // 先检查是否已触发取消
            combinedToken.ThrowIfCancellationRequested();

            // 将合并后的令牌传入业务方法，确保业务逻辑能响应取消
            await ProcessNotification(context.Message, combinedToken);
        }
        catch (OperationCanceledException ex)
        {
            // 记录取消日志
            // 仅在非应用停止导致的取消时重新抛出，避免消息重复处理
            if (!_appLifetime.ApplicationStopping.IsCancellationRequested)
                throw;
        }
    }

    private async Task ProcessNotification(NotificationMessage message, CancellationToken cancellationToken)
    {
        // 业务逻辑中需定期检查取消令牌，比如在耗时操作的间隙
        foreach (var item in message.Items)
        {
            cancellationToken.ThrowIfCancellationRequested();
            // 执行具体业务操作
            await Task.Delay(500, cancellationToken);
        }
    }
}

3. 配置K8s Pod的终止宽限期 #

确保K8s Pod的terminationGracePeriodSeconds值大于MassTransit的StopTimeout和应用的ShutdownTimeout，给系统足够时间处理完正在消费的消息，避免Pod被强制杀死。

apiVersion: v1
kind: Pod
metadata:
  name: your-app-pod
spec:
  containers:
  - name: your-app-container
    image: your-app-image
    terminationGracePeriodSeconds: 35 # 需大于MassTransit的25秒停止超时

关键说明 #

• UseCancelWhenStopping：让MassTransit在接收端点停止时，主动触发消费上下文的取消令牌。
• 令牌合并：解决了消费上下文令牌不及时响应应用停止信号的问题，确保取消能立即传递到业务逻辑。
• 终止宽限期配置：避免K8s在MassTransit还在处理消息时强制杀死Pod，保证消息处理的完整性。

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