由于网段，MassTransit 与 RabbitMq 请求/响应错误的回复地址

Question

我有一个在 Masstransit 中使用请求/响应消息的 Web 应用程序。 这适用于测试环境，没问题。

但是，在客户部署方面，我们面临一个问题。在客户站点，我们确实有两个网段 A 和 B。执行数据库调用的组件在网段 A，Web 应用程序和 RabbitMq 服务器在网段 B。

由于安全限制，段 A 中的组件必须通过具有给定地址的负载均衡器。组件本身可以通过 Masstransit 连接到 RabbitMQ。到目前为止一切顺利。

然而，段 B 上的 Web 组件使用 RabbitMq 服务器的直接地址。当 Web 组件现在开始请求/响应调用时，我可以看到消息到达了分段 A 中的组件。 但是我看到消费者试图在“错误”的地址上调用 RabbitMQ 服务器。它使用 Web 组件用来发出请求的地址。但是，A 段中的组件应回复“负载均衡器”地址。

有没有办法配置或告诉 RespondAsync 调用使用为该组件配置的连接地址？

当然，最简单的方法是让 Web 组件也通过负载均衡器连接，但由于网络分段/安全设置，负载均衡器只能从分段 A 访问。

感谢任何输入/帮助。

Answer 1

我在 rabbitmq federation 中遇到了类似的问题。这就是我所做的。

ResponseAddressSendObserver

class ResponseAddressSendObserver : ISendObserver
{
    private readonly string _hostUriString;

    public ResponseAddressSendObserver(string hostUriString)
    {
        _hostUriString = hostUriString;
    }

    public Task PreSend<T>(SendContext<T> context)
        where T : class
    {
        if (context.ResponseAddress != null)
        {
            // Send relative response address alongside the message
            context.Headers.Set("RelativeResponseAddress",
                context.ResponseAddress.AbsoluteUri.Substring(_hostUriString.Length));
        }
        return Task.CompletedTask;
    }
    ...
}

ResponseAddressConsumeFilter

class ResponseAddressConsumeFilter : IFilter<ConsumeContext>
{
    private readonly string _hostUriString;

    public ResponseAddressConsumeFilter(string hostUriString)
    {
        _hostUriString = hostUriString;
    }

    public Task Send(ConsumeContext context, IPipe<ConsumeContext> next)
    {
        var responseAddressOverride = GetResponseAddress(_hostUriString, context);

        return next.Send(new ResponseAddressConsumeContext(responseAddressOverride, context));
    }

    public void Probe(ProbeContext context){}

    private static Uri GetResponseAddress(string host, ConsumeContext context)
    {
        if (context.ResponseAddress == null)
            return context.ResponseAddress;

        object relativeResponseAddress;
        if (!context.Headers.TryGetHeader("RelativeResponseAddress", out relativeResponseAddress) || !(relativeResponseAddress is string))
            throw new InvalidOperationException("Message has ResponseAddress but doen't have RelativeResponseAddress header");

        return new Uri(host + relativeResponseAddress);
    }
}

ResponseAddressConsumeContext

class ResponseAddressConsumeContext : BaseConsumeContext
{
    private readonly ConsumeContext _context;

    public ResponseAddressConsumeContext(Uri responseAddressOverride, ConsumeContext context)
        : base(context.ReceiveContext)
    {
        _context = context;
        ResponseAddress = responseAddressOverride;
    }

    public override Uri ResponseAddress { get; }

    public override bool TryGetMessage<T>(out ConsumeContext<T> consumeContext)
    {
        ConsumeContext<T> context;
        if (_context.TryGetMessage(out context))
        {
            // the most hackish part in the whole arrangement
            consumeContext = new MessageConsumeContext<T>(this, context.Message);
            return true;
        }
        else
        {
            consumeContext = null;
            return false;
        }
    }

    // all other members just delegate to _context 
}

以及配置总线时

var result = MassTransit.Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri(hostAddress), h =>
    {
        h.Username(...);
        h.Password(...);
    });

    cfg.UseFilter(new ResponseAddressConsumeFilter(hostAddress));

    ...
});
result.ConnectSendObserver(new ResponseAddressSendObserver(hostAddress));

所以现在相对响应地址与消息一起发送并在接收端使用。

文档不建议使用 observers 修改任何内容，但在这种情况下应该没问题。

也许三个是更好的解决方案，但我还没有找到一个。高温