masstransit - MassTransit 请求/响应场景中的超时发布消息

Question

我正在尝试设置 MassTransit 请求/响应场景。问题是消息永远不会到达消费者。我在 PublishRequest 上收到“超时等待响应”错误。日志文件中没有显示其他错误。正在 msmq 中创建消息。

异常信息:

Exception type: TargetInvocationException Exception message: Exception has been thrown by the target of an invocation. at System.RuntimeTypeHandle.CreateInstance(RuntimeType type, Boolean publicOnly, Boolean noCheck, Boolean& canBeCached, RuntimeMethodHandleInternal& ctor, Boolean& bNeedSecurityCheck) at System.RuntimeType.CreateInstanceSlow(Boolean publicOnly, Boolean skipCheckThis, Boolean fillCache, StackCrawlMark& stackMark) at System.RuntimeType.CreateInstanceDefaultCtor(Boolean publicOnly, Boolean skipCheckThis, Boolean fillCache, StackCrawlMark& stackMark)
at System.Activator.CreateInstance(Type type, Boolean nonPublic) at System.Activator.CreateInstance(Type type) at System.Web.Mvc.DefaultControllerFactory.DefaultControllerActivator.Create(RequestContext requestContext, Type controllerType)

Timeout waiting for response, RequestId: 08cfa243-4a88-ba3a-20cf-307f54910000 at MassTransit.RequestResponse.RequestImpl1.Wait() in
d:\BuildAgent-03\work\8d1373c869590c5b\src\MassTransit\RequestResponse\RequestImpl.cs:line
124 at
MassTransit.RequestResponseExtensions.PublishRequest[TRequest](IServiceBus
bus, TRequest message, Action1 configureCallback) in d:\BuildAgent-03\work\8d1373c869590c5b\src\MassTransit\RequestResponseExtensions.cs:line 31 at Producer.Website.Controllers.AccountController..ctor() in c:\Users\rick\Documents\Visual Studio 2012\Projects\ConsumerTest1\Producer.Website\Controllers\AccountController.cs:line 56

生产者设置:

            _bus = ServiceBusFactory.New(sbc =>
            {
                sbc.UseMsmq();
                sbc.VerifyMsmqConfiguration();

                sbc.UseMulticastSubscriptionClient();
                sbc.SetNetwork("Test");

               sbc.ReceiveFrom("msmq://localhost/consumer_test_1");

            });

生产者发送消息:

                var message = new AccountNewMessage()
                {
                    CorrelationId = CombGuid.Generate(),
                    UserName = “blah blah”,
                    Password = “yada yada”
                };

                this._bus.PublishRequest(message, r =>
                    {
                        r.SetTimeout(30.Seconds());

                        r.Handle<AccountNewMessageResponse>(m =>
                            {
                                var response = m;
                            });
                    });

消费者设置:

        this.bus = ServiceBusFactory.New(sbc =>
            {
                sbc.UseMsmq();
                sbc.VerifyMsmqConfiguration();

                sbc.UseMulticastSubscriptionClient();
                sbc.SetNetwork("Test");

                sbc.ReceiveFrom("msmq://localhost/consumer_test_2");

                sbc.Subscribe(subs => subs.Instance(new AccountNewMessageConsumer()));
            });

消费者:

public class AccountNewMessageConsumer : Consumes<AccountNewMessage>.Context
{
    public void Consume(IConsumeContext<AccountNewMessage> context)
    {
        context.Respond(new AccountNewMessageResponse()
        {
            CorrelationId = context.Message.CorrelationId,
            ErrorCode = "1",
            UserId = new Random().Next(1, 10000).ToString()
        });
    }
}

消息:

[Serializable]
public class AccountNewMessage : CorrelatedBy<Guid>
{
    public Guid CorrelationId { get; set; }
    public string UserName { get; set; }
    public string Password { get; set; }
}

[Serializable]
public class AccountNewMessageResponse : CorrelatedBy<Guid>
{
    public Guid CorrelationId { get; set; }
    public string UserId { get; set; }
    public string ErrorCode { get; set; }
}

我做错了什么？谢谢。

Answer 1

您必须告诉 MassTransit 如何存储它的订阅信息(对于 MSMQ)。

有两种选择:MSMQ Multicast，将订阅信息保存在内存中；MSMQ Runtime Services，将订阅信息存储在数据库中，因此它在 session 之间持续存在.

您选择使用哪个取决于您需要永久订阅还是临时订阅 - 来自 the docs :

Permanent subscriptions represent a subscription that you want to have stay around even if your process is shut down (maybe you are doing an upgrade and don’t want to miss a message). A temporary subscription is to be used in the case where you won’t care if you miss a message while shut down.

您没有理由不能在开发/PoC 期间使用多播并稍后切换到 RuntimeServices。文档页面还展示了如何在配置中设置每个，当然，对于 RuntimeServices，您还必须设置数据库。

(请注意，多播可能需要一些时间才能自行设置，因此您可能需要暂停测试系统以预热，然后再开始通过它发送消息。)

编辑:我附加了几个验证订阅和消费者的例程:您可以在设置订阅时调用第一个(即在 sbc.Subscribe 期间)，在配置总线后调用第二个。也许他们会帮助找到问题？

private void ValidateSubscriptions(Configurator configurator)
{
    var errors = configurator.Validate();

    Console.WriteLine("Subscription Validation");
    Console.WriteLine("-----------------------");

    foreach (var err in errors.Where(e => string.IsNullOrEmpty(e.Value) == false))
    {
        Console.WriteLine("Type: {0} Message: {1} Key: {2} Value: {3}",
            err.Disposition, err.Message, err.Key, err.Value);
    }
}

private static void ValidateBus(Configurator bus)
{
    var errors = bus.Validate();

    Console.WriteLine("Consumer Validation");
    Console.WriteLine("-------------------");

    foreach (var err in errors.Where(e => string.IsNullOrEmpty(e.Value) == false))
    {
        Console.WriteLine("Type: {0} Message: {1} Key: {2} Value: {3}",
            err.Disposition, err.Message, err.Key, err.Value);
    }
}