MassTransit集成Kafka时publishEndpoint.Publish()无法发送消息

问题详情 #

使用MassTransit整合Kafka时，调用ITopicProducer<T>.Produce()可以成功向Kafka发送消息，但调用IPublishEndpoint.Publish()时，消息无法发送，且无任何错误、异常抛出，日志中也没有相关错误记录。

已知正常状态 #

• producer.Produce()可成功发送消息到Kafka；
• Kafka与Zookeeper服务运行正常；
• 目标Kafka主题已存在且可正常访问。

相关代码 #

Startup.cs配置 #

services.AddMassTransit(x =>
{
    x.UsingInMemory();

    x.AddRider(rider =>
    {          
        rider.AddProducer<OrderSubmittedIntegrationEvent>("order-submitted"); 

        rider.AddConsumers(Assembly.GetAssembly(typeof(Startup)));

        rider.UsingKafka((context, k) =>
        {
            k.Host(Configuration["Kafka:Host"]);


            k.TopicEndpoint<OrderSubmittedIntegrationEvent>("order-submitted", "consumer-group-A", e =>
            {
                e.AutoOffsetReset = AutoOffsetReset.Earliest;
                e.ConfigureConsumer<OrderSubmittedConsumer>(context);
            });                  
        });
    });

});

事件发布代码（OrderService.cs） #

public class OrderService : IOrderService
{
    protected IConfiguration _configuration;
    protected readonly IAsyncRepository<Order, long> _repository;

    private readonly IPublishEndpoint _publishEndpoint;
    private readonly ITopicProducer<OrderSubmittedIntegrationEvent> _producer;

    public OrderService(IAsyncRepository<Order, long> repsitory,
                            IConfiguration configuration,
                            IPublishEndpoint publishEndpoint,
                            ITopicProducer<OrderSubmittedIntegrationEvent> producer)           
    {
        _repository = repsitory;
        _configuration = configuration;
        _publishEndpoint = publishEndpoint;
        _producer = producer;
    }

    public async Task<Result> SubmitOrder(long id)
    {
        var integrationEvent = new OrderSubmittedIntegrationEvent(
            order.Id,
            order.OrderDate,
            order.OrderItems.Select(n => new OrderItemSubmittedIntegrationEvent(n.ItemClassId, n.Units)).ToList()
        );

        //await _publishEndpoint.Publish(integrationEvent);
        await _producer.Produce(integrationEvent);
        return validationResult;
    }
}

问题解答 #

1. 使用Kafka时是否需要显式启动MassTransit总线？ #

不需要显式手动启动，但你的配置存在关键问题：你调用x.UsingInMemory()配置了内存总线，却没有通过AddMassTransitHostedService()启动它；更重要的是，当前注入的IPublishEndpoint是绑定到内存总线的，调用Publish()时消息只会进入内存总线，不会转发到Kafka。

2. IPublishEndpoint是否需要特定的消费者注册才能成功发布消息？ #

不需要，发布消息不依赖对应消费者的存在。你的场景里消息发不出去，核心原因是IPublishEndpoint没有关联到Kafka Rider，和消费者是否注册无关。

3. Kafka Rider是否缺少特定配置导致Publish无法正常工作？ #

是的，你的配置没有将IPublishEndpoint与Kafka Rider关联起来，需要做以下调整：

1. 移除x.UsingInMemory()（如果不需要内存总线），或者配置内存总线将发布消息转发到Kafka；
2. 在Kafka Rider配置中添加发布拓扑，将事件类型映射到指定Kafka主题：

   rider.UsingKafka((context, k) =>
   {
       k.Host(Configuration["Kafka:Host"]);
       
       // 配置发布拓扑，将事件映射到目标Kafka主题
       k.Publish<OrderSubmittedIntegrationEvent>(x => x.Topic = "order-submitted");
   
       k.TopicEndpoint<OrderSubmittedIntegrationEvent>("order-submitted", "consumer-group-A", e =>
       {
           e.AutoOffsetReset = AutoOffsetReset.Earliest;
           e.ConfigureConsumer<OrderSubmittedConsumer>(context);
       });                  
   });
   

3. 添加services.AddMassTransitHostedService()，确保MassTransit总线和Rider能正常启动运行。

总结 #

核心问题是你注入的IPublishEndpoint属于内存总线而非Kafka Rider，导致发布的消息无法进入Kafka。通过移除不必要的内存总线配置、添加Kafka发布拓扑、启动MassTransit托管服务即可解决问题。

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