👇🏽 🚶🏼 🍠 Apache Kafka-私の概要 🤛 🏐 👨🏿‍🏫

これは私の概要です。ここでは、次のようなカフカの概念について簡単かつ本質的に触れます。

-トピック

-サブスクライバー（コンシューマー）

-出版社（プロデューサー）

-グループ、グループ、パーティション

-ストリーム

カフカ-基本

Kafkaを勉強するとき、質問を使って、例を使用して実験的に受け取る必要がある答えがありました。これはこの要約で概説されています。開始方法と開始場所以下のリンクのいずれかを資料に記載します。

Apache Kafkaは、Javaプラットフォームのメッセージマネージャーです。 Kafkaには、 パブリッシャーがメッセージを書き込むメッセージサブジェクトがあり、これらのメッセージを読むトピックにサブスクライバーがいます。ディスパッチプロセスのすべてのメッセージはディスクに書き込まれ、コンシューマーに依存しません。

Kafkaには、テーマ、セクション、既製のパブリッシャー、登録者などを作成するためのユーティリティセットが含まれています。Kafkaを動作させるには、ZooKeeperコーディネーターが必要です。）、バッチファイルはそれぞれのbinフォルダーにあり、ユーティリティもあります。

に入るユーティリティでKafkaテーマを作成します

kafka-topics.bat --create --zookeeper localhost：2181 --replication-factor 1 --partitions 1 --topic out-topic

ここでは、zookeeperサーバーを示し、replication-factorはメッセージログレプリカの数、partitionsはトピック内のセクションの数（以下を参照）、トピック自体は「アウトトピック」です。

簡単なテストのために、付属のkafka-console-consumerおよびkafka-console-producerアプリケーションを使用できますが、私は自分でやるつもりです。サブスクライバーは実際にはグループ化されています。これにより、さまざまなアプリケーションがトピックからのメッセージを並行して読み取ることができます。

各アプリケーションは独自のキューを持ち、そこから最後に読み取られたメッセージ（オフセット）のポインターを移動します。これはコミットと呼ばれます。そのため、パブリッシャーがトピックにメッセージを送信した場合、実行中または接続するとすぐに、このトピックの受信者による読み取りが保証されます。さらに、同じトピックから読み取るが、異なるグループにある異なるクライアント（client.id）がある場合、それらは互いに関係なく、準備ができたときにメッセージを受信します。

したがって、メッセージのフォロワーと、1つのトピックからの消費者による独立した読書を想像できます。

しかし、トピック内のメッセージが送信されるよりも早く到着し始める場合があります。消費者はそれらをより長く処理します。これを行うために、トピックはセクション（パーティション）を提供し、このトピックの1つのグループでコンシューマーを実行できます。

その後、負荷分散が行われ、トピックおよびグループ内のすべてのメッセージが1つのコンシューマーを通過するわけではありません。そして、メッセージをセクションに分配する方法が選択されます。いくつかの戦略があります：ラウンドロビン-これは、キーのハッシュ値による円内、または書き込むセクション番号の明示的な指示です。この場合のサブスクライバは、セクション全体に均等に分散されます。たとえば、グループ内のグループよりもサブスクライバーが多い場合、だれかがメッセージを受信しません。このようにして、スケーラビリティを改善するためのセクションが作成されます。

たとえば、1つのセクションでトピックを作成した後、2つのセクションに変更しました。

kafka-topics.bat --zookeeper localhost：2181 --alter --topic out-topic --partitions 2

1つのトピックで1つのグループにパブリッシャーと2つのサブスクライバーを起動しました（Javaプログラムの例を以下に示します）。グループ名とクライアントIDを構成する必要はありません。Kafkaがこれを処理します。

my_kafka_run.cmd com.home.SimpleProducer out-topic（パブリッシャー）

my_kafka_run.cmd com.home.SimpleConsumer out-topic testGroup01 client01（最初のサブスクライバー）

my_kafka_run.cmd com.home.SimpleConsumer out-topic testGroup01 client02（2番目のサブスクライバー）

ペアのパブリッシャーでキーの入力を開始すると、値は誰が取得したかがわかります。したがって、たとえば、キーハッシュの配布戦略に従って、メッセージm：1はクライアントclient01に到達しました。

メッセージn：クライアントclient02への1

ペアを指定せずにkey：valueのペアを入力し始めると（パブリッシャーでこの機会を作りました）、サークル戦略が選択されます。最初のメッセージ「m」はclient01にヒットし、すでに3回client02にヒットしています。

そして、例えば、この形式のキーのようなセクションを持つ別のオプション：値：パーティション

以前、ハッシュ戦略では、m：1は別のクライアント（client01）に移動し、現在はセクション（No. 1、0から番号付け）の明示的な指示でclient02に移動しています。

同じトピックに対して異なるグループ名testGroup02でサブスクライバーを起動すると、メッセージはサブスクライバーに並行して独立して送信されます。最初のものが読み、2番目のものがアクティブでなかった場合、アクティブになるとすぐに読みます。

