Ereignisgesteuerte Fitness-Architektur mit Node.js

Ever eine Fitness-App genutzt und sich gefragt, wie sie Ihren einstündigen Lauf sofort verarbeitet, Ihnen ein neues Abzeichen verleiht und Ihre Position in der wöchentlichen Bestenliste aktualisiert? Wenn das alles in der Zeit geschah, die der Spinner „Save Workout“ zum Verschwinden brauchte, haben Sie wahrscheinlich die Magie der asynchronen Verarbeitung erlebt.

In einer traditionellen monolithischen Architektur muss der Server, wenn ein Benutzer ein Training einreicht, mehrere Aufgaben nacheinander ausführen: die Daten validieren, sie in der Datenbank speichern, Erfolge berechnen, Benutzerstatistiken aktualisieren und Bestenlisten aktualisieren. Das kann zu langsamen API-Antwortzeiten und einer schlechten Benutzererfahrung führen, insbesondere unter hoher Last. Wenn ein Schritt fehlschlägt, muss möglicherweise der gesamte Prozess zurückgesetzt werden.

In diesem Tutorial lösen wir dieses Problem, indem wir ein einfaches, aber leistungsstarkes ereignisgesteuertes System mit Node.js und RabbitMQ aufbauen. Wir erstellen zwei Microservices:

1. workout-service: Eine Express-API, die Trainingsdaten von Benutzern entgegennimmt und sie sofort an eine Nachrichtenwarteschlange sendet. Ihre einzige Aufgabe ist es, schnell und zuverlässig Daten anzunehmen.
2. processing-service: Ein Hintergrund-Worker, der auf neue Trainingsnachrichten aus der Warteschlange wartet, die aufwendige Arbeit erledigt (etwa Statistiken berechnen und auf Erfolge prüfen) und die Ergebnisse protokolliert.

Dieser entkoppelte Ansatz macht unsere Anwendung skalierbarer, widerstandsfähiger und leichter wartbar. ✨

Voraussetzungen:

• Node.js und npm (oder yarn) sind installiert.
• Docker und Docker Compose (um RabbitMQ einfach auszuführen).
• Grundlegendes Verständnis von JavaScript (async/await) und Express.js.

Das Problem verstehen: Der monolithische Engpass

Stellen Sie sich vor, Ihre Fitness-App wird ein großer Erfolg. Tausende von Benutzern beenden gleichzeitig ihr Training. Ihr einzelner API-Endpunkt gerät nun ins Stocken:

• Hohe Latenz: Benutzer starren sekundenlang auf einen Ladespinner und warten, bis alle Berechnungen nach dem Training abgeschlossen sind.
• Geringe Zuverlässigkeit: Der Drittanbieterdienst, den Sie zur Berechnung des Höhengewinns nutzen, fällt aus. Nun kann niemand mehr Trainings speichern, weil ein Glied in der Kette defekt ist.
• Skalierungsprobleme: Die Annahme von Trainings ist leichtgewichtig, die Berechnung von Erfolgen jedoch CPU-intensiv. Sie müssen die gesamte Anwendung skalieren, nur um die Verarbeitungslast zu bewältigen – das ist ineffizient und kostspielig.

Eine ereignisgesteuerte Architektur mit einer Nachrichtenwarteschlange löst diese Probleme. Der workout-service nimmt das Training einfach an und veröffentlicht ein Ereignis (eine Nachricht) in einer RabbitMQ-Warteschlange. Das ist ein sehr schneller Vorgang. Der processing-service verarbeitet diese Ereignisse in seinem eigenen Tempo, und Sie können ihn unabhängig skalieren. Wenn der Verarbeitungsdienst ausfällt, bleibt die Nachricht in der Warteschlange und kann verarbeitet werden, sobald der Dienst wieder verfügbar ist.

Architektur der ereignisgesteuerten Trainingsverarbeitung

Das folgende Diagramm zeigt unseren asynchronen Ablauf der Trainingsverarbeitung:

Voraussetzungen: Unsere Umgebung einrichten

Zuerst richten wir unsere Projektstruktur und Abhängigkeiten ein.

