无法启动多个 Kafka 消费者

大佬教程收集整理的这篇文章主要介绍了无法启动多个 Kafka 消费者，大佬教程大佬觉得挺不错的，现在分享给大家，也给大家做个参考。

如何解决无法启动多个 Kafka 消费者？

我有一个控制台应用程序，它有多个 BACkgroundservices，每个应用程序都使用 Confluent.Kafka nuget 包 (v1.6.2) 从同一个 Kafka 主题中读取。该主题有 3 个分区。

当应用程序启动时，所有后台服务都会调用它们的构造函数，但是只有一个 ExecuteAsync 方法被调用。如果我添加一个 Task.Delay() - 毫秒数似乎并不重要 - 在每个 ExecuteAsync 开始时，一切正常，所有后台服务都运行。

据我所知，没有引发异常。

有没有人知道可能会发生什么，或者在哪里可以进一步查看？

代码如下：

using Confluent.Kafka;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.HosTing;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

namespace KafkaConsumer
{
    class Program
    {
        static voID Main(String[] args)
        {
            CreateHostBuilder(args).build().Run();
        }

        private static IHostBuilder CreateHostBuilder(String[] args) =>
            Host.CreateDefaultBuilder(args)
                .Configureservices((hostContext,services) =>
                {
                    services.AddHostedservice<ConsumerA>();
                    services.AddHostedservice<ConsumerB>();
                    services.AddHostedservice<ConsumerC>();
                });
    }

    public class ConsumerA : BACkgroundservice
    {
        private Readonly ILogger<ConsumerA> _logger;
        private Readonly IConsumer<Ignore,String> _consumer;

        public ConsumerA(ILogger<ConsumerA> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                bootstrapServers = @"server:port",GroupID = "Group1",autoOffsetreset = autoOffsetreset.EarlIEst
            };

            _consumer = new ConsumerBuilder<Ignore,String>(config).build();
            _logger.Loginformation("ConsumerA constructor");
        }

        protected overrIDe async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);

            _logger.Loginformation("ConsumerA starTing");
            _consumer.Subscribe(new List<String> { "topic" });

            while (!cancellationToken.IsCancellationrequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }

    public class ConsumerB : BACkgroundservice
    {
        private Readonly ILogger<ConsumerB> _logger;
        private Readonly IConsumer<Ignore,String> _consumer;

        public ConsumerB(ILogger<ConsumerB> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                bootstrapServers = @"server:port",String>(config).build();
            _logger.Loginformation("ConsumerB constructor");
        }

        protected overrIDe async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);

            _logger.Loginformation("ConsumerB starTing");
            _consumer.Subscribe(new List<String> { "topic" });

            while (!cancellationToken.IsCancellationrequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }

    public class ConsumerC : BACkgroundservice
    {
        private Readonly ILogger<ConsumerC> _logger;
        private Readonly IConsumer<Ignore,String> _consumer;

        public ConsumerC(ILogger<ConsumerC> logger)
        {
            _logger = logger;
            var config = new ConsumerConfig()
            {
                bootstrapServers = @"server:port",String>(config).build();
            _logger.Loginformation("ConsumerC constructor");
        }

        protected overrIDe async Task ExecuteAsync(CancellationToken cancellationToken)
        {
            // await Task.Delay(10);
            
            _logger.Loginformation("ConsumerC starTing");
            _consumer.Subscribe(new List<String> { "topic" });

            while (!cancellationToken.IsCancellationrequested)
            {
                _ = _consumer.Consume(cancellationToken);
            }
        }
    }
}

和输出：

（没有延迟）：

info: KafkaConsumer.ConsumerA[0]
      ConsumerA constructor
info: KafkaConsumer.ConsumerB[0]
      ConsumerB constructor
info: KafkaConsumer.ConsumerC[0]
      ConsumerC constructor
info: KafkaConsumer.ConsumerA[0]
      ConsumerA starTing

（添加延迟）：

info: KafkaConsumer.ConsumerA[0]
      ConsumerA constructor
info: KafkaConsumer.ConsumerB[0]
      ConsumerB constructor
info: KafkaConsumer.ConsumerC[0]
      ConsumerC constructor
info: Microsoft.HosTing.lifetime[0]
      Application started. Press Ctrl+C to shut down.
info: Microsoft.HosTing.lifetime[0]
      HosTing environment: Production
info: Microsoft.HosTing.lifetime[0]
      Content root path: c:\users\..\kafkaconsumer\bin\DeBUG\net5.0
info: KafkaConsumer.ConsumerC[0]
      ConsumerC starTing
info: KafkaConsumer.ConsumerA[0]
      ConsumerA starTing
info: KafkaConsumer.ConsumerB[0]
      ConsumerB starTing

解决方法

protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        // await Task.Delay(10);
        
        _logger.LogInformation("ConsumerC starTing");
        await Task.Run(() => _consumer.Subscribe(new List<String> { "topic" }));
        
        while (!cancellationToken.IsCancellationrequested)
        {
             _await Task.Run(() => consumer.Consume(cancellationToken));
        }
    }

var starTing1 = service1.ExecuteAsync(@R_801_8352@ //all called in sequence without awaits inbetween
var starTing2 = service2.ExecuteAsync(@R_801_8352@
var starTing3 = service3.ExecuteAsync(@R_801_8352@ 
...
//will await the starTings all at once later on

protected Task ExecuteAsync(...) 
{
    return Task.Run(() => { //runs the below on a separate thread from the threadpool
        _logger.LogInformation("ConsumerC starTing");
        _consumer.Subscribe(new List<String> { "topic" });

        while (!cancellationToken.IsCancellationrequested)
        {
            _ = _consumer.Consume(cancellationToken);
        }
    });
}

大佬总结

以上是大佬教程为你收集整理的无法启动多个 Kafka 消费者全部内容，希望文章能够帮你解决无法启动多个 Kafka 消费者所遇到的程序开发问题。

如果觉得大佬教程网站内容还不错，欢迎将大佬教程推荐给程序员好友。

本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ：384754419，请注明来意。