5

我正在尝试使用带有 RabbitMQ 的 Automatonymous 为状态机实现一个简单的示例/演示。不幸的是，我找不到一个可以重建/学习的东西（我找到了ShoppingWeb，但在我看来它绝非简单）。在我看来，文档也缺乏信息。

这是我想到的状态机示例（对不起，它很丑）： 在此处输入图像描述 请注意，这个示例完全是虚构的，是否有意义并不重要。这个项目的目的是让 Automatonymous 变得“温暖”。

我想做/拥有的是：

  • 四个应用程序运行：
    1. 状态机本身
    2. “请求者”发送请求被解释
    3. “验证器”或“解析器”检查提供的请求是否有效
    4. 解释给定请求的“解释器”
  • 这方面的一个例子可能是：
    • 请求者发送“x=5”
    • 验证器检查是否包含“=”
    • 口译员说“5”

我的状态机实现如下所示：

public class InterpreterStateMachine : MassTransitStateMachine<InterpreterInstance>
    {
        public InterpreterStateMachine()
        {
            InstanceState(x => x.CurrentState);
            Event(() => Requesting, x => x.CorrelateBy(request => request.Request.RequestString, context => context.Message.Request.RequestString)
                .SelectId(context => Guid.NewGuid())); 
            Event(() => Validating, x => x.CorrelateBy(request => request.Request.RequestString, context => context.Message.Request.RequestString));
            Event(() => Interpreting, x => x.CorrelateBy(request => request.Request.RequestString, context => context.Message.Request.RequestString));

            Initially(
                When(Requesting)
                    .Then(context =>
                    {
                        context.Instance.Request = new Request(context.Data.Request.RequestString);                        
                    })
                    .ThenAsync(context => Console.Out.WriteLineAsync($"Request received: {context.Data.Request.RequestString}"))
                    .Publish(context => new ValidationNeededEvent(context.Instance))
                    .TransitionTo(Requested)
                );

            During(Requested,
                When(Validating)
                    .Then(context =>
                    {
                        context.Instance.Request.IsValid = context.Data.Request.IsValid;
                        if (!context.Data.Request.IsValid)
                        {
                            this.TransitionToState(context.Instance, Error);
                        }
                        else
                        {
                            this.TransitionToState(context.Instance, RequestValid);
                        }
                    })
                    .ThenAsync(context => Console.Out.WriteLineAsync($"Request '{context.Data.Request.RequestString}' validated with {context.Instance.Request.IsValid}"))
                    .Publish(context => new InterpretationNeededEvent(context.Instance))
                    ,
                Ignore(Requesting),
                Ignore(Interpreting)
                );

            During(RequestValid,
                When(Interpreting)
                    .Then((context) =>
                    {
                        //do something
                    })
                    .ThenAsync(context => Console.Out.WriteLineAsync($"Request '{context.Data.Request.RequestString}' interpreted with {context.Data.Answer}"))
                    .Publish(context => new AnswerReadyEvent(context.Instance))
                    .TransitionTo(AnswerReady)
                    .Finalize(),
                Ignore(Requesting),
                Ignore(Validating)
                );

            SetCompletedWhenFinalized();
        }

        public State Requested { get; private set; }
        public State RequestValid { get; private set; }
        public State AnswerReady { get; private set; }
        public State Error { get; private set; }

        //Someone is sending a request to interprete
        public Event<IRequesting> Requesting { get; private set; }
        //Request is validated
        public Event<IValidating> Validating { get; private set; }
        //Request is interpreted
        public Event<IInterpreting> Interpreting { get; private set; }


        class ValidationNeededEvent : IValidationNeeded
        {
            readonly InterpreterInstance _instance;

            public ValidationNeededEvent(InterpreterInstance instance)
            {
                _instance = instance;
            }

            public Guid RequestId => _instance.CorrelationId;

            public Request Request => _instance.Request;
        }

        class InterpretationNeededEvent : IInterpretationNeeded
        {
            readonly InterpreterInstance _instance;

            public InterpretationNeededEvent(InterpreterInstance instance)
            {
                _instance = instance;
            }

            public Guid RequestId => _instance.CorrelationId;
        }

        class AnswerReadyEvent : IAnswerReady
        {
            readonly InterpreterInstance _instance;

            public AnswerReadyEvent(InterpreterInstance instance)
            {
                _instance = instance;
            }

            public Guid RequestId => _instance.CorrelationId;
        }    
    }

然后我有这样的服务：