Wir erstellen ein Monorepo mit zwei Paketen: workout-service und processing-service.

mkdir fitness-app-backend
cd fitness-app-backend
mkdir packages
cd packages
mkdir workout-service
mkdir processing-service

1. RabbitMQ mit Docker

Der einfachste Weg, RabbitMQ für die Entwicklung auszuführen, ist Docker. Erstellen Sie eine Datei docker-compose.yml im Stammverzeichnis Ihres Verzeichnisses fitness-app-backend.

# docker-compose.yml
version: '3.8'
services:
  rabbitmq:
    image: "rabbitmq:3.13-management"
    ports:
      - "5672:5672"  # AMQP port for our services
      - "15672:15672" # Management UI
    environment:
      - RABBITMQ_DEFAULT_USER=user
      - RABBITMQ_DEFAULT_PASS=password

Führen Sie es nun aus:

docker-compose up -d

Sie können die RabbitMQ-Management-Oberfläche nun unter http://localhost:15672 aufrufen und sich mit user/password anmelden.

2. Abhängigkeiten der Dienste

Für workout-service:

cd packages/workout-service
npm init -y
npm install express amqplib

Für processing-service:

cd ../processing-service
npm init -y
npm install amqplib

Wir verwenden die beliebte Bibliothek amqplib, um in Node.js mit RabbitMQ zu interagieren.

API zur Annahme von Trainings mit RabbitMQ-Producer erstellen

Der workout-service ist eine Express-API mit einem einzigen Zweck: Trainingsdaten empfangen und sie in unsere RabbitMQ-Warteschlange veröffentlichen.

Was wir tun

Wir erstellen einen Endpunkt /workout, der eine POST-Anfrage mit einer JSON-Nutzlast akzeptiert, die ein abgeschlossenes Training darstellt. Dieser Dienst verbindet sich dann mit RabbitMQ und sendet die Trainingsdaten an eine Warteschlange namens workout_jobs.

Implementierung

Erstellen Sie eine Datei namens index.js in packages/workout-service.

// packages/workout-service/index.js
const express = require('express');
const amqp = require('amqplib');

const app = express();
app.use(express.json());

const RABBITMQ_URL = 'amqp://user:password@localhost:5672';
const QUEUE_NAME = 'workout_jobs';
let channel; // RabbitMQ channel

// --- RabbitMQ Connection ---
async function connectRabbitMQ() {
  try {
    const connection = await amqp.connect(RABBITMQ_URL);
    channel = await connection.createChannel();
    await channel.assertQueue(QUEUE_NAME, { durable: true }); // Durable queue
    console.log('✅ Connected to RabbitMQ and workout_jobs queue is ready.');
  } catch (error) {
    console.error('❌ Failed to connect to RabbitMQ', error);
    process.exit(1); // Exit if connection fails
  }
}

// --- API Endpoint ---
app.post('/workout', (req, res) => {
  const workoutData = req.body;

  // Basic validation
  if (!workoutData || !workoutData.type || !workoutData.duration) {
    return res.status(400).send({ message: 'Invalid workout data' });
  }

  // Send to queue
  const message = Buffer.from(JSON.stringify(workoutData));
  channel.sendToQueue(QUEUE_NAME, message, { persistent: true }); // Persistent message

  console.log(`📥 Workout data sent to queue:`, workoutData);

  res.status(202).send({
    message: 'Workout received and is being processed.',
  });
});

// --- Start Server ---
const PORT = 3000;
app.listen(PORT, async () => {
  await connectRabbitMQ();
  console.log(`🚀 Workout service listening on port ${PORT}`);
});

So funktioniert es

1. RabbitMQ-Verbindung: Die Funktion connectRabbitMQ stellt eine Verbindung her und erstellt einen Channel. Anschließend rufen wir assertQueue mit der Option durable: true auf. Dadurch übersteht die Warteschlange einen Neustart von RabbitMQ.
2. Express-Endpunkt: Der Endpunkt /workout empfängt die Trainingsdaten.
3. Veröffentlichen der Nachricht: Wir wandeln die JSON-Trainingsdaten in einen Buffer um und verwenden channel.sendToQueue. Die Option { persistent: true } weist RabbitMQ an, die Nachricht auf der Festplatte zu speichern, damit sie bei einem Serverabsturz nicht verloren geht.
4. HTTP 202 Accepted: Wir geben den Statuscode 202 Accepted zurück. Das ist entscheidend – er teilt dem Client mit: „Ich habe Ihre Anfrage erhalten und werde sie verarbeiten“, ohne dass er auf den Abschluss der Verarbeitung warten muss.

