メインコンテンツへ移動

ワークフローの並列度の調整

ワークフローの処理の仕組み​

ワークフローは Kafka を使用してタスクを非同期に処理します。イベントが発生すると、システムは対応する Kafka Topic にメッセージを書き込み、ワークフローのコンシューマーサービスが読み取ってフローを実行します。実行中は、ワークシートなどの業務サービスの呼び出しや、データベースの読み書きも行います。

タスクが集中的に発生すると、Kafka がメッセージを一時的に保持し、業務サービスとデータベースへの瞬間的な負荷を軽減します。メッセージの滞留数が継続的に減少し、業務で許容される時間内に処理を完了できる場合は、並列度を変更する必要はありません。

ワークフローのコンシューマーサービスは、md-workflow-consumer コンシューマーグループで次の Topic を読み取ります。

Topic用途
WorkFlowメインフローの実行
WorkFlow-Processサブフローの実行
WorkFlow-Routerワークフローの低速キュー
WorkFlow-Batchワークフローの一括実行
WorkFlow-Buttonボタンで起動したワークフローの実行
WorkSheet行レコードがワークフローの起動条件を満たすか確認
WorkSheet-Batch行レコードがワークフローの起動条件を満たすか一括確認
WorkSheet-Router低速キューで行レコードがワークフローの起動条件を満たすか確認

Kafka Topic は1つ以上のパーティションで構成されます。同じコンシューマーグループ内では、1つのパーティションを同時に割り当てられるコンシューマーは1つだけです。ワークフローで並列に処理できるパーティション数は、Topic のパーティション数、各コンシューマーサービスのスレッド数、インスタンス数で決まります。

パーティション数とコンシューマースレッドの総数を対応させる必要があります。Topic が10パーティションの場合、20スレッドを設定しても、パーティションを割り当てられるのは10スレッドだけです。Topic が20パーティションでもスレッドが5つしかない場合は、同時に処理できるのは5パーティションだけです。

クラスターモードでは、コンシューマースレッドの総数は、各インスタンスのスレッド数とインスタンス数の積になります。実際に処理に参加するスレッド数は、Topic のパーティション数を超えません。

Topic のパーティション数を増やしても、既存のメッセージは再分配されません。滞留中のメッセージは元のパーティションで処理を待ちます。追加したパーティションには、拡張後に書き込まれた新しいメッセージだけが分配ポリシーに従って入ります。そのため、拡張直後に既存パーティションの LAG が減るわけではありません。

調整方法​

ワークフローの並列度を上げる前に、データベースがより高い同時実行負荷に耐えられるか確認してください。CPU やディスク I/O の使用率が高い場合や、スロークエリが多い場合は、Kafka のパーティション、コンシューマースレッド、サービスインスタンスを増やすと、データベースの負荷がさらに増加します。

MongoDB のスロークエリ最適化を参照し、スロークエリ、インデックス、クエリ条件、サーバーリソースを確認してください。データベースのボトルネックを解消してから、ワークフローの並列度を調整します。

単一サーバーモード​

単一サーバーモードでは、ワークフローのコンシューマープロセスを1つ使用します。Kafka Topic のパーティション数とコンシューマースレッド数を増やすことで、並列処理能力を高められます。

通常、すべてのサービスが同じサーバーにデプロイされます。調整前に、サーバーの CPU、メモリ、ディスク I/O を確認してください。使用率が上限に近い場合は、先にサーバーリソースを増強します。

