PHP- und Apache Kafka-Integration für effiziente Nachrichtenwarteschlange und -verteilung

Mit der kontinuierlichen Weiterentwicklung moderner Internetanwendungen müssen immer mehr Anwendungen große Mengen an Datenkommunikation bewältigen. Die traditionelle Art und Weise, diese Datenkommunikation zu handhaben, besteht darin, Polling oder das Blockieren von E/A zu verwenden, aber diese Methoden können den Anforderungen moderner Anwendungen nicht mehr gerecht werden, da sie sehr ineffizient sind. Um dieses Problem zu lösen, hat die Industrie eine Technologie namens Nachrichtenwarteschlangen- und -verteilungssystem entwickelt.

Im Nachrichtenwarteschlangen- und Verteilungssystem sendet der Nachrichtenproduzent die Nachricht an die Warteschlange, während der Nachrichtenkonsument die Nachricht aus der Warteschlange erhält und entsprechende Vorgänge ausführt. Dieser Ansatz kann die Effizienz der Datenkommunikation erheblich verbessern, da Probleme wie Abfragen und das Blockieren von E/A vermieden werden können.

In diesem Artikel besprechen wir, wie man mithilfe der PHP- und Apache Kafka-Integration eine effiziente Nachrichtenwarteschlange und -verteilung erreicht.

Einführung in Apache Kafka

Apache Kafka ist ein skalierbares verteiltes Messaging-System mit hohem Durchsatz, geringer Latenz. Es kann große Nachrichtenmengen verarbeiten und horizontal skalieren, um höhere Lasten zu bewältigen. Zu den Hauptkomponenten von Apache Kafka gehören:

1. Broker: Jeder Knoten im Kafka-Cluster ist ein Broker und für die Speicherung und Weiterleitung von Nachrichten verantwortlich.
2. Thema: Jede Nachricht muss einem Thema zugeordnet werden, das ein logisches Konzept der Nachrichtenproduktion und -nutzung darstellt.
3. Partition: Jedes Thema kann in mehrere Partitionen unterteilt werden, und jede Partition enthält mehrere geordnete Nachrichten.
4. Produzent: Nachrichtenproduzent, sendet Nachrichten an den Broker.
5. Consumer: Nachrichtenkonsument, liest Nachrichten vom Broker.
6. Verbrauchergruppe: Eine Gruppe von Verbrauchern konsumiert gemeinsam Nachrichten in einer oder mehreren Partitionen.
7. Offset: Die Nummer der Nachricht, die zur eindeutigen Identifizierung einer Nachricht verwendet wird.

PHP integriert mit Apache Kafka

Um Apache Kafka verwenden zu können, müssen wir die Kafka-Erweiterung für PHP verwenden. Diese Erweiterung stellt alle APIs bereit, die PHP zum Betrieb von Kafka benötigt.

Zuerst müssen wir die Kafka-Erweiterung installieren, die wir über PECL installieren können:

pecl install kafka

Nach der Installation der Erweiterung können Sie sie verwenden. Hier ist ein einfaches Beispiel für die Nachrichtenproduktion und -nutzung mit PHP und Apache Kafka:

<?php
$brokers = 'kafka:9092';    // Kafka集群地址
$topic = 'test';            // Topic名称

// 创建一个Kafka生产者
$producer = new RdKafkaProducer();
$producer->setLogLevel(LOG_DEBUG);
$producer->addBrokers($brokers);

// 创建一个Kafka消费者
$conf = new RdKafkaConf();
$conf->set('group.id', 'myGroup');
$consumer = new RdKafkaConsumer($conf);
$consumer->addBrokers($brokers);

// 生产消息
$topicProducer = $producer->newTopic($topic);
for ($i = 0; $i < 10; $i++) {
    $topicProducer->produce(RD_KAFKA_PARTITION_UA, 0, 'Message ' . $i);
}

// 消费消息
$topicConsumer = $consumer->newTopic($topic);
$topicConsumer->consumeStart(0, RD_KAFKA_OFFSET_BEGINNING);
while (true) {
    $message = $topicConsumer->consume(0, 1000);
    if (null === $message) {
        continue;
    }
    if ($message->err) {
        throw new Exception('Error occurred while consuming message');
    }
    echo $message->payload . PHP_EOL;
}

