時間を旅するエンジン: Kafka メッセージのタイミングの謎を解明

1. 要約

1. Kafka メッセージの遅延とタイミングの概要

Kafka メッセージのレイテンシとタイミングは、ほとんどのリアルタイム データ ストリーミング アプリケーションにとって重要です。この章では、Kafka データ ストリーム処理を理解するための重要な要素であるこれら 2 つのコア概念について詳しく説明します。

（1）Kafkaのメッセージレイテンシとは何ですか？

Kafka メッセージのレイテンシとは、メッセージがプロデューサーから送信されてからコンシューマーが受信するまでの時間差を指します。これは、データ ストリーミング アプリケーションのリアルタイム性とパフォーマンスに直接影響するため、重要な概念です。理想的には、メッセージは最小限の遅延で配信される必要がありますが、現実には、さまざまな要因によって遅延が影響を受ける可能性があります。

メッセージが遅延する要因には次のようなものがあります。

• ネットワーク遅延: メッセージはネットワークを経由して Kafka クラスターに到達し、その後コンシューマーに到達する必要があります。ネットワーク遅延は、ネットワーク トポロジ、帯域幅、ルーティングなどの要因によって影響を受ける可能性があります。
• ハードウェア パフォーマンス: ディスク、メモリ、CPU の速度など、Kafka クラスターのハードウェア パフォーマンスは、メッセージの書き込みと読み取りの速度に影響します。
• Kafka 内部処理: Kafka クラスターの内部処理機能も重要な要素です。メッセージは、パーティション分割、ログのセグメント化、レプリケーションなどの処理手順を経る必要があり、処理の遅延が発生する可能性があります。

（２）メッセージの遅延はなぜ重要なのか？

メッセージの遅延が非常に重要である理由は、リアルタイム データ処理アプリケーションの信頼性とリアルタイム パフォーマンスに直接関係しているためです。金融取引処理などの一部のアプリケーションでは、数ミリ秒の遅延でもトランザクションの失敗や不整合が発生する可能性があります。監視およびログ処理アプリケーションでは、遅延が長すぎるとデータが不正確になったり、適時性が失われたりする可能性があります。

Kafka メッセージのレイテンシを管理および最適化することは、高負荷下でもアプリケーションの応答性を維持するための重要な要素です。レイテンシの原因を理解するだけでなく、対応する最適化戦略を採用する必要もあります。

（３）Kafkaメッセージのタイミングとは何ですか？

Kafka メッセージのタイミングとは、メッセージが送信された順序で受信されることを意味します。つまり、メッセージ A がメッセージ B より前に送信された場合、メッセージ A はメッセージ B より前に消費される必要があります。時間順に処理する必要があるアプリケーションでは、メッセージの順序を維持することが重要です。

メッセージの順序を維持することは、Kafka の強力な機能です。 Kafka では、各パーティション内のメッセージが順序付けられているため、各パーティションはメッセージのタイミングを保証できます。ただし、パーティションが複数ある場合は、コンシューマーの処理速度の一貫性のなさによってタイミングが影響を受ける可能性があるため、グローバル メッセージのタイミングを維持するために何らかの戦略を採用する必要があります。

（４）メッセージ遅延とタイミングの関係

メッセージの遅延とメッセージのタイミングには密接な関係があります。メッセージの遅延が大きすぎると、遅いメッセージが早いメッセージより先に処理される可能性があるため、メッセージの時間順序が失われる可能性があります。したがって、メッセージの遅延を管理する方法を理解することには、メッセージのタイミングを維持することも含まれます。

次の章では、Kafka メッセージのレイテンシを管理および最適化する方法と、リアルタイム データ処理アプリケーションのニーズを満たすためにメッセージのタイミングを維持する方法について詳しく説明します。

2. 遅延の原因

Kafka メッセージのレイテンシを効果的に管理および最適化するには、レイテンシの原因を深く理解する必要があります。遅延の一般的な原因は次のとおりです。

（１）Kafka内部遅延

Kafka 内部レイテンシとは、 Kafka の内部コンポーネントとパーティション割り当てに関連するレイテンシを指します。これらの要因は、Kafka 内のメッセージの配布、複製、再バランス調整に影響を与える可能性があります。