Hintergrund-Worker für die Trainingsverarbeitung erstellen

Dieser Dienst ist unser Hintergrund-Worker. Er verbindet sich mit derselben RabbitMQ-Warteschlange, wartet auf Nachrichten und „verarbeitet“ sie.

Was wir tun

Der Worker holt Trainingsnachrichten einzeln aus der Warteschlange workout_jobs. Für jedes Training simuliert er die Verarbeitung, indem er einige Statistiken berechnet und auf einen einfachen Erfolg prüft.

Implementierung

Erstellen Sie eine Datei namens index.js in packages/processing-service.

// packages/processing-service/index.js
const amqp = require('amqplib');

const RABBITMQ_URL = 'amqp://user:password@localhost:5672';
const QUEUE_NAME = 'workout_jobs';

async function startWorker() {
  try {
    const connection = await amqp.connect(RABBITMQ_URL);
    const channel = await connection.createChannel();
    await channel.assertQueue(QUEUE_NAME, { durable: true });

    // This ensures the worker only gets one message at a time
    channel.prefetch(1);
    
    console.log(`[ P ] Waiting for workouts in ${QUEUE_NAME}. To exit press CTRL+C`);

    channel.consume(QUEUE_NAME, (msg) => {
      if (msg !== null) {
        const workoutData = JSON.parse(msg.content.toString());
        console.log(`[ → ] Received workout:`, workoutData);

        // Simulate processing
        processWorkout(workoutData);

        // Acknowledge the message
        channel.ack(msg);
        console.log('[ ✔ ] Done processing. Acknowledged message.');
      }
    }, { noAck: false }); // `noAck: false` requires us to manually acknowledge

  } catch (error) {
    console.error('❌ Failed to start worker', error);
  }
}

function processWorkout(data) {
    // 1. Calculate stats
    const caloriesBurned = data.duration * 8.5; // Simple heuristic
    console.log(`[ ⚙️ ] Processing: Calculated ${caloriesBurned.toFixed(2)} calories burned.`);

    // 2. Check for achievements
    if (data.type === 'Running' && data.distance >= 5) {
        console.log(`[ 🏆 ] Achievement Unlocked: "5K Runner"!`);
        // In a real app, you would save this to the database.
    }

    if (data.duration > 60) {
        console.log(`[ 🏆 ] Achievement Unlocked: "Hour of Power"!`);
    }

    // Simulate a delay for processing
    // In a real app, this could be database calls, API requests, etc.
    const processingTime = Math.random() * 2000 + 500; // 0.5 to 2.5 seconds
    const start = Date.now();
    while (Date.now() - start < processingTime) {
        // blocking for demo purposes
    }
}

startWorker();

So funktioniert es

1. Verbindung und Warteschlange: Der Worker verbindet sich mit derselben dauerhaften Warteschlange workout_jobs.
2. channel.prefetch(1): Dies ist eine zentrale Einstellung für Worker. Sie weist RabbitMQ an, keine neue Nachricht zu senden, bis der Worker die aktuelle verarbeitet und bestätigt hat. So wird der Worker nicht überlastet.
3. channel.consume: Damit beginnt das Abhören von Nachrichten. Die Callback-Funktion wird für jede Nachricht ausgeführt.
4. noAck: false: Wir setzen die Bestätigung ausdrücklich auf manuell. Das ist entscheidend für die Zuverlässigkeit. Wenn unser Worker mitten in der Verarbeitung abstürzt, erkennt RabbitMQ, dass die Nachricht nie bestätigt wurde, und stellt sie erneut in die Warteschlange, damit ein anderer (oder derselbe) Worker es erneut versuchen kann.
5. processWorkout(): Diese Funktion enthält unsere Geschäftslogik. Sie ist ein Platzhalter für komplexere Berechnungen, die Sie in einer echten Anwendung durchführen würden.
6. channel.ack(msg): Sobald processWorkout abgeschlossen ist, rufen wir ack(msg) auf, um RabbitMQ mitzuteilen: „Diese Nachricht wurde erfolgreich verarbeitet; Sie können sie sicher löschen.“

