我似乎在这个问题上找不到太多。我遇到的问题是我需要运行 2 个以上长时间运行的后台服务,但只有第一个注册服务的 ExecuteAsync 被执行。我尝试通过 BackgroundService 实现它并将代码放在 ExecuteAsync 中,我尝试直接实现 IHostedService 并将长时间运行的代码放在 StartAsync 中。
我认为问题是 return Task.CompletedTask; 永远不会被调用。例如,我有两个 Kafka 消费者实现为 BackgroundServices 除了主题和 OnMessage 方法之外,代码看起来都相同
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
var kafkaEndpoint = _kafkaConfig.Endpoint;
var kafkaTopic = "PhoenixEventStore";
var consumerConfig = new Dictionary<string, object>
{
{ "group.id", "consumer1" },
{ "bootstrap.servers", kafkaEndpoint },
{ "auto.offset.reset", "earliest" }
};
using (var consumer = new Consumer<Null, string>(consumerConfig, null, new StringDeserializer(Encoding.UTF8)))
{
consumer.OnMessage += (obj, msg) =>
{
Log.Information($"Consumer1 Received {msg.Value}");
};
consumer.OnPartitionEOF += (_, end) =>
{
Log.Information($"Consumer1 Reached end of topic {end.Topic} partition {end.Partition}.");
};
consumer.OnError += (_, error) =>
{
Log.Error($"Consumer1 Error: {error}");
};
consumer.Subscribe(new List<string>() { kafkaTopic });
while (!stoppingToken.IsCancellationRequested)
{
consumer.Poll(TimeSpan.FromSeconds(10));
}
//consider setting value that check whether the consumer has stopped polling.
}
return Task.CompletedTask;
}
由于这两个服务都长时间运行,因此永远不会命中 Task.Complete。但是,如果我注释掉 while 循环,则两个服务 ExecuteAsync 都会被命中,而不仅仅是第一个注册的。
我找到了一种似乎可行的解决方法,但想知道是否有其他人有更好的方法。
基本上我重构代码让长时间运行的代码在一个名为 StartConsumer 的 void 方法中运行,然后让我的 ExecuteAsync 看起来像这样
protected override Task ExecuteAsync(CancellationToken stoppingToken)
{
Task.Run(() => StartConsumer(stoppingToken));
return Task.CompletedTask;
}
两种服务都使用注册
services.AddHostedService<MyHostedService1>
services.AddHostedService<MyHostedService2>