问题描述
我正在进行的一个项目需要从 SQS 读取消息，因此我决定使用 Akka 来分发这些消息的处理.
A project I'm working on requires the reading of messages from SQS, and I decided to use Akka to distribute the processing of these messages.
由于 Camel 支持 SQS，并且在 Consumer 类中内置了供 Akka 使用的功能，我认为最好以这种方式实现端点并读取消息，尽管我没有看到很多人的例子这样做.
Since SQS is support by Camel, and there is functionality built in for use in Akka in the Consumer class, I imagined it would be best to implement the endpoint and read messages this way, though I had not seen many examples of people doing so.
我的问题是我无法足够快地轮询队列以保持队列为空或接近空.我最初的想法是，我可以让消费者以 X/s 的速率通过 Camel 从 SQS 接收消息.从那里，我可以简单地创建更多的消费者来达到我需要处理消息的速度.
My problem is that I cannot poll my queue quickly enough to keep my queue empty, or near empty. What I originally thought was that I could get a Consumer to receive messages over Camel from SQS at a rate of X/s. From there, I could simply create more Consumers to get up to the rate at which I needed messages processed.
我的消费者:
import akka.camel.{CamelMessage, Consumer}
import akka.actor.{ActorRef, ActorPath}
class MyConsumer() extends Consumer {
def endpointUri = "aws-sqs://my_queue?delay=1&maxMessagesPerPoll=10&accessKey=myKey&secretKey=RAW(mySecret)"
var count = 0
def receive = {
case msg: CamelMessage => {
count += 1
}
case _ => {
println("Got something else")
}
}
override def postStop(){
println("Count for actor: " + count)
}
}
如图所示，我设置了 delay=1 以及 &maxMessagesPerPoll=10 来提高消息的速率，但我无法生成多个具有相同端点的消费者.
As shown, I've set delay=1 as well as &maxMessagesPerPoll=10 to improve the rate of messages, but I'm unable to spawn multiple consumers with the same endpoint.
我在文档中读到 默认情况下假定端点不支持多个消费者. 我相信这也适用于 SQS 端点，因为产生多个消费者只会给我一个消费者运行系统一分钟后，输出消息是 Count for actor: x 而不是其他输出 Count for actor: 0.
I read in the docs that By default endpoints are assumed not to support multiple consumers. and I believe this holds true for SQS endpoints as well, as spawning multiple consumers will give me only one consumer where after running the system for a minute, the output message is Count for actor: x instead of the others which output Count for actor: 0.
如果这有用的话；在单个消费者上使用此当前实现，我每秒可以读取大约 33 条消息.
If this is at all useful; I'm able to read approximately 33 messages/second with this current implementation on the single consumer.
这是从 Akka 的 SQS 队列中读取消息的正确方法吗?如果是这样，我有没有办法让它向外扩展，以便我可以将消息消耗率提高到接近 900 条消息/秒?
Is this the proper way to be reading messages from an SQS queue in Akka? If so, is there way I can get this to scale outward so that I can increase my rate of message consumption closer to that of 900 messages/second?
推荐答案
遗憾的是 Camel 目前不支持在 SQS 上并行消费消息.
Sadly Camel does not currently support parallel consumption of messages on SQS.
http://camel.465427.n5.nabble.com/Amazon-SQS-listener-as-multi-threaded-td5741541.html
为了解决这个问题，我编写了自己的 Actor 来使用 aws-java-sdk 轮询批处理消息 SQS.
To address this I've written my own Actor to poll batch messages SQS using the aws-java-sdk.
def receive = {
case BeginPolling => {
// re-queue sending asynchronously
self ! BeginPolling
// traverse the response
val deleteMessageList = new ArrayList[DeleteMessageBatchRequestEntry]
val messages = sqs.receiveMessage(receiveMessageRequest).getMessages
messages.toList.foreach {
node => {
deleteMessageList.add(new DeleteMessageBatchRequestEntry(node.getMessageId, node.getReceiptHandle))
//log.info("Node body: {}", node.getBody)
filterSupervisor ! node.getBody
}
}
if(deleteEntryList.size() > 0){
val deleteMessageBatchRequest = new DeleteMessageBatchRequest(queueName, deleteMessageList)
sqs.deleteMessageBatch(deleteMessageBatchRequest)
}
}
case _ => {
log.warning("Unknown message")
}
}
虽然我不确定这是否是最好的实现，当然可以对其进行改进，以使请求不会经常遇到空队列，但它确实适合我目前能够从同一队列轮询消息的需求排队.
Though I'm not certain if this is the best implementation, and it could of course be improved upon so that requests are not constantly hitting an empty queue, it does suit my current needs of being able to poll messages from the same queue.
使用这个从 SQS 获取大约 133(消息/秒)/actor.
Getting about 133 (messages/second)/actor from SQS with this.
这篇关于Akka、SQS 和 Camel 的消费者投票率的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，WP2