In diesem Beispiel erstellen wir zunächst einen Kafka-Produzenten und einen Kafka-Konsumenten. Dann haben wir im Produzenten 10 Nachrichten an das angegebene Thema gesendet; im Verbraucher haben wir die Nachrichten vom angegebenen Thema konsumiert und ihren Inhalt ausgegeben.

Zu diesem Zeitpunkt haben wir die einfache Nachrichtenproduktion und -nutzung mit PHP und Apache Kafka erfolgreich implementiert. Als Nächstes besprechen wir, wie man mit PHP und Apache Kafka erweiterte Funktionen implementiert.

Erweiterte Anwendungsbeispiele

In tatsächlichen Anwendungen müssen wir normalerweise einige erweiterte Funktionen implementieren, wie zum Beispiel:

1. Nachrichtenverteilung: Nachrichten an bestimmte Verbraucher senden.
2. Verbrauchergruppe: Ermöglicht mehreren Verbrauchern, gemeinsam Nachrichten zu einem oder mehreren Themen zu konsumieren.
3. Offset-Konfiguration: ermöglicht die Steuerung, wo Nachrichten gelesen werden.

Hier besprechen wir, wie diese Funktionen implementiert werden.

Nachrichtenverteilung

In praktischen Anwendungen müssen wir normalerweise den Nachrichtenfluss steuern. Beispielsweise möchten wir möglicherweise, dass nur bestimmte Verbraucher bestimmte Nachrichten konsumieren. Um diese Funktionalität zu erreichen, können wir für jeden Verbraucher eine Warteschlange erstellen und dann bestimmte Nachrichten bestimmten Warteschlangen zuweisen.

Hier ist ein Beispiel, das zwei Verbraucher verwendet, um zwei verschiedene Aufgaben zu konsumieren.

<?php

$brokers = 'kafka:9092';    // Kafka集群地址
$topic = 'test';            // Topic名称

// 创建一个Kafka消费者组
$conf = new RdKafkaConf();
$conf->set('group.id', 'myGroup');
$consumer = new RdKafkaKafkaConsumer($conf);
$consumer->subscribe([$topic]);

// 创建两个Kafka生产者，一个生产者用于向消费者1发送消息，另一个生产者用于向消费者2发送消息
$producer1 = new RdKafkaProducer();
$producer1->addBrokers($brokers);
$producer1Topic = $producer1->newTopic($topic . '_1');

$producer2 = new RdKafkaProducer();
$producer2->addBrokers($brokers);
$producer2Topic = $producer2->newTopic($topic . '_2');

// 消费消息
while (true) {
    $message = $consumer->consume(1000);
    if (null === $message) {
        continue;
    }
    if ($message->err) {
        throw new Exception('Error occurred while consuming message');
    }

    echo 'Received message: ' . $message->payload . PHP_EOL;

    // 根据消息内容分配给不同的生产者
    if ($message->payload === 'task1') {
        $producer1Topic->produce(RD_KAFKA_PARTITION_UA, 0, $message->payload);
    } elseif ($message->payload === 'task2') {
        $producer2Topic->produce(RD_KAFKA_PARTITION_UA, 0, $message->payload);
    }
}

In diesem Beispiel verwenden wir zwei Produzenten, um Nachrichten an zwei verschiedene Verbraucher zu verteilen. Wenn ein Verbraucher eine Nachricht erhält, können wir sie basierend auf dem Nachrichteninhalt einem bestimmten Produzenten zuordnen. Diese Methode kann uns helfen, den Nachrichtenfluss zu kontrollieren und eine redundante Verarbeitung von Nachrichten zu vermeiden.

Verbrauchergruppe

Bei gewöhnlichen Kafka-Verbrauchern konsumieren verschiedene Verbraucher in derselben Gruppe gemeinsam dasselbe Thema und erhalten dieselbe Nachricht. Dies liegt daran, dass Kafka Partitionen automatisch ausgleicht und sicherstellt, dass jede Partition nur von einem Verbraucher verarbeitet wird.

In PHP können wir group.id verwenden, um Verbraucher zu gruppieren und die Funktion von Verbrauchergruppen zu implementieren.

Das Folgende ist ein Beispiel für eine Kafka-Konsumentengruppe, die Nachrichten innerhalb derselben Gruppe parallel verarbeiten kann:

<?php

