如何使用状态机通过 MassTransit 将消费者连接到 Kafka 主题答案

【问题标题】：How to connect consumer to Kafka topic with MassTransit using state machine如何使用状态机通过 MassTransit 将消费者连接到 Kafka 主题
【发布时间】：2021-01-12 11:02:07
【问题描述】：

当有两个服务时：

服务 1 托管状态机并生成有关主题的消息。服务 2 应该使用此消息。如何正确设置服务 2 以使用消息？

当代码变成这样时，它不起作用：

services.AddMassTransit(mt =>
{
    mt.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context, SnakeCaseEndpointNameFormatter.Instance));

    mt.AddRider(rider =>
    {
        rider.UsingKafka((ctx, kafka) =>
        {
            kafka.Host("kafka_url");
        });

        rider.AddConsumer<OrderConsumer>()
            .Endpoint(e =>
            {
                e.Name = "queue_name";
                e.Temporary = false;
                e.ConcurrentMessageLimit = 8;
            });
    });
});

当我这样做时，它会抛出 System.ArgumentException: 'The consumer type was not found: OrderConsumer'

services.AddMassTransit(mt =>
{
    mt.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context, SnakeCaseEndpointNameFormatter.Instance));

    mt.AddRider(rider =>
    {
        rider.UsingKafka((ctx, kafka) =>
        {
            kafka.Host("kafka_url");

            kafka.TopicEndpoint<Null, OrderMessage>("queue_name", "group_id", cfg =>
            {
                cfg.AutoOffsetReset = AutoOffsetReset.Earliest;
                cfg.ConfigureConsumer<OrderConsumer>(ctx);
            });
        });
    });
});

【问题讨论】：

我不认为 .Endpoint() 扩展对 Kafka 有任何作用，因此请考虑将其删除。但是，您确实需要在 AddRider 块中使用 AddConsumer()，否则会发生异常。

标签： apache-kafka masstransit

【解决方案1】：

您需要混合发布的两个示例：

services.AddMassTransit(mt =>
{
    mt.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context, SnakeCaseEndpointNameFormatter.Instance));

    mt.AddRider(rider =>
    {
        rider.AddConsumer<OrderConsumer>()
        
        rider.UsingKafka((ctx, kafka) =>
        {
            kafka.Host("kafka_url");

            kafka.TopicEndpoint<Null, OrderMessage>("queue_name", "group_id", cfg =>
            {
                cfg.AutoOffsetReset = AutoOffsetReset.Earliest;
                cfg.ConfigureConsumer<OrderConsumer>(ctx);
            });
        });
    });
});

【讨论】：