标签: message-bus

还有吗？对我来说,MB知道订阅者和发布者,并充当调解者,通知订阅者新消息(实际上是"推送"模型).另一方面,MQ更像是一种"拉"模型,消费者将消息从队列中拉出来.

我完全偏离了这里吗？

最近我一直在阅读很多有关分布式消息传递和相关模式的内容.我使用了一些工具支持的例子,比如例如NServiceBus.

许多这些模式都在互联网上描述.我最近读到的其中一些是:

• Message Broker:http://msdn.microsoft.com/en-us/library/ff648849.aspx
• 消息总线:http://msdn.microsoft.com/en-us/library/ms978583.aspx
• SOA中的消息传递模式:http://msdn.microsoft.com/en-us/library/aa480027.aspx
• Udi Dahan关于差异的帖子:http://www.udidahan.com/2011/03/24/bus-and-broker-pubsub-differences/

如果使用像NService bus这样的工具来做很多工作而不考虑基础设施问题,那么当我尝试实现基本的Message Bus和命令处理程序时,一些问题已经得到了解决.事实上,当谈到这些模式时,我看不出它们之间存在很多差异.

我不会粘贴代码,因为它很长,但我发现了两篇博文,很好地描述了我想谈的实现的想法.

• 消息总线:http://brentedwards.net/2010/04/13/roll-your-own-simple-message-bus-event-aggregator/

这个想法很简单,消息总线跟踪订阅者并在他们感兴趣的情况下将消息发送给不同的订阅者.

• 命令处理程序:http://weblogs.asp.net/shijuvarghese/archive/2011/10/18/cqrs-commands-command-handlers-and-command-dispatcher.aspx

它与消息总线非常相似.命令总线为给定的命令类型调用命令处理程序.

所以在这两种情况下都有相似之处.

使用一种模式比另一种模式有什么真正的差异和好处(我不是在谈论支持工具).我错过了什么？

第二个问题是.没有支持工具,消息总线是否有价值？我不认为自己会为自己的所有权利提供支持.

对于一个冗长而令人困惑的问题我很抱歉,但请不要犹豫,询问更多细节.

我正在试图澄清亚马逊的SQS死信队列究竟在做什么.

据http://aws.typepad.com/aws/2014/01/amazon-sqs-new-dead-letter-queue.html

死信队列 - SQS队列的ARN(亚马逊资源名称),它将接收消费者收到最大数量后未成功处理的消息.

这听起来不像Poision Queue吗？关键的区别在于消费者确实收到了消息.当信息可能正常,但无法传递时,可能是由于服务中断而导致死信.http://www.eaipatterns.com/DeadLetterChannel.html

其中,因为这听起来像被成功地接收到该消息多次,但处理消息失败,这是我理解是有害消息队列的含义.

消息总线与队列

死信模式在普通旧队列的上下文中有不同的含义吗？由于SQS只是一个队列,而不是消息总线,因此它不负责传递消息.相反,它等待拾取(请求)消息.因此传统的死信模式并不真正适用,因为没有消息总线尝试传递消息而无法找到收件人.

SQS可以像消息总线一样吗？

是否有办法通过SQS设置通道和侦听器,而不是显式轮询来自队列的消息？

我正在测试ZeroMQ作为中型系统的Pub-Sub(服务总线式).我们有大约50个节点,所有节点都应该是发布者和订阅者.网络是一种星形拓扑,但边缘相互"交谈".我们需要动态发现(不需要硬编码参与者的网络地址),也没有SPOF(单点故障).

我已经阅读了http://zeromq.org/whitepapers:0mq-3-0-pubsub,据我所知,动态发现的建议0MQ方式涉及转发订阅和发布的代理节点(XPUB/XSUB).我考虑在我们的系统中使用这样的代理作为中央调解器,但是,我对这种架构有以下关注:(A)代理节点是SPOF - 当它失败时整个系统不起作用(B)所有流量,包括数据,通过代理节点,这意味着延迟和性能问题.

假设我正确理解了pub-sub白皮书,是否有一种相对简单的方法可以在ZeroMQ中实现pub-sub + dynamic-discovery + no-SPOF？

补充一点:我已经排除了多播(PGM)解决方案,因为大多数消息都有一个/很少的相关方,我们不喜欢过度拥挤的网络.

我成功地将 MassTransit 用于一个愚蠢的示例应用程序，我从发布者控制台应用程序发布一条消息（一个事件），并在两个不同的消费者处接收它，这些消费者也是使用 RabbitMq 的控制台应用程序。

这是整个示例项目 git repo： https : //gitlab.com/DiegoDrivenDesign/DiDrDe.MessageBus

