Ein Projekt, an dem ich arbeite, erfordert das Lesen von Nachrichten von SQS, und ich habe beschlossen, Akka zu verwenden, um die Verarbeitung dieser Nachrichten zu verteilen.

Da SQS von Camel unterstützt wird und Funktionen für die Verwendung in Akka in der Consumer-Klasse integriert sind, stellte ich mir vor, dass es am besten ist, den Endpunkt zu implementieren und Nachrichten auf diese Weise zu lesen, obwohl ich nicht viele Beispiele von Personen gesehen hatte, die dies taten.

Mein Problem ist, dass ich meine Warteschlange nicht schnell genug abfragen kann, um sie leer oder fast leer zu halten. Was ich ursprünglich dachte, war, dass ich einen Verbraucher dazu bringen könnte, Nachrichten über Camel von SQS mit einer Rate von X / s zu empfangen. Von dort aus konnte ich einfach mehr Konsumenten erstellen, um die Rate zu erreichen, mit der die Nachrichten verarbeitet werden mussten.

Mein Verbraucher:

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

Wie gezeigt habe ich eingestelltdelay=1 ebenso gut wie&maxMessagesPerPoll=10 um die Rate der Nachrichten zu verbessern, aber ich bin nicht in der Lage, mehrere Verbraucher mit demselben Endpunkt zu erzeugen.

Ich habe in den Dokumenten gelesen, dassBy default endpoints are assumed not to support multiple consumers. und ich glaube, dass dies auch für SQS-Endpunkte gilt, da ich durch das Laichen mehrerer Konsumenten nur einen Konsumenten habe, bei dem nach einer Minute Laufzeit des Systems die Ausgabemeldung angezeigt wirdCount for actor: x anstelle der anderen die ausgebenCount for actor: 0.

Wenn das überhaupt nützlich ist; Mit dieser aktuellen Implementierung kann ich ungefähr 33 Nachrichten pro Sekunde auf dem einzelnen Konsumenten lesen.

Ist dies die richtige Methode, um Nachrichten aus einer SQS-Warteschlange in Akka zu lesen? Wenn ja, gibt es eine Möglichkeit, dies nach außen zu skalieren, damit ich meine Nachrichtenverbrauchsrate näher an die von 900 Nachrichten / Sekunde heranführen kann?