masstransit - 派生类型不会发布给 MassTransit 中的消费者

Question

我在为派生类型发布通用消息以及使用 MassTransit v2.8.0 调用处理程序时遇到问题。

如果我发布类型为 HtmlBlockNewMessage 的消息，永远不会调用消费者。如果我发布 ServiceBusMessage对象并将消费者更改为 Consumes<ServiceBusMessage>.Context ，消费者被调用。

派生类型的代码失败。它仅适用于父类型 (ServiceBusMessage)。

类型:

[Serializable]
public class ServiceBusMessage
{
}

[Serializable]
public class ServiceBusResponse
{
    public int ResultCode { get; set; } 
}

// Request
[Serializable]
public class ContentItemMessage : ServiceBusMessage
{
    public string SiteId { get; set; }
    public string PageId { get; set; }
    public string ContentItemId { get; set; }
}

[Serializable]
public class HtmlBlockNewMessage : ContentItemMessage
{
    public string HtmlData { get; set; }
}

// Response
[Serializable]
public class ContentItemMessageResponse : ServiceBusResponse
{
    public string Name { get; set; }
    public string ItemType { get; set; }
    public string ItemHandler { get; set; }       
}

[Serializable]
public class HtmlBlockNewMessageResponse : ContentItemMessageResponse
{
    public string DataId { get; set; }
}

消费者:

public class HtmlBlockConsumer : Consumes<HtmlBlockNewMessage>.Context
{
    private IHtmlDataService htmlDataService;
    public static ILogger Logger { get; set; }

    public HtmlBlockConsumer()
        : this(null)
    {
    }

    public HtmlBlockConsumer(IHtmlDataService htmlDataService)
    {
        Logger = Log4NetLogger.GetLogger();

        this.htmlDataService = htmlDataService ?? IoC.Container.Resolve<IHtmlDataService>();
    }

    public void Consume(IConsumeContext<HtmlBlockNewMessage> message)
    {
    // Do some stuff

        message.Respond(new HtmlBlockNewMessageResponse() { ResultCode = 1 } );
    }

发布者端的总线注册:

        var bus = ServiceBusFactory.New(sbc =>
        {
            sbc.EnableMessageTracing();

            sbc.UseMsmq();
            sbc.VerifyMsmqConfiguration();
            sbc.UseMulticastSubscriptionClient();
            sbc.SetNetwork("Test");
            sbc.UseXmlSerializer();
            sbc.UseControlBus();

            sbc.ReceiveFrom("msmq://localhost/AuctionCMS.Web.Publisher");

            MtServiceBus.ValidateBus(sbc);
        });

        IoC.Container.RegisterInstance(bus);

消费者端的总线注册:

        var bus = ServiceBusFactory.New(sbc =>
        {
            sbc.EnableMessageTracing();

            sbc.UseMsmq();
            sbc.VerifyMsmqConfiguration();
            sbc.UseMulticastSubscriptionClient();
            sbc.SetNetwork("Test");
            sbc.UseXmlSerializer();
            sbc.UseControlBus();

            sbc.ReceiveFrom("msmq://localhost/AuctionCMS.Consumer");

            sbc.Subscribe(subs =>
            {
                // These are being manually registered due to some issues getting
                // StructureMap to scan my assemblies
                subs.Instance(new HtmlBlockConsumer());
                subs.Instance(new BrowserConsumer());
                subs.Instance(new OfferConsumer());
            });
        });

        IoC.Container.RegisterInstance(bus);

我的发布扩展:

