StructureMap MassTransit 问题：检测到双向依赖关系

Question

当尝试使用 MassTransit RabbitMQ 和 StructureMap 快速连续处理多条消息时，出现错误：

检测到双向依赖关系！

这种情况只发生在消费者身上，错误如下：

MT-Fault-ExceptionType：StructureMap.Building.StructureMapBuildException MT-Fault-Message：检测到双向依赖关系！ 检查下面的 StructureMap 堆栈跟踪： 1.) Acme.Co.WorkerService.Consumers.SomeCommandConsumer 的实例

MT-Fault-StackTrace：在 lambda_method(Closure, IBuildSession, IContext) 在 StructureMap.Building.BuildPlan.Build（IBuildSession 会话，IContext 上下文） 在 StructureMap.SessionCache.GetObject(Type pluginType, Instance instance, ILifecycle 生命周期) 在 StructureMap.SessionCache.GetDefault（类型 pluginType，IPipelineGraph pipelineGraph） 在 StructureMap.BuildSession.GetInstanceT 在 MassTransit.Pipeline.ConsumerFactories.DelegateConsumerFactory1.<Send>d__21.MoveNext() 在 MassTransit.Pipeline.Filters.ConsumerMessageFilter2.<GreenPipes-IFilter<MassTransit-ConsumeContext<TMessage>>-Send>d__4.MoveNext() at MassTransit.Pipeline.Filters.ConsumerMessageFilter2.>-Send>d__4.MoveNext() 在 GreenPipes.Filters.TeeFilter1.<Send>d__5.MoveNext() at GreenPipes.Filters.OutputPipeFilter2.d__7.MoveNext() 在 GreenPipes.Filters.OutputPipeFilter2.<SendToOutput>d__7.MoveNext() at GreenPipes.Filters.DynamicFilter1.d__9.MoveNext() 在 MassTransit.Pipeline.Filters.DeserializeFilter.d__4.MoveNext() 在 GreenPipes.Filters.RescueFilter`2.-Send>d__5.MoveNext()

我看不到任何循环依赖或双向，但我可能遗漏了一些东西。

命令消费者的代码（基本上）是：

public class SomeCommandConsumer : IConsumer<ISomeCommand>
{
    private readonly IRepository _repository;
    private readonly IAnotherRepository _anotherRepository;

    public SomeCommandConsumer(IRepository repository, IAnotherRepository anotherRepository)
    {
        _repository = repository;
        _anotherRepository = anotherRepository;
    }

    public async Task Consume(ConsumeContext<ISomeCommand> context)
    {
        try
        {
            await DoSomeWork(context);
            await context.Publish<ISomeProcessCompleteEvent>(new
            {
                context.Message.ID,
                SomeProperty = SomeEnum.SomeValue
            });
        }
        catch (Exception e)
        {
            HandleException(e);
        }
    }
}

而且 StructureMap 注册是有效的：

public class WorkflowServiceRegistry : Registry
{
    public WorkflowServiceRegistry()
    {
        var configuration = FabricRuntime.GetActivationContext().GetConfigurationPackageObject(Constants.ServiceFabricConfigurationSection);
        var baseUri = configuration.Settings.Sections[Constants.ConfigurationRabbitMqConfigSection].Parameters[Constants.ConfigurationRabbitMqBaseUri].Value;
        var username = configuration.Settings.Sections[Constants.ConfigurationRabbitMqConfigSection].Parameters[Constants.ConfigurationRabbitMqUsername].Value;
        var password = configuration.Settings.Sections[Constants.ConfigurationRabbitMqConfigSection].Parameters[Constants.ConfigurationRabbitMqPassword].Value;
        var connectionString = configuration.Settings.Sections[Constants.ConfigurationDatabaseSection].Parameters[Constants.ConfigurationDatabaseConnectionString].Value;
        var dbTimeout = int.Parse(configuration.Settings.Sections[Constants.ConfigurationDatabaseSection].Parameters[Constants.ConfigurationDatabaseTimeout].Value);

        For<IAnotherRepository>().Use<AnotherRepository>();
        For<IRepository>()
            .Use<Repository>()
            .Ctor<string>()
            .Is(connectionString);
        ForConcreteType<SomethingCompleteEventConsumer>()
            .Configure
            .Ctor<string>("baseUri")
            .Is(baseUri);
        ForConcreteType<SomeCommandConsumer>();
        ForConcreteType<SomethingCompleteEventConsumeObserver>();
        For<IBusControl>().Use(container => BusConfigurator.ConfigureBus(
            baseUri,
            username,
            password,
            (cfg, host) => cfg.ReceiveEndpoint(host, Constants.ControllerSerivceAnalysisProcessStartQueue, SomeCommandQueueConfiguration(container)),
            (cfg, host) => cfg.ReceiveEndpoint(host, Constants.ControllerSerivceActionServiceCompleteQueue, SomethingCompleteEventQueueConfiguration(container))
        ));
        For<ICommunicationListener>().Use<MassTransitListener>();
    }

    private static Action<IRabbitMqReceiveEndpointConfigurator> SomeCommandQueueConfiguration(IContext container)
    {
        return e =>
        {
            e.Consumer(container.GetInstance<SomeCommandConsumer>);
        };
    }

    private static Action<IRabbitMqReceiveEndpointConfigurator> SomethingCompleteEventQueueConfiguration(IContext container)
    {
        return e =>
        {
            e.Consumer(container.GetInstance<SomethingCompleteEventConsumer>);
            e.Observer(container.GetInstance<SomethingCompleteEventConsumeObserver>());
        };
    }
}

只是想知道这是否有人以前见过？

请注意，代码是服务结构解决方案的一部分，每个服务都有自己类似的“引导程序”，或者注册 StructureMap，而且，为了简洁起见，缺少一些代码。

Answer 1

您应该使用 MassTransit.StructureMap 集成库，而不是使用 .Consumer(container.GetInstance<T>)，只需使用 .Consumer<T>(container)。

这将使用 StructureMap 范围提供程序，以便消费者工厂为消费者的所有依赖项创建一个嵌套的生命周期范围。您已经在使用 IContext 进行构建器/注册，所以应该可以。

另外，这不是连接观察者的方式，至少在应该使用的意义上不是。有一个 ConnectObserver 方法，应该在启动总线之前调用它。此外，观察者本身没有生命周期范围，但使用 StructureMap 范围提供程序会在 ConsumeContext 有效负载中添加一个 ILifetimeScope，可用于在运行时解析依赖关系。