如何使用MassTransit在RabbitMQ中基于标头消费消息？

huangapple
go评论83阅读模式
英文:

How do I consume a message based on a header with MassTransit in RabbitMQ?

问题

我正在尝试在RabbitMQ中使用MassTransit创建一个头部交换（headers exchange），以便消费者只根据消息中的头部信息从特定队列中消费消息。

我配置了生产者部分如下所示：

  1. builder.Services.AddMassTransit(mt =>
  2. {
  3.     mt.SetKebabCaseEndpointNameFormatter();
  5.     mt.UsingRabbitMq((context, cfg) =>
  6.     {
  7.         cfg.Host("localhost", "/", h =>
  8.         {
  9.             h.Username("guest");
  10.             h.Password("guest");
  11.         });
  13.         cfg.Publish<OrderSubmitted>(p =>
  14.         {
  15.             p.ExchangeType = "headers";
  16.         });
  17.     });
  18. });

这是我的消费者配置部分：

  1. builder.Services.AddMassTransit(mt =>
  2. {
  3.     mt.AddConsumer<OrderPickupConsumer>();
  4.     mt.AddConsumer<OrderDeliveryConsumer>();
  6.     mt.UsingRabbitMq((context, cfg) =>
  7.     {
  8.         cfg.Host("localhost", "/", h =>
  9.         {
  10.             h.Username("guest");
  11.             h.Password("guest");
  12.         });
  14.         cfg.ReceiveEndpoint("OrderPickup", re =>
  15.         {
  16.             re.ConfigureConsumer<OrderPickupConsumer>(context);
  17.             re.Bind<OrderSubmitted>(x =>
  18.             {
  19.                 x.ExchangeType = "headers";
  20.                 x.SetBindingArgument("headers", new Dictionary<string, string>
  21.                 {
  22.                     { "Transport", "pickup" },
  23.                     { "x-match", "all" }
  24.                 });
  25.             });
  26.         });
  28.         cfg.ReceiveEndpoint("OrderDelivery", re =>
  29.         {
  30.             re.ConfigureConsumer<OrderDeliveryConsumer>(context);
  31.             re.Bind<OrderSubmitted>(x =>
  32.             {
  33.                 x.ExchangeType = "headers";
  34.                 x.SetBindingArgument("headers", new Dictionary<string, string>
  35.                 {
  36.                     { "Transport", "delivery" },
  37.                     { "x-match", "all" }
  38.                 });
  39.             });
  40.         });
  41.     });
  42. });

我发布消息的方式如下：

  1. _bus.Publish<OrderSubmitted>(new
  2. {
  3.     __Header_Transport = "pickup",
  4.     Product = "Pizza"
  5. });
  7. _bus.Publish<OrderSubmitted>(new
  8. {
  9.     __Header_Transport = "delivery",
  10.     Product = "Burgers"
  11. });

就我理解，上述设置是正确的，然而，当启动接收应用程序时出现错误：
ArgumentException: The MassTransit.RabbitMqTransport.Topology.ExchangeEntity entity settings did not match the existing entity

我有一个干净的RabbitMQ实例，没有任何先前存在的队列，甚至当RabbitMQ根本没有启动时也会出现错误。

我的配置是否有问题？或者这可能是MassTransit中的一个bug吗？

英文:

I am trying to create a headers exchange in RabbitMQ with MassTransit, so a consumer only consumes messages from a specific queue based on headers in the message.

I configured my producer:

  1. builder.Services.AddMassTransit(mt =&gt;
  2. {
  3.     mt.SetKebabCaseEndpointNameFormatter();
  5.     mt.UsingRabbitMq((context, cfg) =&gt;
  6.     {
  7.         cfg.Host(&quot;localhost&quot;, &quot;/&quot;, h =&gt;
  8.         {
  9.             h.Username(&quot;guest&quot;);
  10.             h.Password(&quot;guest&quot;);
  11.         });
  13.         cfg.Publish&lt;OrderSubmitted&gt;(p =&gt;
  14.         {
  15.             p.ExchangeType = &quot;headers&quot;;
  16.         });
  17.     });
  18. });