対象 Topic のパーティション数とワークフローのコンシューマースレッド数は、いずれもデフォルトで10です。調整時は20～30を推奨します。この例では、20パーティション、20コンシューマースレッドに設定します。

  1. ストレージコンポーネントのコンテナに入ります。

    docker exec -it $(docker ps | grep mingdaoyun-sc | awk '{print $1}') bash

  2. 各 Topic のパーティション数とメッセージの滞留数を確認します。

    /usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group md-workflow-consumer

    コマンド出力の PARTITION はパーティション番号、LAG はそのパーティションのメッセージ滞留数です。

  3. 各 Topic の滞留状況に応じて、パーティション数を増やします。この例では20に設定します。

    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkFlow
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkFlow-Batch
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkFlow-Button
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkFlow-Process
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkFlow-Router
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkSheet
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkSheet-Batch
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 20 --topic WorkSheet-Router

    Kafka Topic のパーティション数は増やすことだけが可能で、減らすことはできません。滞留状況に応じて段階的に調整してください。

  4. docker-compose.yaml に、ワークフローのコンシューマースレッド数を指定する環境変数を設定します。

    ENV_WORKFLOW_CONSUMER_THREADS: '20'
    ENV_WORKFLOW_ROUTER_CONSUMER_THREADS: '20'

    ENV_WORKFLOW_CONSUMER_THREADS は通常のワークフローキューのスレッド数、ENV_WORKFLOW_ROUTER_CONSUMER_THREADS は WorkFlow-Router 低速キューのスレッド数を制御します。通常は、対応する Topic のパーティション数と同じ値に設定します。

  5. ストレージコンポーネントのコンテナを終了し、管理ツールのインストールディレクトリに戻って HAP サービスを再起動し、設定を反映します。

    bash ./service.sh restartall

クラスターモード​

クラスターモードでは、HAP マイクロサービスを Kubernetes 上で実行し、Kafka を独立してデプロイします。Topic のパーティション数とコンシューマーサービスのレプリカ数を増やすことで、並列処理能力を高められます。

調整前に、クラスターノードの CPU、メモリ、ディスク I/O、および Kubernetes Pod の CPU・メモリ使用率を確認してください。使用率が上限に近い場合は、リソースのボトルネックを解消してから並列度を上げます。

workflowconsumer と workflowrouterconsumer の各 Pod には、デフォルトで10のコンシューマースレッドが設定されています。Topic のパーティション数は30～120を推奨します。この例では40パーティションとし、2つのコンシューマーサービスにそれぞれ4つの Pod を設定します。

  1. Kafka サーバーにログインし、各 Topic のパーティション数とメッセージの滞留数を確認します。

    /usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group md-workflow-consumer

    コマンド出力の PARTITION はパーティション番号、LAG はそのパーティションのメッセージ滞留数です。

  2. 各 Topic の滞留状況に応じて、パーティション数を増やします。この例では40に設定します。

    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkFlow
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkFlow-Batch
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkFlow-Button
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkFlow-Process
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkFlow-Router
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkSheet
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkSheet-Batch
    /usr/local/kafka/bin/kafka-topics.sh --alter --bootstrap-server 127.0.0.1:9092 --partitions 40 --topic WorkSheet-Router

    Kafka Topic のパーティション数は増やすことだけが可能で、減らすことはできません。滞留状況に応じて段階的に調整してください。

  3. ワークフローのコンシューマーサービスを4つの Pod に拡張します。

    kubectl scale deployment workflowconsumer --replicas=4
    kubectl scale deployment workflowrouterconsumer --replicas=4

    HAP サービスの再起動時に元のレプリカ数に戻らないよう、service.yaml でも workflowconsumer と workflowrouterconsumer のレプリカ数を4に変更してください。

    例：

    apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: workflowconsumer # <-- サービス名
      namespace: default
    spec:
      replicas: 4 # <-- レプリカ数

    workflowconsumer は通常のワークフローキュー、workflowrouterconsumer は WorkFlow-Router 低速キューを処理します。各 Pod のスレッド数を変更する場合は、Topic のパーティション数に応じてレプリカ数を再計算してください。

  4. ワークフローの同時処理数を増やした後は、Kubernetes クラスターの監視で、ワークフローが呼び出すマイクロサービスの Pod の CPU 使用率を確認します。特定のサービスの CPU 使用率が継続的に高い場合は、そのサービスのレプリカ数を増やします。この例では、worksheetonlyworkflow、worksheetonlyworkflowr、command を4つの Pod に拡張します。

    kubectl scale deployment worksheetonlyworkflow --replicas=4
    kubectl scale deployment worksheetonlyworkflowr --replicas=4
    kubectl scale deployment command --replicas=4

    HAP サービスの再起動時に元のレプリカ数に戻らないよう、service.yaml でも上記サービスのレプリカ数を4に変更してください。