ElasticMQ 0.7.0: Long Polling, Non-Blocking Implementation Using Akka and Spray

本文主要是介绍ElasticMQ 0.7.0: Long Polling, Non-Blocking Implementation Using Akka and Spray,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

ElasticMQ 0.7.0: Long Polling, Non-Blocking Implementation Using Akka and Spray

原文作者:Adam Warski
原文地址:https://dzone.com/articles/elasticmq-070-long-polling-non
译者微博:@从流域到海域
译者博客:blog.csdn.net/solo95
本文同样刊载于腾讯云+:https://cloud.tencent.com/developer/article/1018703

ElasticMQ 0.7.0:长轮询,使用Akka和Spray的非阻塞实现

ElasticMQ 0.7.0,一个附带基于actor的Scala以及兼容Amazon SQS插件的消息队列系统刚刚发布。

这是一次重大的重写(即版本升级),升级之后将在核心使用Akka actors 并在REST层使用Spray。到目前为止,只有核心和SQS模块被重写, 日志( journaling),SQL后端和副本(replication)模块的重写尚未完成。

主要的客户端改进是:

  • 支持长轮询,这是SQS前一段时间的补充
  • 更简单的独立服务器 - 只需下载一个jar包

使用长时间的轮询的过程中,当收到消息时,可以指定一个额外的的MessageWaitTime属性。如果队列中没有消息,,ElasticMQ将等待MessageWaitTime几秒钟直到消息到达,而不是用空响应完成请求。这有助于减少带宽的使用(不需要非常频繁地进行请求),进而提高系统整体性能(发送后立即收到消息)并降低SQS成本。

独立的服务器现在是一个单一的jar包。要运行本地内存SQS实现(例如,测试使用SQS的应用程序),只需要下载jar文件并运行:

java -jar elasticmq-server-0.7.0.jar

这将在http://localhost:9324启动服务器。当然,接口和端口都是可配置的,详情请参阅自述文件。像以前一样,您也可以使用任何基于JVM的语言来运行嵌入式服务器。

实现说明

出于好奇,下面是对ElasticMQ如何实现的简短描述,包括核心系统,REST层,Akka数据流使用和长轮询实现。所有的代码都可以在GitHub上找到。

如前所述,ElasticMQ现在使用Akka和Spray来实现,并且不包含任何阻塞调用。一切都是异步的。

核心

核心系统是基于角色的。有一个主角色(main actor)(QueueManagerActor),它知道系统中当前创建了哪些队列,并提供了创建和删除队列的可能性。

为了与actor沟通,使用了类型化问答模式。例如,要查找一个队列(一个队列也是一个actor),就会定义一个消息:

case class LookupQueue(queueName:String)extends Replyable [Option [ActorRef]]

用法如下所示:

import org.elasticmq.actor.reply._
val lookupFuture: Future[Option[ActorRef]] = queueManagerActor ? LookupQueue("q2")

如前所述,每个队列都是一个actor,并且已经封装了队列状态。我们可以使用简单的可变数据结构,而不需要任何线程同步,因为角色模型(actor model)为我们处理了这个问题。有一些消息可以发送给queue-actor,例如:

case class SendMessage(message: NewMessageData)   extends Replyable[MessageData]
case class ReceiveMessages(visibilityTimeout: VisibilityTimeout, count: Int, waitForMessages: Option[Duration])     extends Replyable[List[MessageData]]
case class GetQueueStatistics(deliveryTime: Long) extends Replyable[QueueStatistics]
Rest层

SQS查询/ REST层是使用Spray来实现的,这是一个基于Akka的轻量级REST/HTTP工具包。

除了基于角色的非阻塞IO实现外,Spray还提供了强大的路由库spray-routing。它包含一些内置的指令,用于在请求方法(get / post等)上进行匹配,提取表单参数中的查询参数或匹配请求路径。但它也可以让你使用简单的指令组合来定义你自己的指令。一个典型的ElasticMQ route示例如下所示:

