MassTransit RoutingSlip中MessageData跨活动自动处理异常解决方案问询

根因说明 #

7.2.x及更早版本的MassTransit未实现RoutingSlip活动间传递数据的MessageData自动处理能力：只有首次发送到RoutingSlip的入口消息会触发全局MessageData序列化逻辑，活动执行过程中输出的Log、活动间传递的变量，都不会触发MessageData的自动持久化、序列化/反序列化流程，你遇到的4类异常均是该缺陷导致。

解决方案 #

方案1：版本升级（推荐） #

直接升级MassTransit到8.0及以上版本，该版本已完整支持RoutingSlip场景下的MessageData自动处理：

• 无需手动调用IMessageDataRepository，只要属性定义为MessageData<T>类型，自动根据MessageDataDefaults的配置判断是内联还是持久化到仓库
• 活动间传递的Log和Variables都会自动做MessageData的序列化、反序列化，不需要额外处理
• 支持直接定义MessageData<List<Guid>>这类具体对象类型，不用手动序列化/反序列化Stream，代码可大幅简化：

public interface Log
{
    MessageData<List<Guid>> Ids { get; set; }
}

public interface Arguments
{
    List<Guid> Ids { get; set; }
}

public class MyActivity : IActivity<Arguments, Log>
{
    public async Task<CompensationResult> Compensate(CompensateContext<Log> context) 
    {
        var ids = await context.Log.Ids.Value;
        return context.Compensated();
    }
    
    public async Task<ExecutionResult> Execute(ExecuteContext<Arguments> context)
    {
       return context.CompletedWithVariables(
          new Log { Ids = context.Arguments.Ids },
          new { Ids = context.Arguments.Ids });
    }
}

public interface SecondArguments
{
    MessageData<List<Guid>> Ids { get; set; }
}

public class SecondActivity: IActivity<SecondArguments>
{       
    public async Task<ExecutionResult> Execute(ExecuteContext<SecondArguments> context)
    {
       var ids = await context.Arguments.Ids.Value;
       return context.Completed();
    }
}

方案2：7.2.4版本兼容方案 #

如果不能升级版本，可基于你当前的临时方案做优化，保证稳定性：

• 全局固定配置：

MessageDataDefaults.AlwaysWriteToRepository = true;
MessageDataDefaults.Threshold = 0;

强制所有MessageData都持久化到仓库，不走内联逻辑，避免内联对象反序列化失败的问题

• 存储数据时统一调用IMessageDataRepository.PutObject方法，不要直接赋值原始对象到MessageData属性
• 读取数据时统一先获取Value属性再做反序列化，保持你现有代码的写法即可
• 小体积数据直接走普通变量传递，不要用MessageData存储，减少不必要的仓库IO开销

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