• パーティションの不均一な分散: パーティションが不均一に分散されている場合、一部のパーティションが混雑し、他のパーティションが遅れて、メッセージの配信が遅れる可能性があります。
• レプリケーション ラグ: Kafka では、冗長性を確保するために、通常、メッセージが複製されます。レプリケーション ラグは、トピックのすべてのレプリカがメッセージを複製するのにかかる時間です。
• 再バランスの遅延: Kafka クラスターが再バランスされると、メッセージの再配布とレプリケーションによってメッセージ配信の遅延が発生する可能性があります。

2. メッセージ遅延の測定と監視

このセクションでは、Kafka メッセージのレイテンシを測定および監視する方法について詳しく説明します。これにより、問題をより深く理解し、レイテンシのパフォーマンスを改善するための適切なアクションを実行できるようになります。

1. レイテンシ測定

Kafka メッセージのレイテンシを効果的に管理するには、まずそれを測定できる必要があります。一般的なレイテンシ メトリックは次のとおりです。

（１）プロデューサーからカフカへの遅延

これは、メッセージがプロデューサーから Kafka クラスターに送信される間の遅延を指します。このレイテンシを測定するには、次の方法があります。

• 送信タイムスタンプを記録する: プロデューサー側で、各メッセージの送信タイムスタンプを記録します。メッセージが Kafka に正常に書き込まれると、受信タイムスタンプが記録されます。次に、これら 2 つのタイムスタンプを減算することで、メッセージのプロデューサーから Kafka へのレイテンシを取得できます。

送信および受信のタイムスタンプをログに記録する方法のコード例を次に示します。

 // 记录消息发送时间戳long sendTimestamp = System.currentTimeMillis(); ProducerRecord<String, String> record = new ProducerRecord<>("my_topic", "key", "value"); producer.send(record, (metadata, exception) -> { if (exception == null) { long receiveTimestamp = System.currentTimeMillis(); long producerToKafkaLatency = receiveTimestamp - sendTimestamp; System.out.println("生产者到Kafka 延迟：" + producerToKafkaLatency + " 毫秒"); } else { System.err.println("消息发送失败: " + exception.getMessage()); } });

（２）Kafka内部遅延

Kafka 内部レイテンシとは、Kafka クラスター内でのメッセージ配信のレイテンシを指します。これは、次のような Kafka の組み込みメトリックを使用して測定できます。

• エンドツーエンドのレイテンシをログに記録: メッセージがプロデューサーから送信されてからコンシューマーが受信するまでの合計レイテンシを測定します。これには、ネットワーク転送、パーティションのレプリケーション、再バランス調整などの各リンクの時間が含まれます。

次に例を示します。

 // 创建Kafka 消费者Properties consumerProps = new Properties(); consumerProps.put("bootstrap.servers", "kafka-broker:9092"); consumerProps.put("group.id", "my-group"); consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps); // 订阅主题consumer.subscribe(Collections.singletonList("my_topic")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { long endToEndLatency = record.timestamp() - record.timestampType().createTimestamp(); System.out.println("Log End-to-End 延迟：" + endToEndLatency + " 毫秒"); } }

（３）消費者の処理遅延

コンシューマー処理のレイテンシとは、メッセージが Kafka から受信され、コンシューマーによって実際に処理されるまでにかかる時間を指します。このレイテンシを測定するには、次の方法があります。

• 消費タイムスタンプを記録する: コンシューマー側では、各メッセージの受信タイムスタンプと処理タイムスタンプを記録します。これら 2 つのタイムスタンプの差を計算することで、メッセージのコンシューマー処理の遅延を取得できます。

以下は、消費タイムスタンプを記録する方法のコード例です。

 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps); consumer.subscribe(Collections.singletonList("my_topic")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { long receiveTimestamp = System.currentTimeMillis(); long consumerProcessingLatency = receiveTimestamp - record.timestamp(); System.out.println("消费者处理延迟：" + consumerProcessingLatency + " 毫秒"); } }

2. 監視およびメトリクスツール

Kafka メッセージのレイテンシを測定および監視する場合、適切なツールとシステムを使用することが重要です。ここでは、コード例を含め、Kafka メッセージのレイテンシを効果的に監視するのに役立つツールと手順をいくつか紹介します。

（1）Kafka組み込みメトリクス

Kafka には、さまざまな方法で監視できる組み込みメトリックが用意されています。以下は、Kafka の JMX インターフェースを通じてこれらのメトリックにアクセスする方法を示した例です。

JConsole を使用して Kafka ブローカーに直接接続します。

