Yuebon.EventBus.Kafka/KafkaConsumerHostService.cs

这里只显示每个文件的开头 60 行。登录后可以解锁完整代码。

微信扫码关注公众号，代码整理好后第一时间通知你。

using Autofac;
using Confluent.Kafka.Admin;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;

namespace Yuebon.EventBus.Kafka;

/// <summary>
/// Kafka consumer 监听服务
/// </summary>
public class KafkaConsumerHostService : BackgroundService
{
    private readonly string AUTOFAC_SCOPE_NAME = "blogcore_event_bus";
    private readonly ILogger<KafkaConsumerHostService> _logger;
    private readonly IConsumer<string, byte[]> _consumer;
    private readonly KafkaOptions _options;
    private readonly IEventBusSubscriptionsManager _subsManager;
    private readonly ILifetimeScope _autofac;
    private CancellationTokenSource cts = new();
    public KafkaConsumerHostService(ILogger<KafkaConsumerHostService> logger,
        IOptions<KafkaOptions> options,
        IEventBusSubscriptionsManager eventBusSubscriptionsManager,
        ILifetimeScope autofac)
    {
        _autofac = autofac;
        _subsManager = eventBusSubscriptionsManager;
        _logger = logger;
        _options = options.Value;
        _consumer = new ConsumerBuilder<string, byte[]>(new ConsumerConfig
        {
            BootstrapServers = _options.Servers,
            GroupId = _options.GroupId,
            AutoOffsetReset = AutoOffsetReset.Earliest,
            AllowAutoCreateTopics = true,
            EnableAutoCommit = false,
            LogConnectionClose = false
        }).SetErrorHandler(ConsumerClient_OnConsumeError)
        .Build();
    }
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var result =  await FetchTopicAsync();
        if (result)
        {
            _consumer.Subscribe(_options.Topic);
            while (!cts.Token.IsCancellationRequested)
            {
                var consumerResult = _consumer.Consume(cts.Token);
                try
                {
                    if (consumerResult.IsPartitionEOF || consumerResult.Message.Value == null) continue;

                    var @event = Protobuf.Deserialize<string>(consumerResult.Message.Value);
                    await ProcessEvent(consumerResult.Message.Key, @event);
                }
                catch (ConsumeException e)
                {
                    _logger.LogError($"Error occured: {e.Error.Reason}");