我使用.NETCore2.2上的backgroundservice实现了一个kafka消费者作为控制台应用程序。我使用confluent kafka dotnet v1.0.1.1作为apache kafka的客户端。我不太清楚如何处理每条信息。
由于处理每条消息可能需要一些时间(最多24小时),因此我将为每条消息启动一个新任务,这样就不会阻止消费者使用新消息。我认为如果我有太多的信息,每次创建一个新任务是不正确的。那么,处理每条信息的正确方法是什么呢?是否可以为每条消息创建某种动态后台服务?
如果已经在处理一条消息,但应用程序崩溃或发生了重新平衡,那么我最终会多次使用和处理同一条消息。我是否应该自动提交偏移量(或者在它被使用之后立即提交),并将消息(或任务)的状态存储在某个地方,比如数据库中?
我知道有壁炉,但我不确定是否需要用它。如果我目前的做法完全错误,请给我一些建议。
以下是consumerservice的实现:
public class ConsumerService : BackgroundService
{
private readonly IConfiguration _config;
private readonly IElasticLogger _logger;
private readonly ConsumerConfig _consumerConfig;
private readonly string[] _topics;
private readonly double _maxNumAttempts;
private readonly double _retryIntervalInSec;
public ConsumerService(IConfiguration config, IElasticLogger logger)
{
_config = config;
_logger = logger;
_consumerConfig = new ConsumerConfig
{
BootstrapServers = _config.GetValue<string>("Kafka:BootstrapServers"),
GroupId = _config.GetValue<string>("Kafka:GroupId"),
EnableAutoCommit = _config.GetValue<bool>("Kafka:Consumer:EnableAutoCommit"),
AutoOffsetReset = (AutoOffsetReset)_config.GetValue<int>("Kafka:Consumer:AutoOffsetReset")
};
_topics = _config.GetValue<string>("Kafka:Consumer:Topics").Split(',');
_maxNumAttempts = _config.GetValue<double>("App:MaxNumAttempts");
_retryIntervalInSec = _config.GetValue<double>("App:RetryIntervalInSec");
}
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
Console.WriteLine("!!! CONSUMER STARTED !!!\n");
// Starting a new Task here because Consume() method is synchronous
var task = Task.Run(() => ProcessQueue(stoppingToken), stoppingToken);
return task;
}
private void ProcessQueue(CancellationToken stoppingToken)
{
using (var consumer = new ConsumerBuilder<Ignore, Request>(_consumerConfig).SetValueDeserializer(new MessageDeserializer()).Build())
{
consumer.Subscribe(_topics);
try
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
var consumeResult = consumer.Consume(stoppingToken);
// Don't want to block consume loop, so starting new Task for each message
Task.Run(async () =>
{
var currentNumAttempts = 0;
var committed = false;
var response = new Response();
while (currentNumAttempts < _maxNumAttempts)
{
currentNumAttempts++;
// SendDataAsync is a method that sends http request to some end-points
response = await Helper.SendDataAsync(consumeResult.Value, _config, _logger);
if (response != null && response.Code >= 0)
{
try
{
consumer.Commit(consumeResult);
committed = true;
break;
}
catch (KafkaException ex)
{
// log
}
}
else
{
// log
}
if (currentNumAttempts < _maxNumAttempts)
{
// Delay between tries
await Task.Delay(TimeSpan.FromSeconds(_retryIntervalInSec));
}
}
if (!committed)
{
try
{
consumer.Commit(consumeResult);
}
catch (KafkaException ex)
{
// log
}
}
}, stoppingToken);
}
catch (ConsumeException ex)
{
// log
}
}
}
catch (OperationCanceledException ex)
{
// log
consumer.Close();
}
}
}
}
暂无答案!
目前还没有任何答案,快来回答吧!