public class RequestService : ServiceControl
    {
        readonly IScheduler scheduler;
        IBusControl busControl;
        BusHandle busHandle;
        InterpreterStateMachine machine;
        InMemorySagaRepository<InterpreterInstance> repository;

        public RequestService()
        {
            scheduler = CreateScheduler();
        }

        public bool Start(HostControl hostControl)
        {
            Console.WriteLine("Creating bus...");

            machine = new InterpreterStateMachine();
            repository = new InMemorySagaRepository<InterpreterInstance>();


            busControl = Bus.Factory.CreateUsingRabbitMq(x =>
            {
                IRabbitMqHost host = x.Host(new Uri(/*rabbitMQ server*/), h =>
                {
                    /*credentials*/
                });

                x.UseInMemoryScheduler();

                x.ReceiveEndpoint(host, "interpreting_answer", e =>
                {
                    e.PrefetchCount = 5; //?
                    e.StateMachineSaga(machine, repository);
                });

                x.ReceiveEndpoint(host, "2", e =>
                {
                    e.PrefetchCount = 1;
                    x.UseMessageScheduler(e.InputAddress);

                    //Scheduling !?

                    e.Consumer(() => new ScheduleMessageConsumer(scheduler));
                    e.Consumer(() => new CancelScheduledMessageConsumer(scheduler));
                });

            });

            Console.WriteLine("Starting bus...");

            try
            {
                busHandle = MassTransit.Util.TaskUtil.Await<BusHandle>(() => busControl.StartAsync());
                scheduler.JobFactory = new MassTransitJobFactory(busControl);
                scheduler.Start();
            }
            catch (Exception)
            {
                scheduler.Shutdown();
                throw;
            }

            return true;
        }

        public bool Stop(HostControl hostControl)
        {
            Console.WriteLine("Stopping bus...");

            scheduler.Standby();

            if (busHandle != null) busHandle.Stop();

            scheduler.Shutdown();

            return true;
        }

        static IScheduler CreateScheduler()
        {
            ISchedulerFactory schedulerFactory = new StdSchedulerFactory();
            IScheduler scheduler = MassTransit.Util.TaskUtil.Await<IScheduler>(() => schedulerFactory.GetScheduler()); ;

            return scheduler;
        }
    }

我的问题是：

  1. 如何发送“初始”请求，以便状态机转换到我的初始状态
  2. 我如何在消费者中“做出反应”以检查已发送的数据，然后像 1 中一样发送新数据？
4

1 回答 1

7

好吧，我想通了。我可能遇到了问题，因为我不仅是 Masstransit/Automatonymous 和 RabbitMQ 的新手，而且还没有太多 C# 经验。

因此，如果有人遇到同样的问题，这就是您所需要的：鉴于上面的示例，需要三种不同的类型以及一些小接口：

  1. 包括特定消费者的发送者（在本例中为“请求者”）
  2. 使用特定消息类型的服务（“验证器”和“解释器”）
  3. 在没有特定消费者的情况下持有状态机的服务
  4. 一些“合同”，它们是定义发送/使用的消息类型的接口

1）这是发件人：

    using InterpreterStateMachine.Contracts;
    using MassTransit;
    using System;
    using System.Threading.Tasks;

    namespace InterpreterStateMachine.Requester
    {
        class Program
        {
            private static IBusControl _busControl;

            static void Main(string[] args)
            {            
                var busControl = ConfigureBus();
                busControl.Start();

                Console.WriteLine("Enter request or quit to exit: ");
                while (true)
                {
                    Console.Write("> ");
                    String value = Console.ReadLine();

                    if ("quit".Equals(value,StringComparison.OrdinalIgnoreCase))
                        break;

                    if (value != null)
                    {
                        String[] values = value.Split(';');

                        foreach (String v in values)
                        {
                            busControl.Publish<IRequesting>(new
                            {
                                Request = new Request(v),
                                TimeStamp = DateTime.UtcNow
                            });
                        }
                    }
                }

                busControl.Stop();
            }


            static IBusControl ConfigureBus()
            {
                if (null == _busControl)
                {
                    _busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
                    {                    
                        var host = cfg.Host(new Uri(/*rabbitMQ server*/), h =>
                        {                        
                            /*credentials*/
                        });

                        cfg.ReceiveEndpoint(host, "answer_ready", e =>
                        {
                            e.Durable = true;
                            //here the consumer is registered
                            e.Consumer<AnswerConsumer>();
                        });
                    });
                    _busControl.Start();
                }
                return _busControl;
            }

            //here comes the actual logic of the consumer, which consumes a "contract"
            class AnswerConsumer : IConsumer<IAnswerReady>
            {
                public async Task Consume(ConsumeContext<IAnswerReady> context)
                {
                    await Console.Out.WriteLineAsync($"\nReceived Answer for \"{context.Message.Request.RequestString}\": {context.Message.Answer}.");
                    await Console.Out.WriteAsync(">");
                }
            }        
        }
    }

