消费者运行时在 Confluent.Kafka 中为主题使用消息

Question

我在 .netCore 项目中使用 Confluent.Kafka(1.4.4) 作为消息代理。在项目启动时，我只为 appSetting.json 文件中的特定服务器设置了“bootstrapservers”，并在必要时使用 API 中的相关类中的以下代码生成消息：

public async Task WriteMessage<T>(string topicName, T message)
    {
        using (var p = new ProducerBuilder<Null, string>(_producerConfig).Build())
        {
            try
            {
                var serializedMessage= JsonConvert.SerializeObject(message);
                var dr = await p.ProduceAsync(topicName, new Message<Null, string> { Value = serializedMessage });
                logger.LogInformation($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
            }
            catch (ProduceException<Null, string> e)
            {
                logger.LogInformation($"Delivery failed: {e.Error.Reason}");
            }
        }
    }

我还在消费者解决方案中添加了以下代码：

public async Task Run()
    {
        using (var consumerBuilder = new ConsumerBuilder<Ignore, string>(_consumerConfig).Build())
        {
            consumerBuilder.Subscribe(new List<string>() { "ActiveMemberCardForPanClubEvent", "CreatePanClubEvent", "RemovePanClubEvent"
            });

            CancellationTokenSource cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) =>
            {
                e.Cancel = true; // prevent the process from terminating.
                cts.Cancel();
            };

            try
            {
                while (true)
                {
                    try
                    {
                        var consumer = consumerBuilder.Consume(cts.Token);
                        if (consumer.Message != null)
                        {
                            using (LogContext.PushProperty("RequestId", Guid.NewGuid()))
                            {
                                //Do something
                                logger.LogInformation($"Consumed message '{consumer.Message.Value}' at: '{consumer.TopicPartitionOffset}'.");
                                await DoJob(consumer.Topic, consumer.Message.Value);
                                consumer.Topic.Remove(0, consumer.Topic.Length);
                            }

                        }
                        else
                        {
                            logger.LogInformation($"message is null for topic '{consumer.Topic}'and partition : '{consumer.TopicPartitionOffset}' .");
                            consumer.Topic.Remove(0, consumer.Topic.Length);
                        }
                    }
                    catch (ConsumeException e)
                    {

                        logger.LogInformation($"Error occurred: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                // Ensure the consumer leaves the group cleanly and final offsets are committed.
                consumerBuilder.Close();
            }
        }
    }

我生成了一条消息，当消费者项目运行时，一切都很顺利，并且正在消费者解决方案中读取该消息。 当消费者项目未运行并且我在 API 中将消息与 API 中的消息生产者排队时，就会出现问题。运行消费者后，该主题没有任何有效消息表明正在生成消息。 我熟悉消息代理并有经验，我知道通过发送消息，它会一直在总线上，直到它被使用，但我不明白为什么在这个项目中它不能与 Kafka 一起使用。

Answer 1

“auto.offset.reset”消费者属性的默认设置是“最新”。

这意味着（在尚未写入偏移量的情况下）如果您将消息写入某个主题然后启动消费者，它将跳过在消费者启动之前写入的任何消息。这可能是您的消费者看不到生产者排队的消息的原因。

解决方法是将“auto.offset.reset”设置为“earliest”，即消费者会从topic上最早的offset开始。

https://docs.confluent.io/current/installation/configuration/consumer-configs.html#auto.offset.reset