RabbitMQ mit dem Beispiel: Mehrere Threads, Kanäle und Warteschlangen

Ich denke, Sie haben einige Probleme mit anfänglichen Verständnis. Ehrlich gesagt, ich bin ein bisschen überrascht zu sehen, die folgenden: both need 5 threads to handle the volume. Wie hast du dich identifizieren müssen, die genaue Zahl? Haben Sie eine garantiert 5 threads reichen wird???

RabbitMQ abgestimmt ist und mal getestet, also es ist alles über die richtige Gestaltung
und der effizienten Verarbeitung.

Lassen Sie uns versuchen, um zu überprüfen das problem und finden eine passende Lösung. BTW, message queue selbst wird nicht garantiert, Sie haben wirklich gute Lösung. Sie haben, um zu verstehen, was Sie tun und auch tun, einige weitere Tests.

Als Sie definitiv wissen, es gibt viele layouts möglich:

Verwende ich layout B als der einfachste Weg, um zu zeigen 1 Produzent N Verbraucher-problem. Da du so besorgt bist, den Durchsatz. BTW, wie man erwarten könnte RabbitMQ verhält sich ganz gut (Quelle). Achten Sie auf prefetchCountich kläre es später:

So ist es wahrscheinlich, message processing logic ist eine richtige Ort, um sicherzustellen, Sie haben genug Durchsatz. Natürlich können Sie überspannen einen neuen thread jedes mal, wenn Sie brauchen, um eine Nachricht zu verarbeiten, aber irgendwann ist ein solcher Ansatz wird töten Ihrem system. Im Grunde, mehr threads, haben Sie größere Latenz erhalten Sie (Sie können überprüfen, Amdahl ' s law wenn Sie wollen).

(siehe Amdahl ' s law illustriert)

Tipp #1: Seien Sie vorsichtig mit threads, ThreadPools verwenden (details)

Einen thread-pool kann beschrieben werden als eine Sammlung von Runnable-Objekten
(work queue) und verbindungen der ausgeführten threads. Diese threads sind
ständig in Betrieb und überprüfen die Arbeit, die Abfrage für die neue Arbeit. Wenn
es gibt neue Arbeit, die Sie ausführen dieser Lauffähig. Der Thread
die Klasse bietet eine Methode, z.B. execute(Runnable r) zum hinzufügen einer neuen
Runnable-Objekt an die Warteschlange.

public class Main {
  private static final int NTHREDS = 10;

  public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(NTHREDS);
    for (int i = 0; i < 500; i++) {
      Runnable worker = new MyRunnable(10000000L + i);
      executor.execute(worker);
    }
    //This will make the executor accept no new threads
    //and finish all existing threads in the queue
    executor.shutdown();
    //Wait until all threads are finish
    executor.awaitTermination();
    System.out.println("Finished all threads");
  }
} 

Tipp #2: Seien Sie vorsichtig mit message-processing-overhead

Ich würde sagen, das ist offensichtlich, Optimierung der Technik. Es ist wahrscheinlich, dass Sie ' ll senden Sie klein und leicht sind, Nachrichten zu verarbeiten. Der gesamte Ansatz wird über kleinere Nachrichten werden kontinuierlich eingestellt und verarbeitet. Große Nachrichten, die schließlich einen bösen Streich spielen, so ist es besser, dies zu vermeiden.

So ist es besser, senden Sie kleine Stücke von Informationen, aber was über die Verarbeitung? Es ist ein Aufwand jedes mal, wenn Sie submit a job. Batch-Verarbeitung kann sehr hilfreich sein im Falle von high-eingehende Nachricht-rate.

Zum Beispiel, sagen wir, wir haben einfach message processing logic und wir nicht wollen, um thread-spezifische Gemeinkosten jeder Zeit die Nachricht verarbeitet wird. Um diese zu optimieren, dass sehr einfache CompositeRunnable can be introduced:

class CompositeRunnable implements Runnable {

    protected Queue<Runnable> queue = new LinkedList<>();

    public void add(Runnable a) {
        queue.add(a);
    }

    @Override
    public void run() {
        for(Runnable r: queue) {
            r.run();
        }
    }
}

Oder tun Sie das gleiche in einer etwas anderen Weise, durch das sammeln von Nachrichten verarbeitet werden:

class CompositeMessageWorker<T> implements Runnable {

    protected Queue<T> queue = new LinkedList<>();

    public void add(T message) {
        queue.add(message);
    }

    @Override
    public void run() {
        for(T message: queue) {
            //process a message
        }
    }
}

In einer Weise, die Sie Nachrichten verarbeiten kann, die mehr effektiv.

Tipp #3: Optimieren der Nachrichtenverarbeitung

Trotz der Tatsache, die Sie wissen, kann die Verarbeitung von Nachrichten in parallel (Tip #1) und reduzieren processing-overhead (Tip #2) Sie haben alles zu tun schnell. Redundante Bearbeitungsschritte, schwere Schleifen und so weiter, die die Leistung beeinträchtigen können, eine Menge. Finden Sie interessante case-study:

Die Verbesserung der Message-Queue-Durchsatz um das Zehnfache durch die Wahl der richtigen XML-Parser

Tipp #4: Anschluss-und Channel-Management

• Starten einen neuen Kanal auf eine bestehende Verbindung besteht aus einem Netzwerk
  round-trip - starten Sie eine neue Verbindung dauert mehrere.
• Jede Verbindung wird ein file-descriptor auf dem server. Kanäle nicht.
• Publishing eine große Nachricht auf einen Kanal zu blockieren, eine Verbindung
  während es geht. Andere als die, die multiplexing-Verfahren ist relativ transparent.
• - Verbindungen, die Veröffentlichung kann blockiert werden, wenn der server
  überlastet - es ist eine gute Idee zu trennen, veröffentlichen und nutzen
  verbindungen
• Bereit sein zu behandeln Nachricht platzt

(Quelle)

Bitte beachten Sie, alle Tipps sind perfekt zusammenarbeiten. Fühlen Sie sich frei, zu lassen mich wissen, wenn Sie weitere Informationen benötigen.

Complete Verbraucher-Beispiel (Quelle)

Bitte beachten Sie Folgendes:

• Kanal.basicQos(prefetch) - Wie Sie bereits gesehen haben prefetchCount sehr hilfreich sein könnte:
  

  Dieser Befehl ermöglicht es dem Verbraucher, wählen Sie eine prefetch-Fenster
  gibt die Menge der unbestätigten Meldungen ist es bereit
  erhalten. Durch die Einstellung der prefetch-Anzahl auf einen Wert ungleich null, der broker
  wird nicht liefern, Mitteilungen an die Verbraucher, würde die Verletzung, die
  limit. Verschieben Sie das Fenster nach vorne, hat der Verbraucher zu erkennen
  der Empfang einer Nachricht (oder eine Gruppe von Nachrichten).

• ExecutorService threadExecutor - Sie können angeben, richtig konfiguriert executor service.

Beispiel:

static class Worker extends DefaultConsumer {

    String name;
    Channel channel;
    String queue;
    int processed;
    ExecutorService executorService;

    public Worker(int prefetch, ExecutorService threadExecutor,
                  , Channel c, String q) throws Exception {
        super(c);
        channel = c;
        queue = q;
        channel.basicQos(prefetch);
        channel.basicConsume(queue, false, this);
        executorService = threadExecutor;
    }

    @Override
    public void handleDelivery(String consumerTag,
                               Envelope envelope,
                               AMQP.BasicProperties properties,
                               byte[] body) throws IOException {
        Runnable task = new VariableLengthTask(this,
                                               envelope.getDeliveryTag(),
                                               channel);
        executorService.submit(task);
    }
}

Können Sie auch prüfen, die folgenden:

• Solution-Architektur Mit Warteschlangen?
• Einige queuing-Theorie: Durchsatz, Latenz und Bandbreite
• Eine quick message queue benchmark: ActiveMQ, RabbitMQ, HornetQ, QPID, Apollo...

InformationsquelleAutor der Antwort

Warum zur Umsetzung alles selbst?

Versuchen, irgendeine Art von integration framework. Sagen wir Camel bereits Bündel-Anschluss an verschiedene Systeme, Rabbit MQ enthalten Camel Rabbit MQ.

Müssen Sie nur definieren Sie Ihre Routen.
Zum Beispiel:

Wollen Sie konsumieren Nachrichten aus der Anmeldung Warteschlange von 5 gleichzeitige Verbraucher in eine Datei.

from("rabbitmq://localhost/Logging ?concurrentConsumers=5")
.to("file://yourLoggingFile")

Gibt es viele Optionenwie man set-Dateien, die Verbraucher. Wie Sie sehen können können Sie definieren, wie viele Verbraucher gerade laichen, indem Sie concurrentConsumers=5 in Ihre URI. Wenn Sie möchten, können Sie erstellen Sie Ihre eigenen Erzeuger oder Verbraucher durch die Implementierung von Prozessor-Schnittstelle.

Ist es sehr vielseitiges und mächtiges framework, die Sie tun können eine Menge Arbeit nur durch die Verwendung der bereitgestellten Komponenten. Das web-Projekt enthält eine Reihe von Beispielen und Dokumentation.

InformationsquelleAutor der Antwort Milan Baran