如何最好地组装大量消息处理程序

Question

我正在开始开发消息处理系统，其中各种消息将从外部源到达并需要“处理”。消息对象基本上是 DTO。

我正在考虑让消息类存在于它们自己的程序集中，可供消息处理系统和外部系统使用，并让处理程序从通用抽象类继承，大致如下：

Messages.dll

public abstract class Message {}

public class FooMessage : Message
{
  ...
}

public class BarMessage : Message
{
  ...
}

MessageHandlers.dll

public abstract class MessageHandler<TMessage> where TMessage : Message
{
  public void Handle(TMessage message);
}

public class FooHandler : MessageHandler<FooMessage>
{
  public void Handle(FooMessage message)
  {
    ...
  }
}

现在的问题： 将所有这些粘合在一起的最佳方法是什么？我当然可以做一些反射并编写一个辅助函数来根据传入消息名称实例化适当的处理程序，但这不正是 IoC 容器应该擅长的事情吗？有没有更适合这项工作的人？意见和指示表示赞赏。

谢谢！

Answer 1

由于您需要使用请求/响应消息传递模式，因此可以使用 Agatha (agatha-rrsl) 项目。
它是该模式的实现，它是成熟的，在重负载下进行了测试，在进程内工作，当然也可以使用 WCF 在进程边界之外工作。

它内置了您提到的所有内容，并与许多流行的 IoC 容器（Windsor、StructureMap、Unity、Spring.NET、Ninject）集成，因此您不必重新发明轮子并开始专注于您的业务逻辑。

您可以在 Davy Brion 的博客此处找到非常有用的文章、使用技巧和大量信息。

Answer 2

假设容器中的所有组件都已正确配置，这就是处理消息的方式（本例中的消息是通过队列传输的序列化消息）：

var formatter = new BinaryFormatter();
using (var s = new MemoryStream(message.AsBytes))
{
    dynamic consumable = formatter.Deserialize(s);

    var consumerType = typeof(IMessageConsumer<>).MakeGenericType(consumable.GetType());
    dynamic consumer = this.container.Resolve(consumerType);
    consumer.Consume(consumable);
}

完整的示例代码可在此处获取：https://github.com/ploeh/CQRSonAzureDemo/blob/master/BookingWorkerRole/AzureQueueMessageProcessor.cs

Answer 3

不久前我根据温莎做了类似的事情。您可以阅读有关此这里。

Answer 4

有多种方法可以注册您的处理程序。我总是从手动配置一切开始。在您的示例中，它看起来像这样：

<!-- language: c# -->
container.Register<IMessageHandler<FooMessage>, FooHandler>();
container.Register<IMessageHandler<BarMessage>, BarHandler>();
// etc

抽象基类，而是定义一个 IMessageHandler 接口。

提示：不要使用 应用程序正在增长，这可能会变得很麻烦，因此就您而言，批量注册应该是正确的选择。所有 DI 容器对此都有不同的机制。大多数容器开箱即用地支持此功能，而其他容器则不支持。以下是如何使用 简单注入器 执行此操作：

<!-- language: c# -->
container.RegisterManyOpenGeneric(typeof(MessageHandler<>),
    typeof(MessageHandler<>).Assembly);

这将注册实现 MessageHandler 并与 MessageHandler 位于同一程序集中。同样，这取决于您选择的容器如何编写。

为了处理传入的任意消息，您将需要一些反射，但您不应该在应用程序本身中执行此操作。只需在应用程序中定义一个良好的抽象，并将反射的使用移至应用程序中具有 DI 配置的部分即可。

例如，您可以定义一个 IMessageProcessor 接口，该接口允许处理任何类型的 Message：

<!-- language: c# -->
public interface IMessageProcessor
{
    void Process(Message message);
}

在您的 DI 配置附近，您可以定义一个 IMessageProcessor 的实现，它可能看起来像这样：

<!-- language: c# -->
private class DIMessageProcessor : IMessageProcessor
{
    private readonly Container container;

    public DIMessageProcessor(Container container)
    {
        this.container = container;
    }

    public void Process(Message message)
    {
        if (message == null) throw new ArgumentNullException("message");

        Type messageType = message.GetType();

        Type handlerType =
            typeof(IMessageHandler<>).MakeGenericType(messageType);

        var handler = this.container.GetInstance(handlerType);

        handlerType.GetMethod("Handle").Invoke(handler, message);
    }
}

将其排除在应用程序之外很重要。您不仅会混合职责，而且会更难添加新行为。例如，考虑防止消息重播（当一条消息被执行多次时）。作为 IMessageProcessor 的装饰器来实现此行为会很容易。除此之外，DIMessageProcessor 很容易进行单元测试，或者替换为性能更高的新实现（并再次进行单元测试）。

有多种方法可以注册DIMessageProcessor，具体取决于您选择的容器。例如，使用简单注入器：

<!-- language: c# -->
container.RegisterSingle<IMessageProcessor>(new DIMessageProcessor(container));

有了这个，您就可以将 IMessageProcessor 注入到需要它的类型中。这是 WCF 的示例：

<!-- language: c# -->
[ServiceKnownType("GetKnownMessageTypes")]
public class WCFMessageService
{
    private readonly IMessageProcessor processor;

    public WCFMessageService()
    {
        this.processor = 
            Global.Container.GetInstance<IMessageProcessor>();
    }

    [OperationContract]
    public void Process(Message message)
    {
        this.processor.Process(message);
    }

    public static IEnumerable<Type> GetKnownMessageTypes(
        ICustomAttributeProvider provider)
    {
        var knownMessageTypes =
            from type in typeof(Message).Assembly.GetTypes()
            where typeof(Message).IsAssignableFrom(type)
            select type;

        return knownMessageTypes.ToArray();
    }
}

我希望这会有所帮助。