$brokers = 'kafka:9092';    // Kafka集群地址
$topic = 'test';            // Topic名称

// 创建一个Kafka消费者组
$conf = new RdKafkaConf();
$conf->set('group.id', 'myGroup');
$conf->set('metadata.broker.list', $brokers);
$conf->set('enable.auto.commit', 'false');
$consumer = new RdKafkaKafkaConsumer($conf);

// 添加需要订阅的topic
$consumer->subscribe([$topic]);

// 处理消息
while (true) {
    $message = $consumer->consume(1000);
    if (null === $message) {
        continue;
    }
    if ($message->err) {
        throw new Exception('Error occurred while consuming message');
    }

    echo 'Received message: ' . $message->payload . PHP_EOL;

    // 处理完消息后手动提交offset
    $consumer->commit();
}

In diesem Beispiel erstellen wir eine Kafka-Konsumentengruppe und fügen Themen hinzu, die abonniert werden müssen. Wir können dann Nachrichten innerhalb derselben Gruppe parallel verarbeiten.

Hinweis: In einer Verbrauchergruppe verbrauchen mehrere Verbraucher gemeinsam eine oder mehrere Partitionen. Beim Datenverbrauch müssen Sie auf das Problem der Multithreading-Verarbeitung derselben Daten achten.

Offset-Konfiguration

In Kafka hat jede Partition einen unabhängigen Offset. Der Verbraucher kann steuern, wo in der Partition er liest und somit welche Nachrichten er liest. Der Verbraucher kann mit dem Lesen der letzten Nachricht oder der neuesten Nachricht beginnen.

In PHP können wir Offset verwenden, um die Leseposition von Nachrichten zu steuern. Das Folgende ist ein Beispiel für die Offset-Konfiguration:

<?php

$brokers = 'kafka:9092';    // Kafka集群地址
$topic = 'test';            // Topic名称

// 创建一个Kafka消费者
$conf = new RdKafkaConf();
$conf->set('group.id', 'myGroup');
$consumer = new RdKafkaKafkaConsumer($conf);

// 订阅topic
$topicConf = new RdKafkaTopicConf();
$topicConf->set('auto.offset.reset', 'earliest');
$topic = $consumer->newTopic($topic, $topicConf);
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

// 消费消息
while (true) {
    $message = $topic->consume(0, 1000);
    if (null === $message) {
        continue;
    }
    if ($message->err) {
        throw new Exception('Error occurred while consuming message');
    }

    echo 'Received message: ' . $message->payload . PHP_EOL;
}

In diesem Beispiel verwenden wir auto.offset.reset, um die Offset-Konfiguration festzulegen. Diese Konfiguration weist den Verbraucher an, mit dem Verarbeiten von Nachrichten ab dem frühesten Offset zu beginnen.

In praktischen Anwendungen können je nach Bedarf unterschiedliche Offsets konfiguriert werden. Wenn der Produzent beispielsweise einige Nachrichten nicht verarbeiten kann, müssen wir möglicherweise das Lesen der Nachrichten an der Stelle neu starten, an der die fehlgeschlagene Nachricht zuvor verarbeitet wurde.

Fazit

In diesem Artikel haben wir besprochen, wie man mithilfe der PHP- und Apache Kafka-Integration eine effiziente Nachrichtenwarteschlange und -verteilung erreichen kann. Wir haben zunächst die Grundlagen von Apache Kafka vorgestellt und dann besprochen, wie man die Kafka-Erweiterung für PHP verwendet, um die Produktion und den Konsum von Nachrichten zu implementieren. Abschließend haben wir besprochen, wie einige erweiterte Funktionen wie Nachrichtenverteilung, Verbrauchergruppen und Offset-Konfiguration implementiert werden.

Durch die Integration von PHP und Apache Kafka können wir eine effiziente Nachrichtenwarteschlange und -verteilung implementieren und so die Reaktionsgeschwindigkeit und den Durchsatz der Anwendung verbessern. Wenn Sie eine Anwendung entwickeln, die große Mengen an Datenkommunikation bewältigen muss, sind Apache Kafka und die Kafka-Erweiterung für PHP möglicherweise eine gute Wahl.

Das obige ist der detaillierte Inhalt vonPHP- und Apache Kafka-Integration für effiziente Nachrichtenwarteschlange und -verteilung. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!