2）这是服务（这里是验证服务）

using InterpreterStateMachine.Contracts;
using MassTransit;
using MassTransit.QuartzIntegration;
using MassTransit.RabbitMqTransport;
using Quartz;
using Quartz.Impl;
using System;
using System.Threading.Tasks;
using Topshelf;

namespace InterpreterStateMachine.Validator
{
    public class ValidationService : ServiceControl
    {
        readonly IScheduler _scheduler;
        static IBusControl _busControl;
        BusHandle _busHandle;        

        public static IBus Bus => _busControl;

        public ValidationService()
        {
            _scheduler = CreateScheduler();
        }

        public bool Start(HostControl hostControl)
        {
            Console.WriteLine("Creating bus...");

            _busControl = MassTransit.Bus.Factory.CreateUsingRabbitMq(x =>
            {
                IRabbitMqHost host = x.Host(new Uri(/*rabbitMQ server*/), h =>
                {
                    /*credentials*/
                });

                x.UseInMemoryScheduler();
                x.UseMessageScheduler(new Uri(RabbitMqServerAddress));

                x.ReceiveEndpoint(host, "validation_needed", e =>
                {
                    e.PrefetchCount = 1;
                    e.Durable = true;
                    //again this is how the consumer is registered
                    e.Consumer<RequestConsumer>();
                });                               
            });

            Console.WriteLine("Starting bus...");

            try
            {
                _busHandle = MassTransit.Util.TaskUtil.Await<BusHandle>(() => _busControl.StartAsync());
                _scheduler.JobFactory = new MassTransitJobFactory(_busControl);
                _scheduler.Start();
            }
            catch (Exception)
            {
                _scheduler.Shutdown();
                throw;
            }                
            return true;
        }

        public bool Stop(HostControl hostControl)
        {
            Console.WriteLine("Stopping bus...");
            _scheduler.Standby();
            _busHandle?.Stop();
            _scheduler.Shutdown();
            return true;
        }

        static IScheduler CreateScheduler()
        {
            ISchedulerFactory schedulerFactory = new StdSchedulerFactory();
            IScheduler scheduler = MassTransit.Util.TaskUtil.Await<IScheduler>(() => schedulerFactory.GetScheduler());

            return scheduler;
        }
    }

    //again here comes the actual consumer logic, look how the message is re-published after it was checked
    class RequestConsumer : IConsumer<IValidationNeeded>
    {
        public async Task Consume(ConsumeContext<IValidationNeeded> context)
        {
            await Console.Out.WriteLineAsync($"(c) Received {context.Message.Request.RequestString} for validation (Id: {context.Message.RequestId}).");

            context.Message.Request.IsValid = context.Message.Request.RequestString.Contains("=");

            //send the new message on the "old" context
            await context.Publish<IValidating>(new
            {
                Request = context.Message.Request,
                IsValid = context.Message.Request.IsValid,
                TimeStamp = DateTime.UtcNow,
                RequestId = context.Message.RequestId
            });
        }
    }
}

验证器使用合约“IValidationNeeded”，然后发布合约“IValidating”，然后状态机本身将使用该合约（“Validating”事件）。

3）消费者服务和状态机服务的区别在于“ReceiveEndpoint”。这里没有注册消费者，但设置了状态机：

...
InterpreterStateMachine _machine = new InterpreterStateMachine();
InMemorySagaRepository<InterpreterInstance> _repository = new InMemorySagaRepository<InterpreterInstance>();
...
x.ReceiveEndpoint(host, "state_machine", e =>
{
    e.PrefetchCount = 1;
    //here the state machine is set
    e.StateMachineSaga(_machine, _repository);
    e.Durable = false;
});

4）最后但并非最不重要的一点是，合同非常小，看起来像这样：

using System;

namespace InterpreterStateMachine.Contracts
{
    public interface IValidationNeeded
    {
        Guid RequestId { get; }
        Request Request { get; }
    }
}

所以总的来说这很简单，我只需要动动脑筋：D

我希望这会对某人有所帮助。

于 2018-06-06T07:25:53.610 回答