问题标签 [masstransit]

我使用 masstransit 作为服务总线，并为我的应用程序使用 pub/sub 模式。

我想知道我是否可以跟踪从公共汽车到订阅者的消息。

我使用 json 序列化程序。

MT 是否有一些“入口点”，我可以在其中获取原始消息正文？

我是 MassTransit 的新手，在我的理解中我错过了一些东西。

假设我有一个服务器场，所有节点都可以做同样的工作。应用程序框架是 CQRS 的风格。这意味着我有两种基本的消息要发布：

• 命令：必须由其中一个服务器处理，其中任何一个（第一个没有工作槽的）
• 事件：必须由所有服务器处理

我已经构建了一个非常简单的 MassTransit 原型（一个每隔 X 秒发送一次问候的控制台应用程序）。

在 API 中，我可以看到有一个“发布”方法。我如何指定它是什么类型的消息（一个与所有服务器）？

如果我查看“处理程序”配置，我可以指定队列 uri。如果我为所有主机指定相同的队列，所有主机都会收到消息，但我不能将执行限制在一台服务器上。

如果我从主机专用队列中侦听，则只有一个服务器会处理消息，但我不知道如何广播另一种消息。

请帮助我了解我所缺少的。

PS：如果它在乎，我的消息系统是rabbitmq。

为了测试，我用这些类创建了一个通用类库：

我还有 3 个只运行 ActualMain 方法的控制台应用程序：

我正在考虑使用MassTransit将事件从一个进程发布到关心该事件的所有其他进程。在我的环境中，订阅该事件的多个进程可以在同一个盒子上运行。

基于 MassTransit 的示例代码，这是一个非常简单的概念验证，它不尝试处理多播（流程也将分布在许多机器上）。将下面的“消息”视为“事件”，并且我希望此过程的每个实例都能获取该消息：

如果我运行此 C# 应用程序的多个实例，则只有一个实例接收到消息，但我需要所有实例来接收它。这种行为与队列的经典概念是一致的，所以我很好。

我想知道的是我是否可以使用 MSMQ/MassTransit 发布一条所有订阅者都会收到的消息，而不是一个订阅者将其出列而其他订阅者不接收它。也许我正在尝试使用锤子并且需要画笔？

在这里要明确一点，我不打算为不同类型的消息共享单个队列，我希望使用 MT 和 MSMQ 为特定类型的消息设置“点对点”发布订阅。

我想在每条消息到达消费者或传奇之前检查它。我想我想要一个 IInboundMessageInterceptor 但我看不到注入自定义的简单方法。如何在 MT 中实现消息拦截？和/或如何使用自定义 IInboundMessageInterceptor 配置总线？

我正在将 MassTransit 视为要在 Web 项目中使用的 ServiceBus 实现。

我正在使用请求/响应模式，并且看到消费者接收消息和响应与请求发布者处理响应之间存在很长的延迟；有时，似乎响应永远不会通过（让它运行了 10 分钟，响应仍然没有通过）。我看到句柄委托被调用并响应的唯一时间是在 30 秒超时时间之后并抛出超时异常；在这种情况下，处理程序委托上设置的断点被命中。

设置是标准事务 - 我有一个正在发布请求的 Web 应用程序，一个正在使用请求和发送响应的控制台应用程序，以便 Web 应用程序处理回调中的响应。

我正在使用 Castle Windsor，并且容器在 Web 项目中使用 WebActivator 进行了初始化：

在 Web 应用程序项目（一个 ASP.NET Web API 项目）中，MassTransit 的 WindsorInstaller 看起来像

在控制台应用程序项目中，WindsorInstaller 看起来像

我有一个ApiController使用以下 GET 操作方法

BasicRequestCommand 和 BasicResponseCommand 看起来像这样

以及响应控制台应用程序中的 BasicRequestCommand 的处理程序：

我预计所有这些都在本地运行，请求/响应最多大约几秒钟。我在配置中遗漏了什么吗？

此外，我想将 MassTransit 连接到 log4net。我正在使用 Windsor 的 log4net 日志记录工具，并且在 web.config 中有一个 log4net 部分。这对于ILoggerWindsor 提供的实现（以及 NHibernate 日志记录）都可以正常工作，但是从文档中不清楚如何配置 MassTransit 以将其用于日志记录。有任何想法吗？

我有一个使用 ServiceStack 构建的 Web API，它通过位于 RabbitMQ 上的 MassTransit 与 TopShelf 服务后端进行通信。大多数时候，一切正常，但有时我们开始看到大量请求开始进入错误队列的问题。

API 比较简单，只有 2 次调用。据我所知，查看日志，主要调用工作正常。然而，另一个调用是用于确定节点状态的调用，并且会导致问题。这两个调用都以相同的方式实现，使用请求/响应方法。

以下是其中一个错误的日志文件：

我确实注意到它在抱怨超时。我们尝试了可变长度，但似乎没有帮助。查看与数据库的连接，消费者中似乎没有任何东西会真正阻止它，所以我不确定超时是由于发送时的多次失败引起的，还是仅仅是连接问题. 通常，响应会在不到一秒的时间内发生。

该请求是通过以下方式提出的：

消费者是：

任何人都可以看到这有什么明显的错误吗？我检查了 API 和服务正在读取不同的队列（错误最终出现在 test_web_error 队列中），并检查了状态请求和响应数据包上是否设置了相关 ID 和 CorrelatedBy。如前所述，这只发生在这个调用中，而不是我们拥有的另一个调用。我也有多台服务器运行它，它似乎一次只发生在一台服务器上，但并不总是同一个服务器。

与直接在RabbitMQ之上构建相比，在MassTransit之上构建有什么好处？

我相信 MassTransit 提供的一个好处是“类型”交换（按接口/类型发布订阅），因此与消息内容是非结构化文本/blob 的普通 RabbitMQ 交换相比，消息的内容是结构化的。

MassTransit 还提供哪些其他好处？

如果您使用相同的端点名称创建发布者和消费者，我们遇到了 MassTransit 丢失消息的情况。

注意下面的代码；如果我为消费者或发布者使用不同的端点名称（例如，发布者为“rabbitmq://localhost/mtlossPublised”），那么消息将同时计算发布和消费匹配；如果我使用相同的端点名称（如示例中），那么我收到的消息比发布的消息少。

这是预期的行为吗？还是我做错了什么，下面的工作示例代码。

我已经实现了很多次都可以正常工作的公共交通传奇。但是，有时消息会进入错误队列或似乎消失了。我正在使用 RabbitMQ。

我想知道： 1. 如何获取导致消息进入错误队列的原因/异常消息？（注意：我的处理程序逻辑位于 try-catch 块中，因此显然这些错误甚至在调用处理程序逻辑之前就发生了） 2. 丢失消息的原因是什么？

提前致谢。

过去 6 个月的学习曲线一直充满挑战，主要罪魁祸首是 CQRS 和 DDD。

这很有趣，我们的项目已经完成了 1/2，而我没有时间深入研究的领域是消息传递框架。

目前我不使用 DTC，所以很有可能如果我的读取模型没有更新，那么我将在读取和写入数据库之间出现不一致。我的读写数据库也将在同一台机器上。我怀疑我们是否会将它们放在不同的机器上。

我的系统中没有大量消息，所以我更关心的是系统的一致性和可靠性。

那么，我是否必须放入像 NServiceBus 这样的消息传递框架（即使读写数据库都在同一台机器上）还是我有其他选择？是的，有学习曲线，但我想如果我不使用它，我会学到很多东西。

另外，如果没有必要，我不想放一层

想法？