Akka、SQS和Camel的消费者投票率

Question

我的问题是，我无法足够快地轮询我的队列，以保持我的队列为空或接近空。我最初的想法是，我可以让使用者以x/s的速率通过Camel从SQS接收消息。从那里，我可以简单地创建更多的消费者，以达到我需要的消息处理速度。

我的消费者：

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以提高消息的速率，但是我无法生成具有相同endpoint的多个使用者。

我在文档中读到，默认情况下假定endpoint不支持多个使用者。我相信SQSendpoint也是如此，因为生成多个使用者将只给我一个使用者，在运行系统一分钟后，输出消息是count for actor:x，而不是其他输出count for actor:0的消息。

如果这是有用的；使用当前的实现，我可以在单个消费者上读取大约33条消息/秒。

这是从Akka中的SQS队列读取消息的正确方式吗？如果是这样，有没有办法让它向外扩展，以便我可以将我的消息消耗率提高到接近900条消息/秒？

Answer 1

遗憾的是，Camel目前不支持在SQS上并行使用消息。

http://camel.465427.n5.nabble.com/amazon-sqs-listener-as-multi-threaded-td5741541.html

为了解决这个问题，我编写了自己的Actor来使用AWS-Java-SDK轮询批处理消息SQS。

  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")
    }
  }