Lambda中配置MassTransit未创建队列与订阅的问题排查

问题背景 #

原本API中的MassTransit配置可正常运行，消费者部署在API内部。迁移消费者到Lambda函数后，移除了API中的AddConsumer和ReceiveEndpoint调用，API目前能正常发布消息、创建主题，但Lambda代码既未创建名为dummy的队列，也未订阅目标主题，构造函数可执行但无任何报错。

核心错误点 #

• Lambda环境不支持MassTransit自动创建队列/订阅：Lambda是事件触发的无服务器计算环境，并非长期运行的服务，MassTransit的ReceiveEndpoint配置逻辑是为长期运行的服务设计的，无法在Lambda中主动创建队列和订阅。
• Lambda触发方式错误：当前用HTTP触发Lambda，但要处理SNS消息，需将Lambda配置为SNS主题的订阅者，而非通过代码手动创建订阅。
• 消费逻辑未执行：FunctionHandler中获取了消费者实例，但未调用Consume方法处理传入的消息。

修正方案 #

1. 调整Lambda的MassTransit配置 #

Lambda无需配置ReceiveEndpoint，仅需注册消费者和AWS客户端基础配置，用于消息反序列化和消费逻辑处理。

2. 配置Lambda为SNS主题订阅者 #

在AWS控制台操作：找到RelationshipCreated主题，添加订阅，选择Lambda作为订阅目标，指定你的Lambda函数。这样API发布消息到主题时，AWS会自动触发Lambda。

修正后的完整Lambda代码 #

public class Function
{
    private readonly IConsumer<RelationshipCreated> _consumer;

    public Function()
    {
        Console.WriteLine("Lambda初始化完成");

        var services = new ServiceCollection();
        services.AddMassTransit(x =>
        {
            x.AddConsumer<RelationshipCreatedConsumer>();
            
            // 仅配置AWS SQS/SNS客户端基础信息
            x.UsingAmazonSqs((context, cfg) =>
            {
                cfg.Host(new Uri("amazonsqs://localhost:4566"), h =>
                {
                    h.AccessKey("your-iam-access-key");
                    h.SecretKey("your-iam-secret-key");

                    h.Config(new AmazonSQSConfig { ServiceURL = "http://host.docker.internal:4566" });
                    h.Config(new AmazonSimpleNotificationServiceConfig { ServiceURL = "http://host.docker.internal:4566" });
                });

                cfg.ConfigureEndpoints(context);
            });
        });

        var provider = services.BuildServiceProvider(true);
        _consumer = provider.GetRequiredService<IConsumer<RelationshipCreated>>();
    }

    public class RelationshipCreatedConsumer : IConsumer<RelationshipCreated>
    {
        public Task Consume(ConsumeContext<RelationshipCreated> context)
        {
            Console.WriteLine("成功处理消息: " + System.Text.Json.JsonSerializer.Serialize(context.Message));
            // TODO: 实现写入S3的业务逻辑
            return Task.CompletedTask;
        }
    }

    // Lambda接收SNS事件的正确参数类型为SNSEvent
    public async Task FunctionHandler(Amazon.Lambda.SNSEvents.SNSEvent evnt, ILambdaContext context)
    {
        Console.WriteLine("收到SNS事件触发");
        foreach (var record in evnt.Records)
        {
            try
            {
                // 解析SNS消息体为目标对象
                var message = System.Text.Json.JsonSerializer.Deserialize<RelationshipCreated>(record.Sns.Message);
                
                // 创建简化版消费上下文，调用消费者逻辑
                var consumeContext = new ConsumeContextWrapper<RelationshipCreated>(message);
                await _consumer.Consume(consumeContext);
                
                Console.WriteLine("消息处理完成");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"消息处理失败: {ex.Message}");
                throw;
            }
        }
    }

    // 辅助类：模拟MassTransit消费上下文（简化实现）
    private class ConsumeContextWrapper<T> : ConsumeContext<T> where T : class
    {
        public T Message { get; }

        public ConsumeContextWrapper(T message)
        {
            Message = message;
        }

        // 实现ConsumeContext<T>的必要抽象方法（空实现或默认值）
        public override Guid MessageId => Guid.NewGuid();
        public override DateTime? SentTime => DateTime.UtcNow;
        public override Guid? RequestId => null;
        public override Guid? CorrelationId => null;
        public override Guid? ConversationId => null;
        public override Guid? InitiatorId => null;
        public override Uri SourceAddress => null;
        public override Uri DestinationAddress => null;
        public override Uri ResponseAddress => null;
        public override Uri FaultAddress => null;
        public override DateTime? ExpirationTime => null;
        public override Headers Headers => new Headers();
        public override HostInfo Host => null;
        public override bool IsResponseAccepted => false;
        public override bool IsFaultAccepted => false;
        public override Task RespondAsync<TResponse>(TResponse message, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task RespondAsync<TResponse>(TResponse message, IResponsePipe pipe, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task RespondAsync(object message, Type messageType, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task RespondAsync(object message, Type messageType, IResponsePipe pipe, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task FaultedAsync(Exception exception, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task FaultedAsync(Exception exception, IFaultPipe pipe, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override void NotifyFaulted(Exception exception) { }
        public override Task Send<T>(T message, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task Publish<T>(T message, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task Publish<T>(T message, IPublishPipe pipe, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task Publish(object message, Type messageType, CancellationToken cancellationToken = default) => Task.CompletedTask;
        public override Task Publish(object message, Type messageType, IPublishPipe pipe, CancellationToken cancellationToken = default) => Task.CompletedTask;
    }
}

关键说明 #

• 参数类型修正：Lambda接收SNS触发时，传入的是SNSEvent对象，需从record.Sns.Message中解析实际消息体，而非直接接收RelationshipCreated类型。
• 消费上下文模拟：由于Lambda并非MassTransit托管的服务，需手动实现简化版的ConsumeContext，确保消费者的Consume方法能正常执行。
• 队列/订阅管理：无需通过代码创建队列和订阅，直接在AWS控制台配置SNS主题到Lambda的订阅即可，AWS会自动完成消息投递。

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