Java-Entwicklung: So verwenden Sie Apache Kafka Streams für die Echtzeit-Stream-Verarbeitung und -Berechnung

Java-Entwicklung: So verwenden Sie Apache Kafka Streams für die Echtzeit-Stream-Verarbeitung und -Berechnung

Einführung:
Mit dem Aufkommen von Big Data und Echtzeit-Computing wird Apache Kafka Streams als Stream-Verarbeitungs-Engine immer beliebter Wird immer häufiger von Entwicklern verwendet. Es bietet eine einfache, aber leistungsstarke Möglichkeit, Echtzeit-Streaming-Daten zu verarbeiten und komplexe Stream-Verarbeitungen und Berechnungen durchzuführen. In diesem Artikel wird die Verwendung von Apache Kafka Streams für die Echtzeit-Stream-Verarbeitung und -Berechnung vorgestellt, einschließlich der Konfiguration der Umgebung, des Schreibens von Code und Beispieldemonstrationen.

1. Vorbereitung:

1. Apache Kafka installieren und konfigurieren: Sie müssen Apache Kafka herunterladen und installieren und den Apache Kafka-Cluster starten. Eine detaillierte Installation und Konfiguration finden Sie in der offiziellen Apache Kafka-Dokumentation.
2. Abhängigkeiten einführen: Kafka Streams-bezogene Abhängigkeiten in das Java-Projekt einführen. Mit Maven können Sie beispielsweise die folgenden Abhängigkeiten zu den POM-Cluster-Verbindungsinformationen hinzufügen. Das Folgende ist ein einfacher Beispielcode:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>2.8.1</version>
</dependency>

Fügen Sie Stream-Verarbeitungs- und Berechnungslogik hinzu:

Nachdem Sie eine Kafka Streams-Anwendung erstellt haben, müssen Sie eine spezifische Stream-Verarbeitungs- und Berechnungslogik hinzufügen. Als einfaches Beispiel gehen wir davon aus, dass wir eine Zeichenfolgennachricht von einem Kafka-Thema namens „input-topic“ erhalten, eine Längenberechnung für die Nachricht durchführen und das Ergebnis dann an ein Kafka-Thema namens „output-topic“ senden. Das Folgende ist ein Beispielcode:
1. 
   

   import org.apache.kafka.streams.KafkaStreams;
   import org.apache.kafka.streams.StreamsBuilder;
   import org.apache.kafka.streams.StreamsConfig;
   import org.apache.kafka.streams.Topology;
   
   import java.util.Properties;
   
   public class KafkaStreamsApp {
   
       public static void main(String[] args) {
           Properties props = new Properties();
           props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
           props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   
           StreamsBuilder builder = new StreamsBuilder();
           // 在这里添加流处理和计算逻辑
   
           Topology topology = builder.build();
           KafkaStreams streams = new KafkaStreams(topology, props);
           streams.start();
   
           // 添加Shutdown Hook，确保应用程序在关闭时能够优雅地停止
           Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
       }
   }

Im obigen Beispielcode wird zunächst ein KStream-Objekt aus dem Eingabethema erstellt und dann jede Nachricht mithilfe der flatMapValues-Operation in Wörter aufgeteilt und statistisch gezählt. Abschließend werden die Ergebnisse an das Ausgabethema gesendet.

3. Beispieldemonstration:
2. Um unsere Echtzeit-Stream-Verarbeitungs- und Computeranwendungen zu überprüfen, können Sie das Kafka-Befehlszeilentool verwenden, um Nachrichten zu senden und Ergebnisse anzuzeigen. Hier sind die Schritte für eine Beispieldemonstration:
   

Eingabe- und Ausgabethemen erstellen:

Führen Sie die folgenden Befehle in der Befehlszeile aus, um Kafka-Themen mit den Namen „Eingabe-Thema“ und „Ausgabe-Thema“ zu erstellen:

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Arrays;

public class KafkaStreamsApp {

    // 省略其他代码...
    
    public static void main(String[] args) {
        // 省略其他代码...
        
        KStream<String, String> inputStream = builder.stream("input-topic");
        KTable<String, Long> wordCounts = inputStream
                .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\W+")))
                .groupBy((key, word) -> word)
                .count();

        wordCounts.toStream().to("output-topic");

        // 省略其他代码...
    }
}

Senden eine Nachricht zum Eingabethema:

Führen Sie den folgenden Befehl in der Befehlszeile aus, um einige Nachrichten an „Eingabethema“ zu senden:
1. 
   

   bin/kafka-topics.sh --create --topic input-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
   bin/kafka-topics.sh --create --topic output-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

bin/kafka-console-consumer.sh --topic Ausgabethema --from- begin --bootstrap -server localhost:9092

bin/kafka-console-producer.sh --topic input-topic --bootstrap-server localhost:9092
>hello world
>apache kafka streams
>real-time processing
>```

3. 查看结果：
在命令行中执行以下命令，从"output-topic"中消费结果消息：

real-time: 1
2. processing: 1
   apache: 1
kafka: 1

streams: 1

hello: 2

world: 1

可以看到，输出的结果是单词及其对应的计数值：

Das obige ist der detaillierte Inhalt vonJava-Entwicklung: So verwenden Sie Apache Kafka Streams für die Echtzeit-Stream-Verarbeitung und -Berechnung. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!