Kafka メッセージトリガーとスクリプトコンシューマーを使用してメッセージを消費する

フローデザイナー の Kafka メッセージトリガーとスクリプトコンシューマーを使用して、Hermes からメッセージを取得します。

始める前に

• 必要なロール： integration_hub_admin
• このドキュメントでは、Hermes への接続を介してメッセージを消費する方法について説明します。接続の設定とテストについては、「Set up a secure connection to the Hermes Messaging Service」および「ADCv2 ゲートウェイを介した Hermes への接続のテスト」を参照してください。
• このページの前半では、フローデザイナー で Kafka メッセージトリガーを使用してメッセージを消費する方法を説明します。トリガーの詳細については、「Create a flow with a Kafka Message trigger」を参照してください。フローデザイナー およびフローの構成の詳細については、「Flow Designer」を参照してください。
• このページの後半では、スクリプトコンシューマーを使用してメッセージを消費する方法を説明します。詳細については、「スクリプトコンシューマーの構成」を参照してください。

手順

1. メッセージを消費するトピックを作成します。
   1. 次のコマンドを使用して、Hermes にトピックを作成します。

      bin/kafka-topics.sh --create --command-config config/bootcamp.properties  --bootstrap-server <instance name>.service-now.com:4000,<instance name>.service-now.com:4001,<instance name>.service-now.com:4002, <instance name>.service-now.com:4003 --topic snc.<instance name>.topic1

      必ず <instance name> をインスタンス名で置き換えてください。
   2. トピックを表示するには、インスタンスにログインし、次の場所に移動します。 すべて > 統合ハブ > ストリームコネクト > トピック.
      注:

      トピックの作成には約 10 分かかります。
2. トピックからメッセージを消費するフロートリガーを作成します。
   1. 移動先 すべて > プロセス自動化 > フローデザイナー.
   2. 選択 新規作成 > フロー.
   3. [フロープロパティ] フォームで、フローの名前を入力し、[送信] を入力します。
      この例では、Topic1 Consumer という名前を使用しています。
   4. トリガーセクションで [トリガーを追加] を選択し、トリガータイプとして [Kafka メッセージ] を選択します。

      

   5. Topic1 トピックを選択します。

      

      虫眼鏡アイコンを使用すると、利用可能なトピックのリストを表示できます。

   6. フローの [アクション] セクションで、次の場所に移動します。 アクション、フロー ロジック、またはサブフローを追加 > フローロジック > 各.
   7. [メッセージ] データピルを [アイテム] フィールドにドラッグし、[完了]を選択します。
   8. フローの [アクション] セクションで、次の場所に移動します。 アクション、フロー ロジック、またはサブフローを追加 > アクション > ServiceNow コア > ログ.
   9. [ペイロード] データピルを [メッセージ] フィールドにドラッグし、[完了]を選択します。

      

   10. フロートリガーを [保存] します。
   11. 設定 [sys_flow_execution_setting] テーブルに移動して、このフローのレポートを有効にします。
   12. 新しいエントリを作成し、上記で作成したフローへのレポートを有効にします。

       

   13. フローに戻り、[アクティブ化 (Activate)] を選択します。
   14. トピックのサブスクリプションを表示するには、次に移動します すべて > 統合ハブ > ストリームコネクト > サブスクリプション.

       

3. Kafka コマンドラインツールを使用して、トピックにメッセージを公開します。
   1. 次のコマンドを実行して、topic にメッセージを公開します。

      bin/kafka-console-producer.sh --topic snc.<instance name>.topic1  --producer.config  config/bootcamp.properties  --bootstrap-server <instance name>.service-now.com:4000,<instance name>.service-now.com:4001,<instance name>.service-now.com:4002,<instance name>.service-now.com:4003

      必ず <instance name> をインスタンス名で置き換えてください。

      このコマンドはメッセージの入力を促すものです。サンプルメッセージをいくつか追加します。この例では、次のメッセージを使用します。

      • テストメッセージ 1
      • テストメッセージ 2
      • テストメッセージ 3

      1 分ほどでメッセージを受信するはずです。
   2. フローに戻り、[実行]を選択します。

      

   3. フロー実行を開きます。

      

   4. このサブスクリプションの統計情報を表示するには、次の場所に移動します。 すべて > 統合ハブ > ストリームコネクト > サブスクリプションをクリックし、サブスクリプションを開いて、統計ビューに移動します。

      

4. トピックからメッセージを消費するフローコンシューマーを作成します。
   1. 移動先 統合ハブ > コンシューマー > スクリプトコンシューマー.
   2. [新規] を選択します。
   3. [名前] フィールドに、スクリプトコンシューマーの名前を入力します。
      この例では、Topic1 Consumer を使用しています。
   4. スクリプトボックスに次のロジックを追加して、メッセージをログに記録するスクリプトを設定します。

      for (var i = 0; i < messages.length; i++) {
            gs.info('Received message ' + JSON.stringify(messages[i].message));
       }
      

      

   5. フォームを保存します。
   6. [関連リンク] セクションで、[新規] を選択して、メッセージを消費する Kafka ストリームを追加します。
   7. [Kafka ストリーム] フォームで、次のフィールド値を設定します。
      • [名前] を「Topic1 Stream」を設定します。
      • [トピック] を前に作成したトピック topic1 に設定します。
      • [最大同時実行] を「1」に設定します。
      • [トピックに保存されているすべてのメッセージ (All the messages stored in the topic)] に[消費を開始 (Start Consuming)] を設定します。このオプションを使用すると、このトピックに既に公開されているメッセージを読むことができます。
      • [メッセージ処理 (Message handling)] を [自動的に最適化 (Automatically optimize)] に設定します。

      

   8. [関連リンク] の下で [アクティブ化] リンクを選択します。
      アクティブ化されたサブスクリプションを確認できるはずです。
   9. メッセージを表示するには、次の場所に移動します。 すべて > システムログ > システムログ をクリックし、[受信済みメッセージ] で始まるメッセージをフィルター処理します。

      

   10. 1 分後にすべてのメッセージが表示されるはずです。

次のタスク

ETL および変換マップコンシューマーを使用したデータのインポート。