我想要一个包含 MassTransit 功能的项目，以便我的 Publisher 和 Consumers 项目对 MassTransit 一无所知。依赖关系应该朝这个方向发展：

• DiDrDe.MessageBus ==> MassTransit
• DiDrDe.MessageBus ==> DiDrDe.Contracts
• DiDrDe.Model ==> DiDrDe.Contracts
• DiDrDe.Publisher ==> DiDrDe.MessageBus
• DiDrDe.Publisher ==> DiDrDe.Contracts
• DiDrDe.Publisher ==> DiDrDe.Model
• DiDrDe.ConsumerOne ==> DiDrDe.Contracts
• DiDrDe.ConsumerOne ==> DiDrDe.MessageBus
• DiDrDe.ConsumerOne ==> DiDrDe.Model
• DiDrDe.ConsumerTwo ==> DiDrDe.Contracts
• DiDrDe.ConsumerTwo ==> DiDrDe.MessageBus
• DiDrDe.ConsumerTwo ==> DiDrDe.Model

请注意 DiDrDe.MessageBus 对 DiDrDe.Model 一无所知，因为它是一个通用项目，应该对任何消息类型都有效。

为了实现这一点，我正在实施适配器模式，以便我的自定义接口IEventDtoBus（发布事件）和IEventDtoHandler<TEventDto>（使用事件）是我的发布者和消费者都知道的。MassTransit 包装器项目（称为 DiDrDe.MessageBus）实现了一个EventDtoBusAdapter由 anIEventDtoBus和 a组成的适配器EventDtoHandlerAdapter<TEventDto>作为我唯一的IConsumer<TEventDto>由一个组成的泛型IEventDtoHandler<TEventDto>

我遇到的问题是 …

这是一个有点普遍的问题，因为它不仅适用于我的场景（使用 Azure 服务总线），而且适用于发布/订阅事件的上下文中的任何事件总线。

问题是：是否更倾向于拥有一个不能在生产者之间共享主题的架构/拓扑？ 换句话说：每个事件生产者一个主题 VS 多个生产者共享一个主题？

我有一个明确的偏好：一个主题应该只由一个制作人拥有和访问，如果其他制作人。但我似乎是团队中唯一一个持这种观点的人，而其他人似乎“为简单起见”在不同事件制作者之间共享同一主题似乎没有任何问题，而且我无法在技术可行性方面真正争论。 .

我希望从更技术的角度找到有价值的答案和良好实践，因为我的推理是从更多业务/组织的角度出发，因为我来自 DDD 背景，而其他人则没有。

• 主题是一对多通信的输出框（我将其解释为一个事件发布者，多个订阅者）
• 一个主题可以处理不同类型的事件消息，只要它们以某种方式相关（当然这是非常相关的）
• 在 DDD 中，有一个有界上下文的概念，我喜欢将微服务/模块视为实现这些有界上下文的一种方式。因此，即使其他一些服务“认为”他们想要发布相关内容并想要访问共享主题进行发布，我认为它们属于不同的有界上下文，并且应该有自己的主题。
• 如果多个生产者确实属于同一个有界上下文，那么我认为只有一个服务（或基础模块）应该负责发布在有界上下文中发生的事件。
• 一个生产者也有可能想要消费来自其他生产者的事件（它也是一个下标者）。生产者订阅相同的主题并且必须根据消息是由自己还是其他人生产来区分消息是没有意义的。

如您所见，从 DDD 的角度来看，需要在同一主题中发布的多个生产者会引发设计气味。我并不是说它不能完成，我试图从技术角度找出是否也应该避免它。

任何有这方面实践经验的人？

PS：有一个关于 Kafka 的类似问题，但我认为这与 Kafka 对发布者 - 订阅者使用不同的技术方法完全相同

更新 1：我不知道 NServiceBus，但我已经在 MassTransit 上工作了一些，当利用 MassTransit 的拓扑创建（这是唯一的方法 afaik）时，它不仅为每个生产者而且每个消息类型创建了不同的主题。

我计划构建一个相当大的应用程序（在并发用户/请求数量方面很大，而不是在功能方面）。

基本上，我将在某处提供服务，即等待命令执行它们，并在稍后确认完成。在发出确认消息之前，此服务将使用服务总线进行通信，从而使执行成为最终结果。

此服务的使用者可以是任何类型的应用程序（WPF、SL、...），但我的主要（也是第一个）客户端将是一个 asp.net MVC 应用程序 + WebApi (.Net 4.5) 或 MVC only (.Net 4.0) ) 与 ajax 控制器操作。

Web 应用程序将依靠 Ajax 调用来保持用户友好的响应式应用程序。

