Java动态创建kafka并保持连接长期可用

一、kafka初始化

首先了解下kafka是怎么初始化的，看源码，就可以了解项目启动的时候，它就读取配置文件里面的kafka配置，然后初始化一个KafkaTemplate连接，然后接下来使用kafka推送或监听，就是这一次连接的。读取的主机和端口是在“bootstrap-servers”后面的那一串配置。后面就是一些生产者或者订阅者的一些配置，自己按照想要的连接进行配置就可。这里的重点不在配置文件怎么配置，不明白的朋友可以去查查。

这种方式是最简单且最常用的，但是有些领导就是要追求点刺激，让我们突破常规，要连接两个地址的kafka连接，并且同时可以使用，这时候我们就不能按照常规的配置，就得在代码中自己创建连接了。

二、自己创建kafka连接

既然是两个连接，那么这个连接就是固定的，那我们可以将连接放入配置文件中，然后从配置文件中先获取到连接的主机和端口。在配置文件中就两个String即可，然后在代码中获取。这时候我们写两个config来进行初始化连接，并且让他们别名不一样，这样就可以从随意使用任意一个连接了。

举个例子，其中的一个连接

package com.inphase.mjai.device.config;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

@EnableKafka
@Configuration
public class KafkaOneConfig {

    @Value("${spring.kafka.one.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public KafkaTemplate<String, String> kafkaOneTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaOneContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        ContainerProperties containerProperties = factory.getContainerProperties();
        containerProperties.setMissingTopicsFatal(false);
        containerProperties.setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    private ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    private Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    private Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }
}

根据以上代码，我们就可以初始化一个别名为KafkaOneConfig的kafka连接。就可以直接在代码中注入使用该连接就行。

但这时候领导又说，连接地址已经被写死了，如果给你三个连接你是不是得改代码，这样太不灵活了，搞一个动态的连接，随时更改连接地址那种。

三、动态创建kafka

同自己创建kafka连接一样，这里就是获取连接主机和端口不一样罢了。可以将kafka的连接存入数据库，然后数据来的时候根据要推送的kafka进行连接。这里只举例创建一个生产者

    private KafkaTemplate<String, String> kafkaDynamicTemplate(String ipAndPort) {
        return new KafkaTemplate<>(producerFactory(ipAndPort));
    }

    private ProducerFactory<String, String> producerFactory(String ipAndPort) {
        return new DefaultKafkaProducerFactory<>(producerConfigs(ipAndPort));
    }

    private Map<String, Object> producerConfigs(String ipAndPort) {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ipAndPort);
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

到这里就可以随意更改ipAndPort来创建不同的连接了。

但是这时候出现了一个问题，就是每次有数据过来都要重新连接一次吗？怎样将连接保存下来，第二次需要推送的时候继续用呢。

想了许久，我就想到，kafka初始化的时候好像就是注入到了SpringBean容器中，然后要使用的时候就从当中取出即可。那我们就可以将连接创建了过后，放入容器中，并且给它一个独特的别名，就可以节约资源不用每次都创建连接了。

先导入一个hutool包

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.6.2</version>
            <scope>compile</scope>
        </dependency>

然后使用其中的SpringUtil

设定别名为ipAndPort的bean，bean类为创建的连接，kafkaDynamicTemplate方法就是上面创建连接的方法。

springUtil.registerBean(ipAndPort,this.kafkaDynamicTemplate(ipAndPort));

这时候就在容器中有了一个kafka连接对象，只要项目没有重新启动，这个连接都是可以使用的。

但是怎么使用呢？

就从容器中通过别名获取即可

kafkaTemplate = springUtil.getBean(ipAndPort,KafkaTemplate.class);

这时候就可以正常使用kafkaTemplate进行数据推送了。

没什么难点，但让我第一次手动操作了SpringBean容器，也是种成长。

有什么还可以优化的地方还望各位大佬指点。