消息处理程序未激活

Question

我遇到了一些关于 rebus 的问题。

这是我的场景。 我们提供三项服务

Identity) 发布了“IdentityCreated”消息

网关）将“UpdateProfileCommand”直接发送到“profile-westeu-input”队列中

Profile) 使用来自输入队列“profile-westeu-input”的消息并订阅“IdentityCreated”消息

Profile Service

中看到的rebus配置

鉴于我已经在温莎城堡注册了我的处理程序。

container.Register(Classes.FromThisAssembly()
                  .BasedOn(typeof(IHandleMessages<>))
                  .WithServiceAllInterfaces()
                  .LifestyleTransient());

我用 Rebus 配置了

var bus = Configure.With(new CastleWindsorContainerAdapter(container))
            .Logging(x => x.Trace())
            .Transport(
                t => t.UseAzureServiceBus(connectionStringNameOrConnectionString: connectionString,
                        inputQueueAddress: ProfileInputQueueName, mode: AzureServiceBusMode.Standard))
            .Options(o => o.SimpleRetryStrategy(ProfileErrorQueueName))
            .Start();

并订阅了这样的消息类型

bus.Subscribe(typeof(Nabufit.Messages.Identity.Events.IdentityCreated)).Wait()

我预计我的处理程序会被自动调用。但是它没有:(。

我尝试了不同的解决方案

• 更改了输入队列的名称
• 创建了一个事件发射器程序，该程序发布了一个“IdentityCreated”类型的事件。在输入队列中查看时，它存在，但不会被 rebus 拾取。

奖金信息：

• 使用 azure 服务总线
• 在 Service Fabric 应用程序中托管 Rebus
• 我的输入队列名为“profile-westeu-input”

Answer 1

在调查该应用程序后，我们发现我们在 OwinCommunicationListener 中的 Webapi 之间共享了 Windsor Container，它具有一些自定义的生命周期依赖项配置。这导致了两个不同的错误。

1. 由于容器配置，拒绝接收事件
2. 从架构上讲，它不智能地与消费进程共享同一个容器

我们最终使用 rebus 提供的 BuiltinHandlerActivation 类构建了一个自定义的 ICommunicationListener 特定于总线消费过程。看起来像这样。

 public class ServiceBusCommunicationListener : ICommunicationListener
{
    private BuiltinHandlerActivator activator;

    public async Task<string> OpenAsync(CancellationToken cancellationToken)
    {
        activator = new BuiltinHandlerActivator();
        RegisterHandlers(activator);

        var connectionString = "...";
        var bus = Configure.With(activator)
            .Logging(x => x.Serilog(Log.Logger))
            .Transport(
                t => t.UseAzureServiceBus(connectionStringNameOrConnectionString: connectionString,
                        inputQueueAddress: "input", mode: AzureServiceBusMode.Standard))
            .Options(o => o.SimpleRetryStrategy("error"))
            .Start();

        return connectionString;
    }

    private void RegisterHandlers(BuiltinHandlerActivator builtinHandlerActivator)
    {
        (...)
    }

    public async Task CloseAsync(CancellationToken cancellationToken)
    {
        if (activator != null)
            activator.Dispose();
    }

    public void Abort()
    {
        if (activator != null)
            activator.Dispose();
    }
}

并将 ServicebusCommunicationListner 注册为 ServiceInstanceListener。

internal sealed class ProfileService : StatelessService
{
    public ProfileService(StatelessServiceContext context)
        : base(context)
    { }

    protected override IEnumerable<ServiceInstanceListener> CreateServiceInstanceListeners()
    {
        return new[]
        {
            new ServiceInstanceListener(context => new ServiceBusCommunicationListener()), 
        };
    }
}