我正在研究 Worker Service in .Net Core 并向其中添加了 DDD。在应用层，我创建了一个 Factory Pattern 来选择一个特定的消息处理程序来处理不同的 Kafka 主题。这种方法运行良好。但是，想知道这是实现它的正确方法吗？我对在 MessageFactory 中注入构造函数参数有点困惑
public interface IMessageFactory
{
IMessageHandler CreateMessage(string topicName);
}工厂方法的实现
public class MessageFactory: IMessageFactory
{
private readonly ILogger<MessageFactory> _logger;
private readonly IOneRepository _oneRepository;
private readonly ITwoRepository _twoRepositoty;
public MessageFactory(ILogger<MessageFactory> logger,
IOneRepository oneRepository,
ITwoRepository twoRepositoty)
{
_logger = logger;
_oneRepository = oneRepository;
_twoRepositoty = twoRepositoty;
}
public IMessageHandler CreateMessage(string topicName)
{
switch (topicName.ToUpper())
{
case KafkaConstants.OneMaster:
return new OneMessageHandler(_logger, _oneRepository );
case KafkaConstants.TwoMaster:
return new TwoMessageHandler(_logger, _twoRepositoty );
default:
return null;
}
}
}
}个人处理程序
public class OneMessageHandler : IMessageHandler
{
private readonly ILogger<MessageFactory> _logger;
private readonly IOneRepository _oneRepository;
public OneMessageHandler (ILogger<MessageFactory> logger,
IOneRepository _oneRepository)
{
_logger = logger;
_oneRepository = oneRepository;
}
public async Task<bool> ProcessMessage(string message)
{
..........
}
}
public class TwoMessageHandler : IMessageHandler
{
private readonly ILogger<MessageFactory> _logger;
private readonly ITwoRepository _twoRepository;
public TwoMessageHandler(ILogger<MessageFactory> logger,
ITwoRepository twoRepository)
{
_logger = logger;
_twoRepository = twoRepository;
}
public async Task<bool> ProcessMessage(string message)
{
..........
}
}工人服务
public class Worker: BackgroundService
{
private readonly IMessageFactory _messageFactory;
public Worker(IMessageFactory messageFactory)
{
_messageFactory = messageFactory;
}
IMessageHandler messageHandler = _messageFactory.CreateMessage("OneMaster");
messageHandler.ProcessMessage(message_from_kafka_response));
}依赖注入
services.AddTransient<IMessageFactory, MessageFactory>();回答1
好吧，我会让所有东西都可注入，所以我可以像这样注册它（所有类都自我解释），而不使用工厂，只使用容器：
container.RegisterSingleton<IHostedService>(x =>
new RepeatBackgroundWorker(
repeatInterval,
new MessageHandler<TKey, TPayload>( //this can also be just something like void IWorker.RunAsync(cancellationToken)
new KafkaMessageProducer<TKey, TPayload>(/*settings like name, consumer group and connection settings*/),
x.GetByKey<IMessageProcessor<TKey, TPayload>>(/*processor name*/))));
//RepeatBackgroundWorker - it will generally repeat some background worker
//MessageHandler - it will consume message, process it, set new offset and until message producer enumerable reaches the end (end of queue)
//KafkaMessageProducer - it will just return something like IAsyncEnumerable<Message<TKey, TPayload>>
//IMessageProcessor - it will be your processor如您所见，您有多种好处：
- 还可以将任何部分包装到通用日志记录或度量系统中，例如 MessageProcessorLogger、MessageProducerLogger 等
- 可以对生产者应用一般过滤（如按 Key 或有效负载中的某些字段分组，例如：按 UserId 分组消息并采取最后执行的操作，或将其包装到某个批处理包装器中等）
- 可以操纵处理程序的行为（如何以及何时重新启动，如何处理错误等）
- 可以操纵各种来源（例如，将 kafka 更改为 rabbitmq 等）
- 可以将整个工厂放入容器或运行时
- 易于在测试中使用，将生产者更改为某些自定义实现，或用某些东西模拟处理器/消费者
- 这样你基本上只需要实现一个处理器/消费者并指定队列的名称（kafka的+消费者组），它会自动为你提供命名良好的指标和日志
PS我没有实现所有类，因为它取决于业务需求
PS2 这基本上是我在生产中使用来压缩 kafka 性能，将所有日志放入 kibana 和所有指标（当您遇到峰值并想要重新平衡向上/向下时，消息时间戳特别有用）从所有来源输出到 prometheus+grafana的盒子。