val listQueuesDirective = action("ListQueues") {rootPath {anyParam("QueueNamePrefix"?) { prefixOption =>// logic}}}

action"Action"URL的body参数中匹配指定的action名称并接受/拒绝请求的地方,rootPath会匹配出空路径(…)。Spray有一个很好的教程,如果你有兴趣,我建议你看看这篇教程。

如何使用路由中的队列角色(queue actors)来完成HTTP请求?

关于Spray的RequestContext好处是,它所做的只是将一个实例传递给你的路由,不需要任何回复。完全放弃请求或使用某个value完成该请求仅仅取决于它的路由。该请求也可以在另一个线程中完成 - 或者,例如,在未来某个线程运行完成时。这正是ElasticMQ所做的。在这里使用mapflatMapfor-comprehensions(这是一个针对map/ flatMap更好的语法)是非常方便的,例如(省略了一些内容):

// Looking up the queue and deleting it are going to be called in sequence,
// but asynchronously, as ? returns a Future
for {queueActor <- queueManagerActor ? LookupQueue(queueName)_ <- queueActor ? DeleteMessage(DeliveryReceipt(receipt))
} {requestContext.complete(200, "message deleted")
}

有时,当流程更复杂时,ElasticMQ会使用Akka Dataflow,这需要启用continuations插件。还有一个类似的项目,使用宏,Scala Async,但这个仍处于早期开发阶段。

使用Akka Dataflow,您可以编写使用Future们的代码,就好像编写正常的序列化代码一样。CPS插件会将其转换为在需要时使用回调。这是一个来自CreateQueueDirectives的例子:

(序列化代码sequential code,也有翻译成顺序代码的,即按顺序执行的代码,过程中不存在多线程异步操作,译者注)

flow {val queueActorOption = (queueManagerActor ? LookupQueue(newQueueData.name)).apply()queueActorOption match {case None => {val createResult = (queueManagerActor ? CreateQueue(newQueueData)).apply()createResult match {case Left(e) => throw new SQSException("Queue already created: " + e.message)case Right(_) => newQueueData}}case Some(queueActor) => {(queueActor ? GetQueueData()).apply()}}
}

这里的重要部分是flow代码块,它界定了转换的范围,以及调用Future提取future内容的apply()。这看起来像完全正常的序列化代码,但是在执行时,因为第一次Future是第一次使用将会异步运行。

长轮询

由于所有的代码都是异步和非阻塞的,实现长轮询非常容易。请注意,从一个队列接收消息时,我们得到一个Future[List[MessageData]]。为了发出响应已完成这个future,HTTP请求也将会以适当的响应来完成。然而,这个future几乎可以立即完成(例如正常情况下),比如在10秒之后 - 代码所需的支持没有变化。唯一要做的就是延迟完成future,直到指定的时间过去或新的消息到达。

实现在QueueActorWaitForMessagesOps中。当接收到消息的请求到达时,队列中没有任何内容产生,而是立即回复(即向发送者actor发送空列表),我们将储存原始请求的引用和发送方actor在map中。使用Akka调度程序,我们还计划在指定的时间超过之后发回空列表并删除条目。

当新消息到达时,我们只需从map上等待一个请求,然后尝试去完成它。同样,所有同步和并发问题都由Akka和actor模型来处理。

请测试新版本,如果您有任何反馈,请让我们知晓!

Adam

这篇关于ElasticMQ 0.7.0: Long Polling, Non-Blocking Implementation Using Akka and Spray的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/623336

相关文章

long long,_int64使用小结

前言:   在16位环境下,int/unsigned int 占16位,long/unsigned long占32位   在32位环境下,int占32位,unsigned int占16位,long/unsigned long占32位 何时需要使用:   long 和 int 范围是[-2^31,2^31),即-2147483648~2147483647,而unsigned范围是[0,2^32),

Apache Flink:Keyed Window与Non-Keyed Window

Apache Flink中,Window操作在流式数据处理中是非常核心的一种抽象,它把一个无限流数据集分割成一个个有界的Window(或称为Bucket),然后就可以非常方便地定义作用于Window之上的各种计算操作。本文我们主要基于Apache Flink 1.4.0版本,说明Keyed Window与Non-Keyed Window的基本概念,然后分别对与其相关的WindowFunction

《长得太长也是错?——后端 Long 型 ID 精度丢失的“奇妙”修复之旅》

引言 在前后端分离的时代,我们的生活充满了无数的机遇与挑战——包括那些突然冒出来的让人抓狂的 Bug。今天我们要聊的,就是一个让无数开发者哭笑不得的经典问题:后端 Long 类型 ID 过长导致前端精度丢失。说到这个问题,那可真是“万恶之源”啊,谁让 JavaScript 只能安全地处理 Number.MAX_SAFE_INTEGER(也就是 9007199254740991)以内的数值呢?

踩坑记录(Long[]ids)

主要针对Long[] ids 的判空问题 问题代码 public void delYnjC(Long[] ids) {if (CollectionUtils.isEmpty(Collections.singleton(ids))) {throw new NullPointerException("参数不能为空");}naturalYnjCMapper.delYnjC(ids);} 修正

class _ContiguousArrayStorage deallocated with non-zero retain count

Xcode报错 : Object 0x11c614000 of class _ContiguousArrayStorage deallocated with non-zero retain count 2. This object's deinit, or something called from it, may have created a strong reference to self w

修改后无警告全面支持non-ARC以及ARC的OpenUDID

OpenUDID Open source initiative for a universal and persistent UDID solution for iOS. 首创的给iOS提供设备唯一标示符的一个开源代码. https://github.com/ylechelle/OpenUDID 使用很简单,引入头文件后执行下面一句话即可: 由于原版OpenUDI

CodeForces 407B Long Path

题意: 有n+1个格子  起点在1  每个格子有个前进1的门  前n个格子有个返回的门(返回前面某个格子)  奇数次走进一个格子就去走返回门  偶数次走前进门  问  走到n+1要走过几道门 思路: 一看就是DP to[i]表示第i个格子的返回门 go[i]表示离开第i个格子需要的步数 sum[i]表示离开前i个格子需要的步数 明显  go[i]=sum[i-1]-sum[to

Command line is too long. Shorten command line for DisplayApplication (1) or

微服务项目启动类起不来,如下 解决办法:IEDA开发环境下 找到你的项目下面的.idea\workspace.xml 添加一个property : <property name="dynamic.classpath" value="true" /> 帮同事看的问题,自己测试没问题就关闭了。图片借用网上的。 参考:参考

Build Min Heap Using Array

Build min-heap using Array. 思路1:首先明白heap的底层implementation就是array,从0开始的parent和left,right的关系为, 如果现在的node index为i,那么parent index就是 (i-1)/2;   left  为2*i+1, right为 2*i+2;          ( i-1 ) / 2