MassTransit发送消息CPU占满问题咨询及CPU消耗原理询问

是否正常？ #

这种情况完全不正常。即便在Azure Function环境下，发送2000条消息也不应该导致CPU持续100%占用长达15分钟——正常场景下这类操作应该在数秒到数分钟内完成，且CPU占用不会长时间处于满载状态。

MassTransit的CPU消耗点 #

MassTransit的CPU消耗主要集中在以下几个环节：

• 消息序列化/反序列化：默认的JSON序列化在高频消息处理时可能产生较高CPU开销，如果消息体结构复杂或体积较大，这部分消耗会更明显。
• 资源创建与销毁：频繁创建PublishEndpoint实例（如代码中每次调用DoJob都通过Build()生成新的服务提供者）会触发大量的资源初始化、连接建立与回收逻辑，重复操作会显著占用CPU。
• 异步调度与线程切换：虽然代码使用了ConfigureAwait(false)，但如果Azure Function宿主的线程调度策略与MassTransit的异步处理逻辑不匹配，会导致频繁的线程上下文切换，额外消耗CPU资源。
• 单条消息的处理开销：逐条发送消息时，每条消息都会触发独立的路由解析、消息校验、传输层交互逻辑，累计起来的额外开销会被放大。

针对代码的优化建议 #

从你提供的代码来看，可先从以下几点优化：

1. 复用IPublishEndpoint实例：不要每次调用DoJob都通过_publishEndpointProvider.Build()创建新的服务提供者，直接在类的构造函数中注入IPublishEndpoint并复用，避免重复创建资源。优化后的代码示例：

private readonly IPublishEndpoint _publishEndpoint;

public YourClass(IPublishEndpoint publishEndpoint)
{
    _publishEndpoint = publishEndpoint;
}

public async Task DoJob(string contextName, RequestDto request) 
{
    foreach (var reference in references)
    {
        await _publishEndpoint.Publish(new ProcessMessage
        {
            MessageCorrelationId = Guid.NewGuid().ToString(),
            MessageCreationTimestamp = DateTimeOffset.UtcNow,
            ContextName = contextName,
            StartDate = request.StartDate,
            EndDate = request.EndDate,
            Reference = reference
        }).ConfigureAwait(false);
    }
}

2. 使用批量发布：将消息收集到集合中，使用MassTransit的批量发布方法（如PublishBatch），减少单条消息处理的重复开销：

var messages = references.Select(reference => new ProcessMessage
{
    MessageCorrelationId = Guid.NewGuid().ToString(),
    MessageCreationTimestamp = DateTimeOffset.UtcNow,
    ContextName = contextName,
    StartDate = request.StartDate,
    EndDate = request.EndDate,
    Reference = reference
}).ToList();

await _publishEndpoint.PublishBatch(messages).ConfigureAwait(false);

3. 更换高效序列化器：替换默认的JSON序列化器为MessagePack等更高效的序列化方案，降低序列化阶段的CPU消耗。
4. 检查Azure Function配置：调整Function的并发执行限制，避免因并发过高导致的CPU资源竞争；同时检查MassTransit与Azure服务的连接配置，合理设置连接数、预取数等参数。

进一步排查方向 #

• 使用性能分析工具（如Azure Application Insights、dotTrace）捕获CPU满载期间的调用栈，定位具体是MassTransit的哪段代码占用了CPU。
• 查看MassTransit的详细日志，确认是否存在重试循环、连接错误、路由解析异常等情况，这些都可能导致CPU持续高占用。
• 测试不同消息数量（如100、500条）的CPU占用情况，观察是否存在阈值效应，帮助定位问题根源。

内容的提问来源于stack exchange，提问作者AsyaK