我对这种成熟的异步架构还很陌生，我有一些问题可以避免将来的头痛：

• 我的 web api 调用可能需要一些时间。我应该如何正确设计 api 以支持长时间运行的操作（某种异步？）。我已经阅读了新的 async 关键字，但为了知识起见，我想了解背后的内容。
• 我对服务的调用将包括发布一条消息并等待确认消息。如果我将它包装在一个方法中，我应该如何编写这个方法？我应该“阻止”直到收到确认（我想我不应该）？我应该返回一个 Task 对象并让消费者决定吗？
• 我也想知道 SignalR 是否可以帮助我。使用signalR，我想我可以使用真正的“即发即弃”命令发出，并路由到客户端以确认消息。
• 我是不是完全跑题了，我应该采取另一种方法吗？

在实现细节/框架方面，我想我会使用：

• Rabbitmq 作为消息系统
• Masstransit 抽象消息系统
• asp.MVC 4 构建 UI
• Webapi 隔离从 UI 控制器发出的命令，并允许其他类型的客户端发出命令

我正在尝试实现一个可消耗的库,它在CQRS的上下文中为每个域读取/写入应用程序服务.命令总线(或Dispatcher,或者在这种情况下可以调用的任何东西)接口可能会或可能不会被公开,但是实现应该从消费者中抽象出来,以鼓励对接口定义的合同进行编程.我不想要求库的使用者必须在DI框架中使用标准约定来设置库,因此使用的DI框架无关紧要(要求基于约定的DI超出了此问题的范围) .

internal interface ICommandMessage
{
    Guid Id { get; }
    DateTime DateRequestedUtc { get; }
}
internal class BaseCommandMessage
{
    /*... Implementation of ICommandMessage for common command data ...*/
}

internal class ExampleCommand : BaseCommandMessage
{
    /*... Additional data required for command ...*/
}

internal class AnotherExampleCommand : BaseCommandMessage
{
    /*... Additional data required for command ...*/
}

internal interface ICommandHandler<in TCommand> where TCommand : class, ICommandMessage
{
    Task HandleAsync(TCommand command);
}

internal class ExampleCommandHandler : ICommandHandler<ExampleCommand>, ICommandHandler<AnotherExampleCommand>
{ …

我想知道我对消息总线和消息队列工作原理的理解是否正确。

首先，我需要明确命名，服务总线与消息总线可以互换使用吗？这是一种发布者-订阅者类型的系统，其中消息被添加到任意数量的发布者的消息集合中，并且任意数量的订阅者都可以从中读取消息，到目前为止我是对的吗？

P1 ---                                         /``````S1
       \________ Service Bus Middleware ------+------ S2
       /           MESSAGE-COLLECTION          \______S3
P2 ---

我不明白的是

1. 订阅者如何知道它感兴趣的消息，我的意思是它显然订阅了它，但是它如何知道它应该订阅哪条消息？，它在哪里看到消息列表，它如何使用它？通过 API 或如何？

2. 订阅者如何接收消息？

3. 何时从 MESSAGE-COLLECTION 中删除消息？我可以想象的是，为每条消息保留一些计数器，该计数器代表订阅者的总数，一旦一个订阅者成功处理该消息，该计数器就会递减。

消息队列也称为消息代理，是一种推拉类型的系统。有任意数量的生产者和任意数量的消费者。每个生产者为每个消费者创建一个队列，并向其提供消息。

             --- Message Queue 1 ---- C1
           /  
P1 ------ +
           \ 
             --- Message Queue 2 ---- C2


P2 ------ + --- Message Queue 1 ---- C1

在这种情况下，一旦消费者成功处理该消息，该消息就会被删除。我对消息队列工作原理的理解是否正确？

我不确定到底是什么的另一个概念是event hub。

在我们的应用程序中,发布者创建一条消息并将其发送到主题.

然后,当所有主题的订阅者都收到消息时,它需要等待.

它没有出现,消息总线实现可以自动执行此操作.因此,我们倾向于让每个订阅者在完成后为客户端发送自己的新消息.

现在,客户端可以接收所有这些消息,并且当它从每个目的地获得一个消息时,做它必须做的任何清理.但是如果客户端(发送者)在确认流中部分崩溃会怎样？为了应对这样的不幸,我需要(重新)实现客户端上已经实现的总线 - 保存传入的确认,直到我得到足够的数量.

我不相信,我们的需求是深奥的 - 您将如何处理发件人(发布者)必须等待来自多个收件人(订阅者)的确认的情况？有点像从每个订户请求(和等待)回收收据到邮件列表...

如果重要的话,我们正在使用RabbitMQ.谢谢!