Alles zusammenfügen

Sehen wir uns nun unser System in Aktion an!

1. RabbitMQ starten:

   docker-compose up -d
   

2. Den Verarbeitungsdienst (Consumer) starten: Öffnen Sie ein Terminal, wechseln Sie zu packages/processing-service und führen Sie aus:

   node index.js
   

   Sie sollten Folgendes sehen: [ P ] Waiting for workouts in workout_jobs...

3. Den Trainingsdienst (Producer) starten: Öffnen Sie ein zweites Terminal, wechseln Sie zu packages/workout-service und führen Sie aus:

   node index.js
   

   Sie sollten Folgendes sehen: 🚀 Workout service listening on port 3000

4. Ein Training senden: Öffnen Sie ein drittes Terminal (oder verwenden Sie ein Tool wie Postman), um eine POST-Anfrage an unsere API zu senden.

   Hier ein Beispielbefehl curl für ein Training, das einen Erfolg freischaltet:

   curl -X POST http://localhost:3000/workout \
   -H "Content-Type: application/json" \
   -d '{
     "userId": "user-123",
     "type": "Running",
     "duration": 45,
     "distance": 5.2,
     "timestamp": "2025-12-10T10:00:00Z"
   }'
   

   Sie sollten sofort eine Antwort erhalten: {"message":"Workout received and is being processed."}

   Sehen Sie sich nun Ihre anderen Terminals an!

   • Ausgabe des Trainingsdienstes:

     📥 Workout data sent to queue: { userId: 'user-123', type: 'Running', ... }
     

   • Ausgabe des Verarbeitungsdienstes:

     [ → ] Received workout: { userId: 'user-123', type: 'Running', ... }
     [ ⚙️ ] Processing: Calculated 382.50 calories burned.
     [ 🏆 ] Achievement Unlocked: "5K Runner"!
     [ ✔ ] Done processing. Acknowledged message.
     

Erfolg! Sie haben ein Training erfolgreich asynchron verarbeitet.

Überlegungen zu Sicherheit und Leistung

• Eingabevalidierung: Unsere aktuelle Validierung ist minimal. Verwenden Sie in einer realen Anwendung eine Bibliothek wie Joi oder Zod, um eingehende Daten im workout-service gründlich zu validieren, bevor sie überhaupt die Warteschlange erreichen.
• Fehlerbehandlung: Was, wenn processWorkout fehlschlägt? Sie sollten es in einen Block try...catch einschließen. Tritt ein Fehler auf, können Sie die Nachricht nack (negativ bestätigen), wodurch sie entweder verworfen oder erneut in die Warteschlange gestellt wird. Bei anhaltenden Fehlern sollten Sie eine Dead Letter Queue (DLQ) einrichten, um problematische Nachrichten zur späteren Prüfung zu speichern.
• Skalierbarkeit: Wenn Ihre Warteschlange workout_jobs zu wachsen beginnt, ist das ein Zeichen, dass Sie mehr Rechenleistung benötigen. Der Vorteil dieser Architektur ist, dass Sie einfach weitere Instanzen des processing-service starten können, ohne den workout-service anzufassen. RabbitMQ verteilt die Nachrichten auf alle verfügbaren Consumer.

Alternative Ansätze

• AWS SQS (Simple Queue Service): Für Produktionsumgebungen ist ein verwalteter Dienst wie SQS oft die bessere Wahl. Er ist hochgradig skalierbar und dauerhaft, und Sie müssen die zugrunde liegenden Server nicht verwalten. Die Konzepte (Warteschlangen, Producer, Consumer) sind sehr ähnlich.
• Kafka: Für Ereignis-Streaming mit hohem Durchsatz (z. B. Echtzeit-Standortverfolgung während eines Trainings) eignet sich eine Stream-Verarbeitungsplattform wie Apache Kafka möglicherweise besser. Sie ist darauf ausgelegt, sehr große Datenmengen in Echtzeit zu verarbeiten.

