将主题列表从应用程序 yml 传递给 KafkaListener

Question

我有以下application.yml：

service:
  kafka:
    groupId: 345
    consumer: 
      topics: 
        -
          name: response    
    producer: 
      topics: 
        -
          name: request1 
          num-partitions: 5
          replication-factor: 1 
        -
          name: request2 
          num-partitions: 3
          replication-factor: 1  

如何使用 spel 访问主题名称列表以传递给 KafkaListener 注释？

@KafkaListener(topics = "#{'${service.kafka.consumer.topics.name}'}", containerFactory = "kafkaListenerContainerFactory")
public void receive(String payload, @Header(KafkaHeaders.RECEIVED_TOPIC)String topic) {

Answer 1

使用配置属性和集合投影...

@ConfigurationProperties("service.kafka.producer")
@Component
public class ConfigProps {

    List<Topic> topics = new ArrayList<>();

    public List<Topic> getTopics() {
        return this.topics;
    }

    public void setTopics(List<Topic> topics) {
        this.topics = topics;
    }

    @Override
    public String toString() {
        return "ConfigProps [topics=" + this.topics + "]";
    }

    public static class Topic {

        private String name;

        private int numPartitions;

        private short replicationFactor;

        public String getName() {
            return this.name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public int getNumPartitions() {
            return this.numPartitions;
        }

        public void setNumPartitions(int numPartitions) {
            this.numPartitions = numPartitions;
        }

        public short getReplicationFactor() {
            return this.replicationFactor;
        }

        public void setReplicationFactor(short replicationFactor) {
            this.replicationFactor = replicationFactor;
        }

        @Override
        public String toString() {
            return "Topic [name=" + this.name + ", numPartitions=" + this.numPartitions + ", replicationFactor="
                    + this.replicationFactor + "]";
        }

    }

}

和

@SpringBootApplication
public class So52741016Application {

    public static void main(String[] args) {
        SpringApplication.run(So52741016Application.class, args);
    }

    @KafkaListener(groupId = "${service.kafka.groupId}", topics = "#{configProps.topics.![name]}")
    public void listener(String in) {

    }

    @Bean
    public SmartLifecycle createTopics(KafkaAdmin admin, ConfigProps props) {
        return new SmartLifecycle() {

            @Override
            public int getPhase() {
                return Integer.MIN_VALUE;
            }

            @Override
            public void stop() {
            }

            @Override
            public void start() {
                try (AdminClient client = AdminClient.create(admin.getConfig())) {
                    CreateTopicsResult createTopics = client.createTopics(props.topics.stream()
                        .map(t -> new NewTopic(t.getName(), t.getNumPartitions(), t.getReplicationFactor()))
                        .collect(Collectors.toList()));
                    createTopics.all().get();
                }
                catch (Exception e) {
//                  e.printStackTrace();
                }
            }

            @Override
            public boolean isRunning() {
                return false;
            }

            @Override
            public void stop(Runnable callback) {
            }

            @Override
            public boolean isAutoStartup() {
                return true;
            }
        };
    }

}

和

2018-10-10 11:20:25.813 INFO 14979 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer：分配的分区：[request1-4，request2-0，request1-0，request2- 1、request1-1、request2-2、request1-2、request1-3]

当然，这只是生产者话题，不过你可以这样处理。