    public static TR Publish<T, TR>(this IServiceBus bus, T message) where T : ServiceBusMessage where TR : ServiceBusResponse
    {
        TR response = null;

        IoC.Container.Resolve<IServiceBus>().PublishRequest(message, callback =>
            {
                callback.SetTimeout(10.Seconds());

                try
                {
                    callback.Handle<TR>(m =>
                        {
                            response = m; /
                        });
                }
                catch (Exception ex)
                {

                    throw;
                }
            });

        return response;
    }

调用代码:

// First I create a message specific to the type of action I am performing 

var message = new HtmlBlockNewMessage() { HtmlData = "Hello" }; 

// Then I call a function which accepts a ContentItemMessage and calls Publish

public void AddContentItem(ContentItemMessage message) 
{
    // Do some preprocessing

    // This call times out
    var response = this.auctionCmsServices.Bus.Publish<ContentItemMessage,
          ContentItemMessageResponse>(message);

    // Do some more processing

}

这是异常(exception)

[RequestTimeoutException: Timeout waiting for response, RequestId: 54910000-307f-20cf-c0c2-08d06b31cf6f]
   MassTransit.RequestResponse.RequestImpl`1.Wait() in d:\BuildAgent-03\work\aa063b4295dfc097\src\MassTransit\RequestResponse\RequestImpl.cs:124
   MassTransit.RequestResponseExtensions.PublishRequest(IServiceBus bus, TRequest message, Action`1 configureCallback) in d:\BuildAgent-03\work\aa063b4295dfc097\src\MassTransit\RequestResponseExtensions.cs:31
   AuctionCMS.Framework.ServiceBus.MtServiceBus.Publish(IServiceBus bus, T message) in c:\Users\rick\Documents\Visual Studio 2012\Projects\AuctionCMS\AuctionCMS.Framework\ServiceBus\MtServiceBus.cs:24
   AuctionCMS.Framework.Entity.Page.AddContentItem(ISite site, String zone, Int32 location, ContentItemMessage message) in c:\Users\rick\Documents\Visual Studio 2012\Projects\AuctionCMS\AuctionCMS.Framework\Entity\Page.cs:48
   AuctionCMS.Framework.Entity.Site.SetDefaultContent() in c:\Users\rick\Documents\Visual Studio 2012\Projects\AuctionCMS\AuctionCMS.Framework\Entity\Site.cs:117
   AuctionCMS.Web.Controllers.AdminSitesController.NewSite(SiteNewModel model, HttpPostedFileBase file) in c:\Users\rick\Documents\Visual Studio 2012\Projects\AuctionCMS\AuctionCMS.Web\Controllers\AdminSitesController.cs:69
   lambda_method(Closure , ControllerBase , Object[] ) +179
   System.Web.Mvc.ReflectedActionDescriptor.Execute(ControllerContext controllerContext, IDictionary`2 parameters) +261
   System.Web.Mvc.ControllerActionInvoker.InvokeActionMethod(ControllerContext controllerContext, ActionDescriptor actionDescriptor, IDictionary`2 parameters) +39
   System.Web.Mvc.Async.<>c__DisplayClass42.<BeginInvokeSynchronousActionMethod>b__41() +34
   System.Web.Mvc.Async.<>c__DisplayClass39.<BeginInvokeActionMethodWithFilters>b__33() +124
   System.Web.Mvc.Async.<>c__DisplayClass4f.<InvokeActionMethodFilterAsynchronously>b__49() +838059
   System.Web.Mvc.Async.<>c__DisplayClass4f.<InvokeActionMethodFilterAsynchronously>b__49() +838059
   System.Web.Mvc.Async.<>c__DisplayClass4f.<InvokeActionMethodFilterAsynchronously>b__49() +838059
   System.Web.Mvc.Async.<>c__DisplayClass4f.<InvokeActionMethodFilterAsynchronously>b__49() +838059
   System.Web.Mvc.Async.<>c__DisplayClass37.<BeginInvokeActionMethodWithFilters>b__36(IAsyncResult asyncResult) +15
   System.Web.Mvc.Async.<>c__DisplayClass2a.<BeginInvokeAction>b__20() +33
   System.Web.Mvc.Async.<>c__DisplayClass25.<BeginInvokeAction>b__22(IAsyncResult asyncResult) +838644
   System.Web.Mvc.<>c__DisplayClass1d.<BeginExecuteCore>b__18(IAsyncResult asyncResult) +28
   System.Web.Mvc.Async.<>c__DisplayClass4.<MakeVoidDelegate>b__3(IAsyncResult ar) +15
   System.Web.Mvc.Controller.EndExecuteCore(IAsyncResult asyncResult) +65
   System.Web.Mvc.Async.<>c__DisplayClass4.<MakeVoidDelegate>b__3(IAsyncResult ar) +15
   System.Web.Mvc.Controller.EndExecute(IAsyncResult asyncResult) +51
   System.Web.Mvc.<>c__DisplayClass8.<BeginProcessRequest>b__3(IAsyncResult asyncResult) +42
   System.Web.Mvc.Async.<>c__DisplayClass4.<MakeVoidDelegate>b__3(IAsyncResult ar) +15
   System.Web.Mvc.MvcHandler.EndProcessRequest(IAsyncResult asyncResult) +51
   System.Web.CallHandlerExecutionStep.System.Web.HttpApplication.IExecutionStep.Execute() +606
   System.Web.HttpApplication.ExecuteStep(IExecutionStep step, Boolean& completedSynchronously) +288

编辑:

我采用了一种通用方法来解决这个问题。从调用者的角度来看，这有点丑陋，但它确实有效。

   public TR AddContentItem<T, TR>(T message) where T : ContentItemMessage where TR : ContentItemMessageResponse
    {
        var response = this.auctionCmsServices.Bus.Publish<T, TR>(message);

        return response;
    }

调用代码现在如下所示:

page.AddContentItem(new HtmlBlockNewMessage() { HtmlData = "This is some html" });

Answer 1

Eugene 的评论是正确的。这里发生的事情是您正在发布 ContentItemMessage 类型的消息。 HtmlBlockNewMessage 的使用者将不会执行，因为消息是作为 ContentItemMessage 和 ServiceBusMessage 发布的。 MassTransit message mis-typing这是关于它如何工作的众多事情之一。

您的选择:

1. 更改 AddContentItem 以使用通用的，可能带有约束
2. 使用反射调用具有正确类型信息的 Publish
3. 重组发布内容的方式，让这不再是问题

最重要的是，您应该始终以您希望收到的类型发布。消息传递中的多态性很棘手。