Fazit

Wir haben erfolgreich ein entkoppeltes, ereignisgesteuertes System für die Verarbeitung von Fitness-Trainings aufgebaut. Durch die Einführung einer Nachrichtenwarteschlange haben wir unsere Anwendung robuster, skalierbarer und in der Lage gemacht, eine reaktionsschnelle Benutzererfahrung zu bieten. Der workout-service bleibt schlank und schnell, während der processing-service unabhängig skaliert werden kann, um jede Last zu bewältigen.

Dieses Muster ist äußerst leistungsfähig und wird in unzähligen Produktionssystemen eingesetzt. Sie können es nun auf andere lang laufende Aufgaben in Ihren Anwendungen anwenden, etwa das Erzeugen von Berichten, das Transcodieren von Videos oder das Aufrufen von Drittanbieter-APIs.

Auswirkungen auf die Leistung: Ereignisgesteuerte Architekturen reduzieren die API-Antwortzeiten um 70–85 % bei Vorgängen mit Hintergrundverarbeitung. Anwendungen mit Nachrichtenwarteschlangen erreichen 99,9 % Verfügbarkeit bei Ausfällen des Verarbeitungsdienstes und eine Kostenreduktion von 50–70 % durch die Effizienz horizontaler Skalierung. Entkoppelte Dienste ermöglichen eine unabhängige Skalierung, was zu einer um 40 % besseren Ressourcennutzung und einer um 60 % schnelleren Funktionsentwicklung für Teams führt, die auf Microservices setzen.

Ihre nächsten Schritte:

• Implementieren Sie eine Dead Letter Queue (DLQ) für fehlgeschlagene Trainingsnachrichten.
• Ergänzen Sie eine komplexere Erfolgslogik.
• Speichern Sie die verarbeiteten Trainingsergebnisse in einer Datenbank (etwa MongoDB oder PostgreSQL).

Weitere Muster der Backend-Architektur finden Sie unter Node.js mit Kubernetes skalieren oder Echtzeit-Bestenlisten mit Node.js und Redis erstellen. Mehr zu ereignisgesteuerten Systemen finden Sie unter IoT-Daten mit AWS Lambda und Kinesis verarbeiten.

Ressourcen

• Offizielle Dokumentation von amqplib: https://www.squaremobius.net/amqp.node/
• RabbitMQ-Tutorials: https://www.rabbitmq.com/getstarted.html
• Open-Source-Datensatz zu Übungen: https://github.com/wrkout/exercises.json

Kurzreferenz

Häufig gestellte Fragen

F: Woran erkenne ich, ob das auf mich zutrifft? A: Besprechen Sie Ihre persönliche und familiäre Krankengeschichte mit Ihrer Ärztin oder Ihrem Arzt, um Ihre individuellen Risikofaktoren einzuschätzen und die geeigneten nächsten Schritte festzulegen.

F: Was sollte ich meine Ärztin oder meinen Arzt fragen? A: Gute Fragen sind: „Was bedeuten meine Ergebnisse?“, „Welche Möglichkeiten habe ich?“, „Welche Änderungen des Lebensstils würden helfen?“ und „Wann sollte ich zur Nachsorge kommen?“

F: Gibt es Änderungen des Lebensstils, die helfen können? A: Den meisten Erkrankungen kommen grundlegende Maßnahmen zugute: ausgewogene Ernährung, regelmäßige körperliche Aktivität (150+ Minuten/Woche), guter Schlaf (7–9 Stunden), Stressbewältigung und der Verzicht auf Tabak.

Haftungsausschluss

Die in diesem Artikel vorgestellten Algorithmen und Verfahren dienen ausschließlich der technischen Weiterbildung. Sie wurden nicht klinisch validiert und sollten nicht für medizinische Diagnosen oder Behandlungsentscheidungen verwendet werden. Holen Sie medizinischen Rat stets bei qualifizierten Fachkräften des Gesundheitswesens ein.