• Kafka ブローカーを起動します。
• JConsole (Java 監視および管理コンソール) を開きます。
• JConsole で Kafka Broker プロセスを選択します。
• さまざまなメトリックを表示するには、「kafka.server」と「kafka.consumer」に移動します。

Jolokia (Kafka JMX HTTP ブリッジ) の使用:

• Jolokia を Kafka ブローカーの JMX HTTP ブリッジとして有効にします。
• メトリクス データを取得するには、Web ブラウザーまたは HTTP リクエストを使用して Jolokia インターフェースにアクセスします。たとえば、cURL を使用して HTTP GET リクエストを作成します。

 curl http://localhost:8778/jolokia/read/kafka.server:name=BrokerTopicMetrics/TotalFetchRequestsPerSec

これは、Kafka Broker トピック メトリックに関する情報を返します。

（２）サードパーティ監視ツール

Kafka の組み込みメトリックに加えて、Prometheus や Grafana などのサードパーティの監視ツールを使用して、メトリック データを収集、視覚化、アラートすることもできます。手順は次のとおりです。

Prometheus を設定します。

• Prometheus サーバーをデプロイして構成します。
• Kafka を監視するための Prometheus 構成ファイルを作成し、メトリック データをスクレイピングする頻度とターゲットを定義します。
• Prometheus サーバーを起動します。

Grafana ダッシュボードを設定します。

• Grafana サーバーをデプロイして構成します。
• Prometheus をデータ ソースとして使用して、Grafana でダッシュボードを作成します。
• メトリック クエリを追加し、アラート ルールと視覚化チャートを構成します。

Kafka のレイテンシ データを視覚化します。

Grafana ダッシュボードでは、プロデューサーから Kafka へのレイテンシ、コンシューマー処理のレイテンシなど、Kafka レイテンシ データを視覚化するためのさまざまなグラフを設定できます。アラート ルールを設定すると、適切なタイミングで通知を受信して​​対処することもできます。

（３）監視ツールの設定と使用

監視ツールを設定して使用するには、次の手順を実行する必要があります。

メトリクスを定義する: プロデューサーから Kafka までのレイテンシ、コンシューマー処理のレイテンシなど、測定する主要なメトリクスを特定します。

アラート ルールを設定する: 問題に迅速に対応するには、メトリック データが事前定義されたしきい値を超えたときに通知を受信するアラート ルールを設定します。

ビジュアル ダッシュボードを作成する: Grafana などの監視ツールを使用してビジュアル ダッシュボードを作成し、メトリック データを一元的に表示して、レイテンシをリアルタイムで監視します。構成可能なグラフとダッシュボードにより、データの傾向をより深く理解できます。

上記の手順とツールは、Kafka メッセージのレイテンシをより適切に測定および監視し、システムのパフォーマンスと信頼性を維持するためのタイムリーなアクションを実行するのに役立ちます。

3. メッセージの遅延を減らす

Kafka メッセージの遅延の原因と、それを測定および監視する方法がわかったので、次はメッセージの遅延を減らす方法を検討してみましょう。 Kafka メッセージの遅延を減らすのに役立つ効果的な方法をいくつか紹介します。

1. Kafka の設定を最適化する

（１）生産者と消費者のパラメータ

プロデューサーパラメータの例:

 # 生产者参数示例acks=all compression.type=snappy linger.ms=20 max.in.flight.requests.per.cnotallow=1

• acks は、プロデューサーがすべてのパーティション レプリカからの確認応答を待機するように all に設定されています。これにより信頼性は向上しますが、待ち時間が長くなる可能性があります。
• compression.type は Snappy を使用してメッセージを圧縮し、ネットワーク転送の遅延を削減します。
• linger.ms は 20 ミリ秒に設定されており、プロデューサーはメッセージを送信する前にさらにメッセージを待機できます。これにより、短時間のメッセージ配信遅延が削減されます。
• パーティション レプリカからの確認応答を受信する前に新しいメッセージが送信されないように、max.in.flight.requests.per.connection は 1 に設定されています。

コンシューマーパラメータの例:

 # 消费者参数示例max.poll.records=500 fetch.min.bytes=1 fetch.max.wait.ms=100 enable.auto.commit=false

• 一度に複数のメッセージを取得してスループットを向上させるために、max.poll.records は 500 に設定されています。
• データが十分でない場合でもメッセージがすぐに取得されるように、fetch.min.bytes は 1 に設定されています。
• メッセージの取得の待機時間を制限するために、fetch.max.wait.ms は 100 ミリ秒に設定されています。
• enable.auto.commit は、メッセージの確認応答を正確に制御できるように、自動コミットのシフトを無効にします。