それぞれグループ、トピックの説明を見ることができます：

kafka-consumer-groups.bat --bootstrap-server localhost：9092 --describe --group testGroup01

kafka-topics.bat --describe --zookeeper localhost：2181 --topic out-topic

SimpleProducerコード

public class SimpleProducer { public static void main(String[] args) throws Exception { // Check arguments length value if (args.length == 0) { System.out.println("Enter topic name"); return; } //Assign topicName to string variable String topicName = args[0].toString(); System.out.println("Producer topic=" + topicName); // create instance for properties to access producer configs Properties props = new Properties(); //Assign localhost id props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); //Set acknowledgements for producer requests. props.put("acks", "all"); //If the request fails, the producer can automatically retry, props.put("retries", 0); //Specify buffer size in config props.put("batch.size", 16384); //Reduce the no of requests less than 0 props.put("linger.ms", 1); //The buffer.memory controls the total amount of memory available to the producer for buffering. props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer(props); BufferedReader br = null; br = new BufferedReader(new InputStreamReader(System.in)); System.out.println("Enter key:value, q - Exit"); while (true) { String input = br.readLine(); String[] split = input.split(":"); if ("q".equals(input)) { producer.close(); System.out.println("Exit!"); System.exit(0); } else { switch (split.length) { case 1: // strategy by round producer.send(new ProducerRecord(topicName, split[0])); break; case 2: // strategy by hash producer.send(new ProducerRecord(topicName, split[0], split[1])); break; case 3: // strategy by partition producer.send(new ProducerRecord(topicName, Integer.valueOf(split[2]), split[0], split[1])); break; default: System.out.println("Enter key:value, q - Exit"); } } } } }

SimpleConsumerコード

 public class SimpleConsumer { public static void main(String[] args) throws Exception { if (args.length != 3) { System.out.println("Enter topic name, groupId, clientId"); return; } //Kafka consumer configuration settings final String topicName = args[0].toString(); final String groupId = args[1].toString(); final String clientId = args[2].toString(); Properties props = new Properties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", groupId); props.put("client.id", clientId); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); //props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); //props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); //Kafka Consumer subscribes list of topics here. consumer.subscribe(Arrays.asList(topicName)); //print the topic name System.out.println("Subscribed to topic=" + topicName + ", group=" + groupId + ", clientId=" + clientId); SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss"); // looping until ctrl-c while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) // print the offset,key and value for the consumer records. System.out.printf("offset = %d, key = %s, value = %s, time = %s \n", record.offset(), record.key(), record.value(), sdf.format(new Date())); } } }

プログラムを実行するために、my_kafka_run.cmdというバッチファイルを作成しました

