Spring Boot と Kafka を実際に使ってみましょう。

Kafka クラスターのインストール、構成、起動

Kafka は Zookeeper に依存する必要があり、Zookeeper 自体を統合します。 Zookeeper では、クラスターの高可用性を確保するために少なくとも 3 つのノードが必要です。以下は、単一の Linux マシンで 3 つの Kafka ノードの疑似クラスター モードを作成する方法です。

1. パッケージをダウンロードする

ダウンロードアドレス: http://kafka.apache.org/downloads

2. パッケージを解凍する

tar -zxvf kafka_2.11-1.0.0.tgz\mv kafka_2.11-1.0.0 kafka1\mv kafka_2.11-1.0.0 kafka2\mv kafka_2.11-1.0.0 kafka3

3. ZKクラスターを作成する

ZK 構成ファイル kafka1-3/config/zookeeper.properties を変更し、対応するパラメータをそれぞれ変更します。

データディレクトリ=/usr/ローカル/kafka/zookeeper1
 dataLogDir=/usr/ local /kafka/zookeeper/log
クライアントポート=2181
最大クライアント接続数=0
ティックタイム=2000
初期制限=100
同期制限=5
サーバー1=127.0.0.1:2888:3888
サーバー2=127.0.0.1:4888:5888
サーバー3=127.0.0.1:6888:7888

/usr/local/kafka/zookeeper1-3ディレクトリにそれぞれmyidファイルを作成し、内容は1〜3に対応します。

ZK を起動し、それぞれ Kafka1-3 ディレクトリに移動します。

 bin/zookeeper-server-start.sh config/zookeeper.properties &

起動レポート ファイルが失敗しました。ファイル ディレクトリを手動で作成し、対応する権限を付与する必要があります。

4. Kafka クラスターを作成する

設定ファイル: kafka1-3/config/server.properties、それぞれ対応するパラメータを変更します。

ブローカーID=1
 zookeeper.connect = ローカルホスト:2181、ローカルホスト:2182、ローカルホスト:2183
リスナー=プレーンテキスト://192.168.12.11:9091
 log.dirs=/tmp/kafka-logs-1

Kafka を起動し、それぞれ Kafka1-3 ディレクトリに移動します。

 bin/kafka-server-start.sh config/server.properties &

起動レポート ファイルが失敗しました。ファイル ディレクトリを手動で作成し、対応する権限を付与する必要があります。

5. クラスターテスト

kafka1 にメッセージを送信:

 bin/kafka-console-producer.sh --broker-list localhost:9091 --topic テスト 

kafka2 および kafka3 でメッセージを消費します。

 bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic 私のレプリケートされたトピック 

Spring Boot 統合 Kafka プラクティス

1. spring-kafka依存関係を追加する

<spring-kafka.version>2.1.0.RELEASE</spring-kafka.version> 
 
 <! --スプリングカフカ-->  
 <依存関係>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>スプリングカフカ</artifactId>
    <バージョン>${spring-kafka.version}</バージョン>
 </依存関係>

2. Spring Bootの自動構成を追加する

自動構成クラス:

• org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration

構成プロパティクラス:

• org.springframework.boot.autoconfigure.kafka.Kafkaプロパティ

春：
  カフカ:
    ブートストラップサーバー:
      - 192.168.101.137:9091
      - 192.168.101.137:9092
      - 192.168.101.137:9093
    プロデューサー：
      再試行: 0
      バッチサイズ: 16384
      バッファメモリ: 33554432
キー-serializer: org.apache.kafka.common.serialization.StringSerializer
      値シリアライザー: org.apache.kafka.common.serialization.StringSerializer
    消費者：
グループID: foo
      自動オフセットリセット: 最も早い
      自動コミットを有効にする: true  
      自動コミット間隔: 100
キー-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      値デシリアライザー: org.apache.kafka.common.serialization.StringDeserializer

3. メッセージを送信する

オートワイヤード
プライベート KafkaTemplate kafkaTemplate; 
 
 @GetMapping( "/送信" )
パブリックオブジェクト送信(文字列メッセージ) {
    kafkaTemplate.send( "テスト" , "名前" , msg);
戻る  「送信OK」 ;
 }

4. メッセージを受信する

任意の Bean で、メッセージの受信をサポートするために @KafkaListener を追加します。

 @KafkaListener(トピック = "テスト" )
パブリックvoid processMessage(文字列コンテンツ) {
    logger.info( "受信したメッセージ、トピック:テスト、メッセージ:{}" 、コンテンツ);
 }