（２）ブローカーパラメータ

Kafka ブローカー パラメータを最適化すると、全体的なパフォーマンスが向上します。次に例を示します。

 # Kafka Broker 参数示例num.network.threads=3 num.io.threads=8 log.segment.bytes=1073741824 log.retention.check.interval.ms=300000

• num.network.threads と num.io.threads は、ハードウェア リソースを最大限に活用するために適切な値に設定されています。
• ディスク パフォーマンスを最大限に活用するために、log.segment.bytes は 1 GB に設定されています。
• ログ セグメントのクリーンアップ頻度を減らすために、log.retention.check.interval.ms は 300,000 ミリ秒に設定されています。

（３）トピックパラメータ

アプリケーションのニーズに合わせて各トピックのパラメータを最適化することも重要です。次に例を示します。

 # 创建Kafka 主题并设置参数示例kafka-topics.sh --create --topic my_topic --partitions 8 --replication-factor 2 --config cleanup.policy=compact

• --partitions 8 は並列処理を増やすためにパーティションの数を 8 に設定します。
• --replication-factor 2 信頼性を高めるためにレプリケーション係数を 2 に設定します。
• --config cleanup.policy=compact データ保持コストを削減するために、クリーンアップ ポリシーをコンパクト ポリシーに設定します。

これらのパラメータを適切に構成することで、Kafka 構成を効果的に最適化し、メッセージの遅延を減らしてパフォーマンスを向上させることができます。アプリケーションのニーズとハードウェア リソースに応じて調整してください。

2. 効率的なプロデューサーとコンシューマーを書く

最後に、レイテンシを削減するには、効率的な Kafka プロデューサー コードとコンシューマー コードを記述することが重要です。以下にベストプラクティスをいくつか示します。

（１）生産者のためのベストプラクティス

• 非同期送信を使用します。メッセージを 1 つずつ送信するのではなく、複数のメッセージを一括して送信します。これにより、ネットワーク通信の回数が減り、スループットが向上します。
• Kafka プロデューサーのバッファリング メカニズムを使用する: Kafka プロデューサーのバッファリング機能を最大限に活用して、ネットワーク通信の数を減らします。
• パーティション キーを使用する: 適切なパーティション キーを選択すると、データが異なるパーティションに均等に分散され、並列処理が向上します。

（２）消費者のベストプラクティス

• マルチスレッド消費を使用する: 複数のコンシューマー スレッドがメッセージを並列に処理できるようにします。これにより、処理能力が向上し、待ち時間が短縮されます。
• 消費者パラメータを調整する: 消費者パラメータを調整する。   fetch.min.bytes  そして スループットとレイテンシのバランスをとるためにfetch.max.wait.ms 。
• メッセージのバッチ処理を使用する: 処理のオーバーヘッドを削減するために、メッセージのバッチをまとめて処理します。

（３）データのシリアル化

データ転送とストレージのオーバーヘッドを削減するには、効率的なデータシリアル化形式を選択することが重要です。推奨される形式は次のとおりです。

• Avro : Apache Avro は、高度に圧縮され、高性能なデータシリアル化フレームワークです。大規模なデータ処理に適しています。
• プロトコル バッファ: Google プロトコル バッファ (ProtoBuf) は、優れたパフォーマンスとコンパクトなデータ表現を備えた軽量のバイナリ データ形式です。

4. Kafka メッセージのタイミング

メッセージのタイミングは、ほとんどのリアルタイム データ ストリーミング アプリケーションにとって中核的な要件です。このセクションでは、メッセージ タイミングの概念、それがなぜ重要なのか、そしてメッセージ タイミングを確保する方法について詳しく説明します。

1. メッセージのタイミングとは何ですか?

メッセージのタイミングとは、メッセージが送信された順序で受信され、処理されるという特性を指します。 Kafka では、各パーティション内のメッセージは順序付けられます。つまり、メッセージはプロデューサーによって送信された順序になります。ただし、複数のパーティションにまたがるメッセージは、順序を維持するために追加の作業が必要になります。

（１）メッセージのタイミングはなぜ重要なのか？

メッセージのタイミングは多くのアプリケーション、特に時間順にデータを処理する必要があるアプリケーションにとって重要です。メッセージのタイミングが重要となるアプリケーション領域をいくつか示します。

