RabbitMq - ConversationId 与 CorrelationId - 哪个更适合跟踪特定请求?

Reb*_*cca 4 c# masstransit rabbitmq

RabbitMQ 似乎有两个非常相似的属性,我并不完全理解其中的区别。ConversationIdCorrelationId

我的用例如下。我有一个生成Guid. 该网站调用 API,将该唯一标识符添加到HttpRequest标头中。这反过来向 RabbitMQ 发布消息。该消息由第一个消费者处理并在别处传递给另一个消费者,依此类推。

出于记录目的,我想记录一个将初始请求与所有后续操作联系在一起的标识符。这对于整个应用程序的不同部分的旅程来说应该是独一无二的。因此。当登录到 Serilog/ElasticSearch 之类的东西时,就可以轻松查看哪个请求触发了初始请求,并且整个应用程序中该请求的所有日志条目都可以关联在一起。

我创建了一个提供程序来查看传入HttpRequest的标识符。我将其称为“CorrelationId”,但我开始怀疑是否真的应该将其命名为“ConversationId”。就RabbitMQ而言,“ConversationId”的想法更适合这个模型,还是“CorrelationId”更好?

这两个概念有什么区别?

在代码方面,我希望做到以下几点。首先在我的 API 中注册总线并配置SendPublish使用CorrelationId来自提供者的。

// bus registration in the API
var busSettings = context.Resolve<BusSettings>();
// using AspNetCoreCorrelationIdProvider
var correlationIdProvider = context.Resolve<ICorrelationIdProvider>();

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    cfg.Host(
        new Uri(busSettings.HostAddress),
        h =>
        {
            h.Username(busSettings.Username);
            h.Password(busSettings.Password);
        });
    cfg.ConfigurePublish(x => x.UseSendExecute(sendContext =>
    {
        // which one is more appropriate
        //sendContext.ConversationId = correlationIdProvider.GetCorrelationId();
        sendContext.CorrelationId = correlationIdProvider.GetCorrelationId();
    }));
});
Run Code Online (Sandbox Code Playgroud)

作为参考,这是我的简单提供程序接口

// define the interface
public interface ICorrelationIdProvider
{
    Guid GetCorrelationId();
}
Run Code Online (Sandbox Code Playgroud)

以及 AspNetCore 实现,它提取调用客户端(即网站)设置的唯一 ID。

public class AspNetCoreCorrelationIdProvider : ICorrelationIdProvider
{
    private IHttpContextAccessor _httpContextAccessor;

    public AspNetCoreCorrelationIdProvider(IHttpContextAccessor httpContextAccessor)
    {
        _httpContextAccessor = httpContextAccessor;
    }

    public Guid GetCorrelationId()
    {
        if (_httpContextAccessor.HttpContext.Request.Headers.TryGetValue("correlation-Id", out StringValues headers))
        {
            var header = headers.FirstOrDefault();
            if (Guid.TryParse(header, out Guid headerCorrelationId))
            {
                return headerCorrelationId;
            }
        }

        return Guid.NewGuid();
    }
}
Run Code Online (Sandbox Code Playgroud)

最后,我的服务主机是简单的 Windows 服务应用程序,它们可以使用已发布的消息。他们使用以下内容来获取 CorrelationId,并且很可能会发布给其他消费者以及其他服务主机。

public class MessageContextCorrelationIdProvider : ICorrelationIdProvider
{
    /// <summary>
    /// The consume context
    /// </summary>
    private readonly ConsumeContext _consumeContext;

    /// <summary>
    /// Initializes a new instance of the <see cref="MessageContextCorrelationIdProvider"/> class.
    /// </summary>
    /// <param name="consumeContext">The consume context.</param>
    public MessageContextCorrelationIdProvider(ConsumeContext consumeContext)
    {
        _consumeContext = consumeContext;
    }

    /// <summary>
    /// Gets the correlation identifier.
    /// </summary>
    /// <returns></returns>
    public Guid GetCorrelationId()
    {
        // correlationid or conversationIs?
        if (_consumeContext.CorrelationId.HasValue && _consumeContext.CorrelationId != Guid.Empty)
        {
            return _consumeContext.CorrelationId.Value;
        }

        return Guid.NewGuid();
    }
}
Run Code Online (Sandbox Code Playgroud)

然后我的消费者中有一个记录器,它使用该提供者来提取CorrelationId

public async Task Consume(ConsumeContext<IMyEvent> context)
{
    var correlationId = _correlationProvider.GetCorrelationId();
    _logger.Info(correlationId, $"#### IMyEvent received for customer:{context.Message.CustomerId}");

    try
    {
        await _mediator.Send(new SomeOtherRequest(correlationId) { SomeObject: context.Message.SomeObject });
    }
    catch (Exception e)
    {
        _logger.Exception(e, correlationId, $"Exception:{e}");
        throw;
    }

    _logger.Info(correlationId, $"Finished processing: {DateTime.Now}");
}
Run Code Online (Sandbox Code Playgroud)

阅读文档,它说明了以下关于“ConversationId”的内容:

对话由发送或发布的第一条消息创建,其中没有可用的现有上下文(例如,当使用 IBus.Send 或 IBus.Publish 发送或发布消息时)。如果使用现有上下文来发送或发布消息,则 ConversationId 将复制到新消息中,以确保同一对话中的一组消息具有相同的标识符。

现在我开始认为我的术语混淆了,从技术上讲,这是一次对话(尽管“对话”就像“电话游戏”)。

那么,CorrelationId在这个用例中,还是ConversationId?请帮我弄明白我的术语!!

Chr*_*son 7

在消息对话(提示不祥的乐谱)中,可以有一条消息(我告诉你做某事,或者我告诉所有正在听的人发生了某事)或多条消息(我告诉你做某事,而你告诉其他人,或者我告诉所有在听的人发生了一些事情,而那些听众告诉他们的朋友,等等,等等)。

使用 MassTransit,从第一条消息到最后一条消息,如果使用得当,这些消息中的每一条都将具有相同的ConversationId. MassTransit 将属性从ConsumeContext,未经修改,复制到消息消费期间的每条传出消息。这使得一切都成为同一轨迹的一部分- 一次对话。

但是,MassTransit 默认不设置 CorrelationId。如果消息属性被命名为 CorrelationId(或 CommandId 或 EventId),它可以自动设置,或者您也可以添加自己的名称。

如果使用的消息上存在 CorrelationId,则任何传出的消息都会将该 CorrelationId 属性复制到 InitiatorId 属性(原因和结果 - 使用的消息启动了后续消息的创建)。这形成了一个链(或跨度,在跟踪术语中),可以用来显示从初始消息开始的消息传播。

CorrelationId 应被视为命令或事件的标识符,以便可以在整个系统日志中看到该命令的效果。

在我看来,您来自 HTTP 的输入可能是 Initiator,因此将该标识符复制到 InitiatorId 并为消息创建一个新的 CorrelationId,或者您可能只想对初始 CorrelationId 使用相同的标识符并让后续消息使用它作为发起者。