This is my consumer configuration:

  1. builder.Services.AddMassTransit(mt =&gt;
  2. {
  3.     mt.AddConsumer&lt;OrderPickupConsumer&gt;();
  4.     mt.AddConsumer&lt;OrderDeliveryConsumer&gt;();
  6.     mt.UsingRabbitMq((context, cfg) =&gt;
  7.     {
  8.         cfg.Host(&quot;localhost&quot;, &quot;/&quot;, h =&gt;
  9.         {
  10.             h.Username(&quot;guest&quot;);
  11.             h.Password(&quot;guest&quot;);
  12.         });
  14.         cfg.ReceiveEndpoint(&quot;OrderPickup&quot;, re =&gt;
  15.         {
  16.             re.ConfigureConsumer&lt;OrderPickupConsumer&gt;(context);
  17.             re.Bind&lt;OrderSubmitted&gt;(x =&gt;
  18.             {
  19.                 x.ExchangeType = &quot;headers&quot;;
  20.                 x.SetBindingArgument(&quot;headers&quot;, new Dictionary&lt;string, string&gt;
  21.                 {
  22.                     { &quot;Transport&quot;, &quot;pickup&quot; },
  23.                     { &quot;x-match&quot;, &quot;all&quot; }
  24.                 });
  25.             });
  26.         });
  28.         cfg.ReceiveEndpoint(&quot;OrderDelivery&quot;, re =&gt;
  29.         {
  30.             re.ConfigureConsumer&lt;OrderDeliveryConsumer&gt;(context);
  31.             re.Bind&lt;OrderSubmitted&gt;(x =&gt;
  32.             {
  33.                 x.ExchangeType = &quot;headers&quot;;
  34.                 x.SetBindingArgument(&quot;headers&quot;, new Dictionary&lt;string, string&gt;
  35.                 {
  36.                     { &quot;Transport&quot;, &quot;delivery&quot; },
  37.                     { &quot;x-match&quot;, &quot;all&quot; }
  38.                 });
  39.             });
  40.         });
  41.     });
  42. });

I publish the messages like this:

  1. _bus.Publish&lt;OrderSubmitted&gt;(new
  2.             {
  3.                 __Header_Transport = &quot;pickup&quot;,
  4.                 Product = &quot;Pizza&quot;
  5.             });
  7.             _bus.Publish&lt;OrderSubmitted&gt;(new
  8.             {
  9.                 __Header_Transport = &quot;delivery&quot;,
  10.                 Product = &quot;Burgers&quot;
  11.             });

As far as I understand the setup above is correct, however when starting the receiving application an error is thrown:
ArgumentException: The MassTransit.RabbitMqTransport.Topology.ExchangeEntity entity settings did not match the existing entity

I have a clean RabbitMQ instance without any previously existing queues and even when RabbitMQ is not started at all I also get the error.

Is there something wrong with my configuration? Or might this be a bug in MassTransit?

答案1

得分: 0

在一天的调查后，我发现当我添加了 re.ConfigureConsumeTopology = false; 时，错误没有显示。

然而，消息未传递到队列。为了解决这个问题，我更改了 SetBindingArgument 配置。

  1. cfg.ReceiveEndpoint("OrderPickup", re =>
  2.         {
  3.             re.ConfigureConsumeTopology = false;
  4.             re.ConfigureConsumer<OrderPickupConsumer>(context);
  5.             re.Bind<OrderSubmitted>(x =>
  6.             {
  7.                 x.ExchangeType = "headers";
  8.                 x.SetBindingArgument("Transport", "pickup");
  9.                 x.SetBindingArgument("x-match", "all");
  10.             });
  11.         });
  13.         cfg.ReceiveEndpoint("OrderDelivery", re =>
  14.         {
  15.             re.ConfigureConsumeTopology = false;
  16.             re.ConfigureConsumer<OrderDeliveryConsumer>(context);
  17.             re.Bind<OrderSubmitted>(x =>
  18.             {
  19.                 x.ExchangeType = "headers";
  20.                 x.SetBindingArgument("Transport", "delivery");
  21.                 x.SetBindingArgument("x-match", "all");
  22.             });
  23.         });
  24. ```现在消息根据标头过滤器发送到正确的队列，并由消费者接收。
  26. <details>
  27. <summary>英文:</summary>
  29. After a day of investigation I found that the error did not show when I added `re.ConfigureConsumeTopology = false;`
  31. However, the message was not delivered to the queue. To fix this I changed the `SetBindingArgument` configuration.

cfg.ReceiveEndpoint("OrderPickup", re =>
{
re.ConfigureConsumeTopology = false;
re.ConfigureConsumer<OrderPickupConsumer>(context);
re.Bind<OrderSubmitted>(x =>
{
x.ExchangeType = "headers";
x.SetBindingArgument("Transport", "pickup");
x.SetBindingArgument("x-match", "all");
});
});

  1.     cfg.ReceiveEndpoint(&quot;OrderDelivery&quot;, re =&gt;
  2.     {
  3.         re.ConfigureConsumeTopology = false;
  4.         re.ConfigureConsumer&lt;OrderDeliveryConsumer&gt;(context);
  5.         re.Bind&lt;OrderSubmitted&gt;(x =&gt;
  6.         {
  7.             x.ExchangeType = &quot;headers&quot;;
  8.             x.SetBindingArgument(&quot;Transport&quot;, &quot;delivery&quot;);
  9.             x.SetBindingArgument(&quot;x-match&quot;, &quot;all&quot;);
  10.         });
  11.     });
  2. Now the messages are sent to the correct queue, based on the header filter and pickup by the consumer.
  4. </details>

huangapple
  • 本文由 发表于 2023年6月22日 15:10:40
  • 转载请务必保留本文链接：https://go.coder-hub.com/76529371.html
  • .net-core
  • masstransit
  • rabbitmq
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定