 @echo off set CLASSPATH="C:\Project\myKafka\target\classes"; for %%i in (C:\kafka_2.11-1.1.0\libs\*) do ( call :concat "%%i" ) set COMMAND=java -classpath %CLASSPATH% %* %COMMAND% :concat IF not defined CLASSPATH ( set CLASSPATH="%~1" ) ELSE ( set CLASSPATH=%CLASSPATH%;"%~1" )

起動例：

my_kafka_run.cmd com.home.SimpleConsumer out-topic testGroup02 client01

カフカストリーム

そのため、Kafkaのフローは、特定の操作、変換を実行した後、たとえば、別のトピックに結果を追加したり、一般的にどこにでもデータベースに保存したりできるトピックから取得される一連のイベントです。操作は、フィルタリング（フィルター）、変換（マップ）、または集約（カウント、合計、平均）のようなものです。これを行うために、対応するクラスKStream、KTableがあります。KTableは、新しいメッセージがトピックに到着すると常に更新される現在の集計値を持つテーブルとして表すことができます。これはどうですか？

たとえば、発行者はイベント（メッセージ）の件名に書き込み、Kafkaはすべてのメッセージをメッセージログに保存します。メッセージログには、たとえば7日間の保持ポリシーがあります。たとえば、相場変更イベントはストリームであるため、平均値を見つけてから、ジャーナルから履歴を取得して平均を計算するストリームを作成します。ここで、キーは株式であり、値は平均です（これは既にステータスのあるテーブルです）。ここには機能があります-フィルタリングなどの操作とは異なり、集計操作はその状態を保持します。したがって、サブジェクトに新しく到着したメッセージ（イベント）は計算の対象となり、結果は保存（状態ストア）され、その後、新しく到着したメッセージはジャーナルに書き込まれ、Streamはそれらを処理し、既に保存された状態に変更を追加します。フィルタリング操作では、状態を保存する必要はありません。そして、ここでも、出版社に関係なく、ストリームがこれを行います。たとえば、パブリッシャーはメッセージを書き込み、プログラム-ストリームはこの時点では機能せず、何も失われず、すべてのメッセージはログに保存され、プログラムストリームがアクティブになるとすぐに、計算を行い、状態を保存し、読み取りメッセージのオフセットを実行しますそれらは読み取られます）、将来的には返されません。さらに、これらのメッセージはジャーナル（kafka-logs）を離れます。ここでは、明らかに、主なことはログ（kafka-logs）とそのストレージポリシーがこれを許可していることです。デフォルトでは、Kafka Streamは状態をRocksDBに保存します。メッセージログとそれに関連するすべて（トピック、オフセット、スレッド、クライアントなど）は、構成ファイル「config \ server.properties」のパラメーター「log.dirs = kafka-logs」で指定されたパスに沿って配置され、ログストレージポリシーも示されます「Log.retention.hours = 48」。ログの例

また、ストリーム状態を含むデータベースへのパスは、アプリケーションパラメータで指定されます

config.put（StreamsConfig.STATE_DIR_CONFIG、 "C：/kafka_2.11-1.1.0/state"）;

状態は、アプリケーションID ごとに個別に保存されます（ StreamsConfig.APPLICATION_ID_CONFIG ）。例

次に、Streamの動作を確認しましょう。サンプルからStreamアプリケーションを準備しましょう。このアプリケーションは、同じ単語の数とアプリケーション、パブリッシャー、サブスクライバーを考慮した配信（実験用の改良を加えたもの）です。トピック内のトピックに書き込みます

my_kafka_run.cmd com.home.SimpleProducerのトピック

Streamアプリケーションはこのトピックを読んで同一の単語の数をカウントします。状態を維持し、アウトトピックを別のトピックにリダイレクトすることは明示的ではありません。ここで、ログと状態（状態ストア）の関係を明確にします。そして、ZooKeeperとKafkaサーバーが開始されます。 App-ID = app_01でストリームを開始します

my_kafka_run.cmd com.home.KafkaCountStreamイントピックapp_01

パブリッシャーとサブスクライバー、それぞれ

my_kafka_run.cmd com.home.SimpleProducerのトピック

my_kafka_run.cmd com.home.SimpleConsumer out-topic testGroup01 client01

ここにあります：

単語の入力を開始し、そのカウントを確認します。どのStream App-IDがそれらをカウントしたかを示します

作業は独立して行われます。ストリームを停止してトピックへの書き込みを続けると、開始時にカウントされます。次に、App-ID = app_02（これもアプリケーションですが、IDが異なります）で2番目のストリームを接続し、ログ（保持ポリシーに従って保存される一連のイベント）を読み取り、数をカウントし、状態を保存して結果を表示します。したがって、異なる時点で動作を開始する2つのスレッドは同じ結果になりました。

今、私たちのジャーナルが古くなっていることを考えてみましょう（保持ポリシー）またはそれを削除し（実行できます）、App-ID = app_03で3番目のストリームを接続します（これのためにKafkaを停止し、kafka-logsを削除して再び開始します）そして新しいトピックを入力します最初の（app_01）スレッドがカウントし続け、新しい3番目のスレッドがゼロから開始されたことを確認してください。

その後、app_02ストリームを実行すると、最初のストリームに追いつき、値が等しくなります。この例から、Kafkaが現在のログを処理する方法、以前に保存した状態に追加する方法などが明らかになりました。

KafkaCountStreamコード

 public class KafkaCountStream { public static void main(final String[] args) throws Exception { // Check arguments length value if (args.length != 2) { System.out.println("Enter topic name, appId"); return; } String topicName = args[0]; String appId = args[1]; System.out.println("Count stream topic=" + topicName +", app=" + appId); Properties config = new Properties(); config.put(StreamsConfig.APPLICATION_ID_CONFIG, appId); config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()); config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 2000); config.put(StreamsConfig.STATE_DIR_CONFIG, "C:/kafka_2.11-1.1.0/state"); StreamsBuilder builder = new StreamsBuilder(); KStream<String, String> textLines = builder.stream(topicName); // State store KTable<String, Long> wordCounts = textLines .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+"))) .groupBy((key, word) -> word) .count(); // out to another topic KStream<String, String> stringKStream = wordCounts.toStream() .map((k, v) -> new KeyValue<>(appId + "." + k, v.toString())); stringKStream.to("out-topic", Produced.with(Serdes.String(), Serdes.String())); KafkaStreams streams = new KafkaStreams(builder.build(), config); // additional to complete the work final CountDownLatch latch = new CountDownLatch(1); // attach shutdown handler to catch control-c Runtime.getRuntime().addShutdownHook(new Thread("streams-shutdown-hook") { @Override public void run() { System.out.println("Kafka Stream close"); streams.close(); latch.countDown(); } }); try { System.out.println("Kafka Stream start"); streams.start(); latch.await(); } catch (Throwable e) { System.exit(1); } System.out.println("Kafka Stream exit"); System.exit(0); } }

カフカのテーマは非常に広範であり、私は自分自身のために最初の一般的なプレゼンテーションを行いました:-)

材料：

開始方法と開始場所

Apache Kafka-私の概要

カフカ-基本

カフカストリーム

More articles: