Spring Boot和Kafka实战自定义复杂配置示例

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
    - <font>"2181:2181"</font><font>
  kafka:
    image: wurstmeister/kafka
    ports:
    - </font><font>"9092:9092"</font><font>
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
</font>

spring:
  kafka:
    consumer:
      group-id: tpd-loggers
      auto-offset-reset: earliest
    # change <b>this</b> property <b>if</b> you are using your own
    # Kafka cluster or your Docker IP is different
    bootstrap-servers: localhost:9092

tpd:
  topic-name: advice-topic
  messages-per-request: 10

<b>import</b> com.fasterxml.jackson.annotation.JsonProperty;
 
<b>public</b> <b>class</b> PracticalAdvice {
    <b>private</b> <b>final</b> String message;
    <b>private</b> <b>final</b> <b>int</b> identifier;
 
    <b>public</b> PracticalAdvice(@JsonProperty(<font>"message"</font><font>) <b>final</b> String message,
                           @JsonProperty(</font><font>"identifier"</font><font>) <b>final</b> <b>int</b> identifier) {
        <b>this</b>.message = message;
        <b>this</b>.identifier = identifier;
    }
 
    <b>public</b> String getMessage() {
        <b>return</b> message;
    }
 
    <b>public</b> <b>int</b> getIdentifier() {
        <b>return</b> identifier;
    }
 
    @Override
    <b>public</b> String toString() {
        <b>return</b> </font><font>"PracticalAdvice::toString() {"</font><font> +
                </font><font>"message='"</font><font> + message + '\'' +
                </font><font>", identifier="</font><font> + identifier +
                '}';
    }
}
</font>

@SpringBootApplication
<b>public</b> <b>class</b> KafkaExampleApplication {
 
    <b>public</b> <b>static</b> <b>void</b> main(String[] args) {
        SpringApplication.run(KafkaExampleApplication.<b>class</b>, args);
    }
 
    @Autowired
    <b>private</b> KafkaProperties kafkaProperties;
 
    @Value(<font>"${tpd.topic-name}"</font><font>)
    <b>private</b> String topicName;
 
    </font><font><i>// Producer configuration</i></font><font>
 
    @Bean
    <b>public</b> Map<String, Object> producerConfigs() {
        Map<String, Object> props =
                <b>new</b> HashMap<>(kafkaProperties.buildProducerProperties());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.<b>class</b>);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                JsonSerializer.<b>class</b>);
        <b>return</b> props;
    }
 
    @Bean
    <b>public</b> ProducerFactory<String, Object> producerFactory() {
        <b>return</b> <b>new</b> DefaultKafkaProducerFactory<>(producerConfigs());
    }
 
    @Bean
    <b>public</b> KafkaTemplate<String, Object> kafkaTemplate() {
        <b>return</b> <b>new</b> KafkaTemplate<>(producerFactory());
    }
 
    @Bean
    <b>public</b> NewTopic adviceTopic() {
        <b>return</b> <b>new</b> NewTopic(topicName, 3, (<b>short</b>) 1);
    }
}
</font>

@RestController
<b>public</b> <b>class</b> HelloKafkaController {
 
    <b>private</b> <b>static</b> <b>final</b> Logger logger =
            LoggerFactory.getLogger(HelloKafkaController.<b>class</b>);
 
    <b>private</b> <b>final</b> KafkaTemplate<String, Object> template;
    <b>private</b> <b>final</b> String topicName;
    <b>private</b> <b>final</b> <b>int</b> messagesPerRequest;
    <b>private</b> CountDownLatch latch;
 
    <b>public</b> HelloKafkaController(
            <b>final</b> KafkaTemplate<String, Object> template,
            @Value(<font>"${tpd.topic-name}"</font><font>) <b>final</b> String topicName,
            @Value(</font><font>"${tpd.messages-per-request}"</font><font>) <b>final</b> <b>int</b> messagesPerRequest) {
        <b>this</b>.template = template;
        <b>this</b>.topicName = topicName;
        <b>this</b>.messagesPerRequest = messagesPerRequest;
    }
 
    @GetMapping(</font><font>"/hello"</font><font>)
    <b>public</b> String hello() throws Exception {
        latch = <b>new</b> CountDownLatch(messagesPerRequest);
        IntStream.range(0, messagesPerRequest)
                .forEach(i -> <b>this</b>.template.send(topicName, String.valueOf(i),
                        <b>new</b> PracticalAdvice(</font><font>"A Practical Advice"</font><font>, i))
                );
        latch.await(60, TimeUnit.SECONDS);
        logger.info(</font><font>"All messages received"</font><font>);
        <b>return</b> </font><font>"Hello Kafka!"</font><font>;
    }
}
</font>

@SpringBootApplication
<b>public</b> <b>class</b> KafkaExampleApplication {
 
    <b>public</b> <b>static</b> <b>void</b> main(String[] args) {
        SpringApplication.run(KafkaExampleApplication.<b>class</b>, args);
    }
 
    @Autowired
    <b>private</b> KafkaProperties kafkaProperties;
 
    @Value(<font>"${tpd.topic-name}"</font><font>)
    <b>private</b> String topicName;
 
    </font><font><i>// Producer configuration</i></font><font>
    </font><font><i>// omitted...</i></font><font>
 
    </font><font><i>// Consumer configuration</i></font><font>
 
    </font><font><i>// If you only need one kind of deserialization, you only need to set the</i></font><font>
    </font><font><i>// Consumer configuration properties. Uncomment this and remove all others below.</i></font><font>
</font><font><i>//    @Bean</i></font><font>
</font><font><i>//    public Map<String, Object> consumerConfigs() {</i></font><font>
</font><font><i>//        Map<String, Object> props = new HashMap<>(</i></font><font>
</font><font><i>//                kafkaProperties.buildConsumerProperties()</i></font><font>
</font><font><i>//        );</i></font><font>
</font><font><i>//        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,</i></font><font>
</font><font><i>//                StringDeserializer.class);</i></font><font>
</font><font><i>//        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,</i></font><font>
</font><font><i>//                JsonDeserializer.class);</i></font><font>
</font><font><i>//        props.put(ConsumerConfig.GROUP_ID_CONFIG,</i></font><font>
</font><font><i>//                "tpd-loggers");</i></font><font>
</font><font><i>//</i></font><font>
</font><font><i>//        return props;</i></font><font>
</font><font><i>//    }</i></font><font>
 
    @Bean
    <b>public</b> ConsumerFactory<String, Object> consumerFactory() {
        <b>final</b> JsonDeserializer<Object> jsonDeserializer = <b>new</b> JsonDeserializer<>();
        jsonDeserializer.addTrustedPackages(</font><font>"*"</font><font>);
        <b>return</b> <b>new</b> DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), <b>new</b> StringDeserializer(), jsonDeserializer
        );
    }
 
    @Bean
    <b>public</b> ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                <b>new</b> ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
 
        <b>return</b> factory;
    }
 
    </font><font><i>// String Consumer Configuration</i></font><font>
 
    @Bean
    <b>public</b> ConsumerFactory<String, String> stringConsumerFactory() {
        <b>return</b> <b>new</b> DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), <b>new</b> StringDeserializer(), <b>new</b> StringDeserializer()
        );
    }
 
    @Bean
    <b>public</b> ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerStringContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                <b>new</b> ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(stringConsumerFactory());
 
        <b>return</b> factory;
    }
 
    </font><font><i>// Byte Array Consumer Configuration</i></font><font>
 
    @Bean
    <b>public</b> ConsumerFactory<String, byte[]> byteArrayConsumerFactory() {
        <b>return</b> <b>new</b> DefaultKafkaConsumerFactory<>(
                kafkaProperties.buildConsumerProperties(), <b>new</b> StringDeserializer(), <b>new</b> ByteArrayDeserializer()
        );
    }
 
    @Bean
    <b>public</b> ConcurrentKafkaListenerContainerFactory<String, byte[]> kafkaListenerByteArrayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, byte[]> factory =
                <b>new</b> ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(byteArrayConsumerFactory());
        <b>return</b> factory;
    }
}
</font>