基于Java的消息分发系统设计

Question

我正在为以下用例寻找一个轻量级和高效的解决方案：

• 网关模块接收资源，以交付给不同的接收方。
• 每个接受者的资源(按到达的顺序)排队。
• 清除进程扫描这些队列，如果资源对某个接收方可用，那么他将它们捆绑在某个标记(唯一的id)下，并发送一个新包可用的通知。

系统特点：

• 受主的数量是动态的。
• 对一个包中的资源数量没有限制。

​

​

该模块将在Java 7下的Tomcat 7中使用(而不是集群)。

我考虑了以下解决办法：

1. JMS -针对每个接收者的动态队列配置，是否可以使用队列中的所有可用消息？每个队列的线程配置(不可伸缩)？
2. AKKA演员。没有找到合适的使用模式。
3. 朴素纯Java实现了，其中队列将被一个线程(循环)扫描。

我认为这是讨论解决这一问题的有效办法的正确场所。在考虑以下几点时，请分享你的想法：

• 适当的第三方框架。
• 资源队列可伸缩扫描。

提前谢谢。

Answer 1

你可以使用各种技术，例如：

• JMS动态队列
• 扩展了LMAX干扰子(例如。https://github.com/hicolour/disruptor-ext)

​

但是从高availability和可伸缩性的原因来看，您应该使用Akka

Akka

您实现的起点将是内置于Akka中的一致散列路由算法--简单地说，这种路由逻辑基于提供的密钥选择一致的路由。路由与您的问题描述相比，是受体。

路由器参与者有两种不同的风格，这为您提供了在基础结构中部署新受体的灵活机制。

• 池-路由器创建路由器作为子角色，并在路由器终止时将其从路由器中移除。
• 组-路由器在外部创建路由器，并且路由器使用参与者选择将消息发送到指定的路径，而不监视终止。

首先，请阅读Akka路由文档，以便更好地理解Akka框架中的路由实现：

• http://doc.akka.io/docs/akka/2.3.7/java/routing.html

您还可以查看这篇关于可伸缩和高可用系统设计的文章：

• http://prochera.com/blog/2014/07/15/building-a-scalable-and-highly-available-reactive-applications-with-akka-load-balancing-revisited/

Q1能让演员知道他的路线(他的哈希键)吗？

参与者可能知道当前处理什么密钥，因为它可能只是消息的一部分-但是您不应该基于这个键构建跨消息逻辑/状态。

消息：

import akka.routing.ConsistentHashingRouter.ConsistentHashable
  class Message(key : String) extends ConsistentHashable with Serializable {
      override def consistentHashKey(): AnyRef = key
  }

演员：

  import akka.actor.{Actor, ActorLogging}

  class EchoActor extends Actor with ActorLogging {

    log.info("Actor created {}", self.path.name)

    def receive = {
      case message: Message =>
        log.info("Received message {} in actor {}", message.consistentHashKey(),             self.path.name)
      case _ => log.error("Received unsupported message");
    }
  }

Q2可以管理除他的邮箱？以外的状态。

参与者状态只能通过它们之间发送的消息进行更改。

如果要初始化包含经典java/spring/..它将能够与非演员的世界/状态互动。dao层，但是这种类型的集成应该尽可能地受到限制，并作为反模式来处理。

Q3是否有一种使用防碰撞配置的方法？

作为API使用者，您需要自己定义collision resistant模型，但是Akka再次给出了所需的基础设施。

1. 在大多数情况下，键将是域的一部分，例如。拍卖id，客户id
2. 如果需要按需生成密钥，则可以使用带有ClusterSingleton扩展的持久化。

Generator可能是负责生成唯一ID的Actor，其他参与者可以使用ask模式获得新的id。

ClusterSingleton使用ClusterSingletonManager初始化，并使用ClusterSingletonProxy获得。

system.actorOf(ClusterSingletonManager.props(
singletonProps = Props(classOf[Generator]),
singletonName = "gnerator",
terminationMessage = End,
role = Some("generator")),
name = "singleton")


system.actorOf(ClusterSingletonProxy.props(
singletonPath = "/user/singleton/generator",
role = Some("generator")),
name = "generatorProxy")

Answer 2

我认为，对于您的问题，JMS将是正确的解决方案。您可以使用RabbitMQ，它有路由器，根据密钥将消息路由到不同的队列，并为消息流和消息确认机制提供内置的解决方案。

Answer 3

你可以用阿帕奇骆驼做这个。它的轻量级和支持大量的企业集成模式。特别是基于内容的路由器是一种可能的解决方案。