【发布时间】：2021-10-27 12:36:17
【问题描述】：
我对 MassTransit 还很陌生，不明白我做错了什么导致以下异常：Messages types must not be System types。
这是我的定义：
[BsonIgnoreExtraElements]
public class ArcProcess : SagaStateMachineInstance, ISagaVersion
{
public Guid CorrelationId { get; set; }
public string CurrentState { get; set; }
public int Version { get; set; }
public Guid ActivationId { get; set; }
}
public static class MessageContracts
{
static bool _initialized;
public static void Initialize()
{
if (_initialized)
return;
GlobalTopology.Send.UseCorrelationId<StartProcessingMessage>(x => x.ActivationId);
GlobalTopology.Send.UseCorrelationId<ReconstructionFinishedMessage>(x => x.ActivationId);
GlobalTopology.Send.UseCorrelationId<ProcessingFinishedMessage>(x => x.ActivationId);
_initialized = true;
}
}
我的 2 个消费者是：
public class StartReconstructionConsumer : IConsumer<StartProcessingMessage>
{
readonly ILogger<StartReconstructionConsumer> _Logger;
private readonly int _DelaySeconds = 5;
public StartReconstructionConsumer(ILogger<StartReconstructionConsumer> logger)
{
_Logger = logger;
}
public async Task Consume(ConsumeContext<StartProcessingMessage> context)
{
var activationId = context.Message.ActivationId;
_Logger.LogInformation($"Received Scan: {activationId}");
await Task.Delay(_DelaySeconds * 1000);
_Logger.LogInformation($"Finish Scan: {activationId}");
await context.Publish<ReconstructionFinishedMessage>(new { ActivationId = activationId });
}
}
public class ProcessingFinishedConsumer : IConsumer<ProcessingFinishedMessage>
{
readonly ILogger<ProcessingFinishedConsumer> _Logger;
public ProcessingFinishedConsumer(ILogger<ProcessingFinishedConsumer> logger)
{
_Logger = logger;
}
public async Task Consume(ConsumeContext<ProcessingFinishedMessage> context)
{
_Logger.LogInformation($"Finish {context.Message.ActivationId}");
await Task.CompletedTask;
}
}
这是 StateMachine 的定义：
public class ArcStateMachine: MassTransitStateMachine<ArcProcess>
{
static ArcStateMachine()
{
MessageContracts.Initialize();
}
public ArcStateMachine()
{
InstanceState(x => x.CurrentState);
Initially(
When(ProcessingStartedEvent)
.Then(context =>
{
Console.WriteLine(">> ProcessingStartedEvent");
context.Instance.ActivationId = context.Data.ActivationId;
})
.TransitionTo(ProcessingStartedState));
During(ProcessingStartedState,
When(ReconstructionFinishedEvent)
.Then(context =>
{
Console.WriteLine(">> ReconstructionFinishedEvent");
context.Instance.ActivationId = context.Data.ActivationId;
})
.Publish(context =>
{
return context.Init<ProcessingFinishedMessage>(new { ActivationId = context.Data.ActivationId });
})
.TransitionTo(ProcessingFinishedState)
.Finalize());
}
public State ProcessingStartedState { get; }
public State ReconstructionStartedState { get; }
public State ReconstructionFinishedState { get; }
public State ProcessingFinishedState { get; }
public Event<StartProcessingMessage> ProcessingStartedEvent { get; }
public Event<ReconstructionStartedMessage> ReconstructionStartedEvent { get; }
public Event<ReconstructionFinishedMessage> ReconstructionFinishedEvent { get; }
public Event<ProcessingFinishedMessage> ProcessingFinishedEvent { get; }
}
MassTransit 的设置如下所示：
var rabbitHost = Configuration["RABBIT_MQ_HOST"];
if (rabbitHost.IsNotEmpty())
{
services.AddMassTransit(cnf =>
{
var connectionString = Configuration["MONGO_DB_CONNECTION_STRING"];
var machine = new ArcStateMachine();
var repository = MongoDbSagaRepository<ArcProcess>.Create(connectionString,
"mongoRepo", "WorkflowState");
cnf.AddConsumer(typeof(StartReconstructionConsumer));
cnf.AddConsumer(typeof(ProcessingFinishedConsumer));
cnf.UsingRabbitMq((context, cfg) =>
{
cfg.Host(new Uri(rabbitHost), hst =>
{
hst.Username("guest");
hst.Password("guest");
});
cfg.ConfigureEndpoints(context);
cfg.ReceiveEndpoint(BusConstants.SagaQueue,
e => e.StateMachineSaga(machine, repository));
});
});
services.AddMassTransitHostedService();
services.AddSwaggerGen(c =>
{
c.SwaggerDoc("v1", new OpenApiInfo { Title = "MyApp", Version = "v1" });
});
}
我有几个问题：
-
实际上何时发布消息作为发布消息的结果？ IE。在我的示例中，
await _BusInstance.Bus.Publish<StartProcessingMessage>(new { ActivationId = id });是从StartReconstructionConsumer使用的 WebApi 调用的，但实际上当状态机开始与Initially(When(ProcessingStartedEvent)...一起操作时？ -
我的处理应确保我已经处于
ProcessingStartedState状态，以便During(ProcessingStartedState, When(ReconstructionFinishedEvent)...正确行事。那么，如何确保在收到StartProcessingMessage时触发的消费者可以发布应该启动During的ReconstructionFinishedMessage？我是否正确构建了消息交换？ -
目前对于
await context.Publish<ReconstructionFinishedMessage>(new { ActivationId = activationId });，我在日志中发现一个异常，指出R-FAULT rabbitmq://localhost/saga.service d4070000-7b3b-704d-0f10-08d99942c959 Nanox.GC.Shared.AppCore.Messages.ReconstructionFinishedMessage ReconCaller.Saga.ArcProcess(00:00:04.1132604)，而消息中的guid实际上是MessageId。我在rabbitmq中的消息被路由到saga.service_error，但Messages types must not be System types: System.Threading.Tasks.Task<Nanox.GC.Shared.AppCore.Messages.ProcessingFinishedMessage> (Parameter 'T')除外。
看来我在这里真的很想念..
我的意图是启动处理，该处理将由几个消费者按顺序处理几个阶段。所以在这里我尝试构建一个简单的状态机，只要有人打电话给StartProcessing，它就会启动，然后每个消费者都会完成它的工作并触发FinishedStepX，这会将状态机提升到一个新的步骤并启动下一个消费者，直到所有处理完成，状态机将报告ProcessingComplete。
感谢您的任何帮助和进步
【问题讨论】：
标签： c# masstransit