は、実行時にカフカの消費者のトピックに動的な値を渡すメッセージ駆動チャネルアダプタ用

Question

要件は、我々はここでそれを達成することができ、サーバ .Howを再起動せずにトピックランタイムで以下の属性を設定することです。 現在、プロパティファイルから値を読み取っていますが、ここでは変更を反映するためにサーバーを再起動する必要があります。

例： sample.properties = topic1

topic.list、topic2

（展開ディレクトリ内）と、サーバの再起動なしで、将来的にtopic3から消費したいと思います。

注：トピックが最終変数であることがわかります。

は、ファイルシステムパス（デプロイメントディレクトリの外側）からキー（topic.list）を読み取ろうとしましたが、運がうまくいきませんでした。

提案があります。

あなたが動的にオンデマンドで追加トピック用のアダプタを追加するには、Java DSLを使用することができます

<int-kafka:message-driven-channel-adapter 

       id="inAdapter" 
       channel="fromKafka" 
       connection-factory="connectionFactory" 
       key-decoder="kafkaKeyDecoder" 
       payload-decoder="kafkaDecoder"        
       topics="${topic.list}" 
       offset-manager="offsetManager"/> 

Answer 1

...

@Autowired 
private IntegrationFlowContext flowContext; 

public void addAnotherListenerForTopics(String... topics) { 
    IntegrationFlow flow = 
     IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(consumerFactory(), topics)) 
      .channel("fromKafka") 
      .get(); 
    this.flowContext.registration(flow).register(); 
} 

と

bean.addAnotherListenerForTopics("added.new"); 

ポンポン：

<dependency> 
    <groupId>org.springframework.integration</groupId> 
    <artifactId>spring-integration-java-dsl</artifactId> 
    <version>1.2.1.RELEASE</version> 
</dependency> 

注意ブローカパーティションを次のように使用している場合新しいコンテナは、既存の割り当ての取り消しを避けるために、異なるグループIDを必要とします。