を使用するためのプロデューサーイベント通知フレームワークの構成 Hermes メッセージングサービス

Hermes メッセージングサービスインスタンスで を使用している場合は、Kafka メッセージバス用のプロデューサーイベント通知フレームワークを構成する必要があります。

始める前に

必要なロール：管理者

を Hermes メッセージングサービス アクティブ化する必要があります。詳細については、「Hermes Messaging Service activation」を参照してください。

次のコード スニペットは、を使用してHermes メッセージングサービストピックにメッセージを送信するために必要な要素を構成するために呼び出す必要がある HermesEventPublisher メソッドを示しています。その後、 メッセージを送信するために ProducerV2 API が呼び出されます。

publishEventToHermesKafka: function(tmfEventPayload) {
  if (gs.nil(tmfEventPayload)) {
    return;
  }
  this.publishMessageThroughProducerV2API(tmfEventPayload)

},

publishMessageThroughProducerV2API: function(message) {
  if (gs.nil(message)) return;
  var comptableTopicSysIds = this.fetchTopicsForMessage(message);
  if (gs.nil(comptableTopicSysIds)) return;

  for (var i = 0; i < comptableTopicSysIds.length; i++) {
    var partitionKey = this.fetchPartionKey(message);
    var headers = this.fetchMessageHeaders(message)
    try {
      var producer = new sn_ih_kafka.ProducerV2();
      producer.send(comptableTopicSysIds[i], partitionKey, JSON.stringify(message), false, headers);
    } catch (ex) {
      this._logger.logErr("Publishing message to Hermes Kafka Failed, error is: " + ex);
    }
  }
},

手順

1. Hermes メッセージングサービス への安全な接続をセットアップします。
   詳細については、「Set up a secure connection to the Hermes Messaging Service」を参照してください。
2. メッセージを Hermes メッセージングサービスに公開するときに、メソッドのHermesEventPublisherOOB - fetchTopicsForMessage(オブジェクト message)デフォルト機能をカスタムトピックピッカーロジックで上書きします。
   HermesEventPublisher - fetchTopicsForMessage() メソッドでは、指定されたメッセージの送信先の Kafka トピックを決定するロジックを指定する必要があります。次に、それらのトピックレコードのsys_idsをアレイで返します。

   メッセージペイロードの属性に基づいてトピックを選択することも、実装で 1 つのトピックのみを使用する場合はメッセージごとに同じトピックを返すこともできます。

   次の例は、渡された各メッセージについて同じトピックレコードsys_idを返す方法を示しています。

   // Return the same topic for all messages if your 
    // implementation only supports one kind of topic
   
   fetchTopicsForMessage: function(message) { 
   // Return the same sys_id for all passed messages which
   // implies that all of the messages are published to the same topic.
   return ['91564fc087f8351063151f473cbb35f8'];
   },

   次の例は、メッセージのイベントタイプに基づいてメッセージの送信先のトピックを識別する方法を示しています。

   // Fetch the topic based on the message content. 
   // This could be based on the message domain type or the message event type.
   
   fetchTopicsForMessage: function(message) { 
   // Implement to return topic based on a message content
   // Implementor may decide a topic based on 'eventType' or any other attribute.
   
   var eventType = message.eventType;
   
   var topicSysIdsArr = this.fetchTopicBasedOnEventType(eventType);
   return topicSysIdsArr
   
   },

3. メソッドの HermesEventPublisherOOB - fetchPartionKey(オブジェクト message) 既定の機能をカスタム ロジックでオーバーライドして、メッセージを Hermes メッセージングサービス.
   次の一覧には、返されるパーティション キーの種類と、それに関連する利点が含まれています。

   • 自動生成された一意の ID。このタイプのパーティションキーにより、トピックパーティション間でメッセージが均等に発行されます。これはデフォルトの実装です。
   • メッセージ・イベント・タイプ (またはその他のメッセージ属性) に基づくユニーク・キー。この種類のパーティション キーは、そのイベントの種類 (または他のメッセージ属性) のメッセージの順序付けを保証します。
   • 同じパーティション キー。これにより、メッセージは同じパーティションに送信されます。このタイプのパーティションキーにより、Kafka コンシューマーグループレベルでのメッセージの厳密な順序付けが保証されます。

   次の例は、各パーティション キーのランダム ID を返す方法を示しています。

   fetchPartionKey: function(message) { 
   // Implement it for custom partition key 
   // Random message keys ensure better distribution of message across partitions. 
   // Same message keys ensure that message always goes to the same partition.   
   // This ensures strict sequencing of messages at the Kafka consumer group level. 
   return gs.generateGUID(); 
   },

   次の例は、メッセージ内のイベントの種類に基づいてパーティション キーを返す方法を示しています。

   // Implementation where the partition key is decided based on the event type.
   fetchPartionKey: function(message) { 
   
   var eventType = message.eventType;
   var partionKey = fetchPartionKeyForMessage(eventType);
   return partionKey;
   },

4. メソッドの HermesEventPublisherOOB - fetchMessageHeaders(オブジェクト message) デフォルト機能をカスタム ロジックで上書きして、指定されたメッセージのヘッダーをフェッチしてから、 を介してメッセージを Hermes メッセージングサービス公開します。
   これらのヘッダーの詳細については、 Apache Kafka のドキュメントを参照してください。

   次の例は、メッセージ内のイベントタイプに基づいてヘッダーを返す方法を示しています。

   fetchMessageHeaders: function(message) { 
   // Returned headers are based on the event type passed in the message object. 
     var eventType = message.eventType;
     var headerObject = fetchHeadersBasedOnEventType(eventType);
     return headerObject;
   },