• 金融: 金融取引では、取引が発生した順序通りに処理されることを保証することが重要です。トランザクションのタイミングが失われると、不整合や誤ったトランザクションが発生する可能性があります。
• ログ記録: ログ記録および監視アプリケーションでは、イベントのタイミングは分析とトラブルシューティングにとって重要です。イベントの時間的な順序が失われると、混乱が生じ、データが不正確になる可能性があります。
• 電子商取引アプリケーション: オンライン ストアでの注文処理では、注文の混乱や不正確さを避けるために、注文の作成、支払い、出荷の手順が正しい順序で実行されるようにする必要があります。

2. メッセージのタイミングを確認する

分散システムでは、特に複数のパーティションにまたがる場合、メッセージのタイミングを確保することが困難になることがあります。メッセージのタイミングを確保するための戦略とベスト プラクティスをいくつか紹介します。

（１）パーティショニングとメッセージのソート

適切なパーティション分割戦略を使用してメッセージを順序付けし、関連するメッセージが同じパーティションに送信されるようにします。これにより、単一のパーティション内でメッセージの順序が維持されます。特定のキーでソートする必要があるメッセージの場合は、カスタム パーティショナーを使用してこれを実現できます。

適切なパーティション分割戦略を使用してメッセージを順序付ける方法のコード例を次に示します。

 // 自定义分区器，确保相关消息基于特定键被发送到同一个分区public class CustomPartitioner implements Partitioner { @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { // 在此处根据key 的某种规则计算分区编号// 例如，可以使用哈希函数或其他方法int numPartitions = cluster.partitionsForTopic(topic).size(); return Math.abs(key.hashCode()) % numPartitions; } @Override public void close() { // 可选的资源清理} @Override public void configure(Map<String, ?> configs) { // 可选的配置} }

（２）データの一貫性

プロデューサーから送信されたメッセージが正しいことを確認します。メッセージが正しい順序で Kafka に送信されるようにするには、メッセージのバッファリング、並べ替え、マージなど、アプリケーション レベルで実装する必要がある場合があります。

以下は、データの一貫性を確保する方法のコード例です。

 // 生产者端的消息排序ProducerRecord<String, String> record1 = new ProducerRecord<>("my-topic", "key1", "message1"); ProducerRecord<String, String> record2 = new ProducerRecord<>("my-topic", "key2", "message2"); // 发送消息producer.send(record1); producer.send(record2); // 消费者端保证消息按照键排序ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { // 处理消息，确保按照键的顺序进行}

（３）消費者の並列性

コンシューマー側では、適切なスレッドとパーティションの割り当てを使用して、メッセージが正しい順序で処理されるようにします。これには、コンシューマー スレッドの数を管理し、各スレッドが 1 つのパーティションのみを処理して順序どおりに処理されないようにすることが含まれる場合があります。

以下は、コンシューマーの並列処理を確実に行う方法のコード例です。

 // 创建具有多个消费者线程的Kafka 消费者Properties consumerProps = new Properties(); consumerProps.put("bootstrap.servers", "kafka-broker:9092"); consumerProps.put("group.id", "my-group"); consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 创建Kafka 消费者KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps); // 订阅主题consumer.subscribe(Collections.singletonList("my-topic")); // 创建多个消费者线程int numThreads = 3; for (int i = 0; i < numThreads; i++) { Runnable consumerThread = new ConsumerThread(consumer); new Thread(consumerThread).start(); }

V. 結論

この技術ブログでは、Kafka メッセージのレイテンシとタイミングの重要性と、メッセージのレイテンシを測定、監視、削減する方法について詳しく説明します。また、メッセージのタイミングの課題と、メッセージのタイミングを確保する方法についても議論しました。リアルタイム データ ストリーミング アプリケーションを構築する開発者にとって、これらの概念を深く理解することは非常に重要です。 Kafka を適切に構成し、ネットワークとハードウェアを最適化し、効率的なプロデューサーとコンシューマーのコードを記述し、メッセージのタイミングを維持することで、高性能で信頼性の高いデータ ストリーミング システムを構築できます。

アプリケーションが金融取引、監視、ログ記録、またはその他の分野のいずれであっても、これらの推奨事項とベスト プラクティスは、Kafka メッセージのレイテンシとタイミングの課題に適切に対処し、データの信頼性と一貫性を確保するのに役立ちます。