springboot使用kafka推送数据到服务端的操作方法带认证_Java

遇到的问题

在实际开发过程中，因为推送数据需要用到kafka，为了比较方便与后续其他需求需要使用kafka，所以开发的过程中是设想能写一个工具类，方便后续的使用，但是，测试不带认证的kafka服务端的时候，发送是正常的，但是实际情况是，对方的服务器需要认证，导致遇到连不上对方服务，推送失败的问题，需要找对方确认对方的认证配置信息。并且在查询怎么处理的时候，验证也出现了比较奇葩的情况，那本次文章就简单写遇到的问题和验证结果。

碰到的天坑

1.度的时候，总是说引入了配置文件就完事，其他的配置不需要再配sasl.jaas.config，但是我实际测试下来不行啊。还是得引入配置文件+配置参数设置。
2.第二个基于未来思考的问题是，这个引入的配置文件内容，是否可以多个。据度来度去的结果说是可以的，按顺序会去逐一匹配直到匹配成功，那就是下面配置的kafkaclient配置块是可以多个，就名称不一样就行。但是碍于条件限制，没试过。😀

处理步骤

加载配置
度了很多文章，都提到了需要在服务启动的时候引入认证配置文件，设置属性 java.security.auth.login.config，度了一下是有两种方式。
a). 在服务启动的时候用参数引入，命令如下：
java -djava.security.auth.login.config=(具体地址自己填，因为我配置文件是跟test.jar同级目录所以无前缀)kafka_jaas_config.config -jar test.jar
b). 在代码中去引入，如：（这方式我是没试的）
system.setproperty(“java.security.auth.login.config”, “kafka_jaas_config.config”);

c). 配置文件的内容（注意password后面那个该死的分号是要的，配置参数设置的时候也是要的）

kafkaclient {  
    org.apache.kafka.common.security.scram.scramloginmodule required  
    username="user"  
    password="123";  
};

设置认证参数（如果有的话）
类似要填的参数是：
security.protocol，sasl.mechanism，sasl.jaas.config（，sasl.username和sasl.password，这两个我看是高版本直接配置据说能认证，不需要sasl.jaas.config，但是低版本是需要sasl.jaas.config，所以建议是可以都配上）
such as :

props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.scramloginmodule required username=\"user\" password=\"123\";");

工具类

package platform.cars.utils;
import org.apache.kafka.clients.producer.producerconfig;
import org.apache.kafka.common.serialization.stringserializer;
import org.springframework.context.annotation.configuration;
import org.springframework.kafka.core.defaultkafkaproducerfactory;
import org.springframework.kafka.core.kafkatemplate;
import org.springframework.kafka.core.producerfactory;
import org.springframework.retry.annotation.backoff;
import org.springframework.retry.annotation.retryable;
import javax.annotation.predestroy;
import java.util.hashmap;
import java.util.map;
import java.util.concurrent.concurrenthashmap;
/**
 * @auther: ms.y
 */
@configuration
public class kafkautil {
    private static final concurrenthashmap<string, kafkatemplate<string, string>> templatecache = new concurrenthashmap<>();
    private map<string, object> kafkaproducerconfigs(string servers, map<string, object> otherconfigs) {
        map<string, object> props = new hashmap<>();
        props.put(producerconfig.bootstrap_servers_config, servers);
        props.put(producerconfig.key_serializer_class_config, stringserializer.class);
        props.put(producerconfig.value_serializer_class_config, stringserializer.class);
        if (otherconfigs != null){
            otherconfigs.foreach(props::put);
        }
        return props;
    }
    public kafkatemplate<string, string> getkafkatemplate(string servers, map<string, object> otherconfigs) {
        return templatecache.computeifabsent(servers, bs -> createkafkatemplate(bs, otherconfigs));
    }
    private kafkatemplate<string, string> createkafkatemplate(string servers, map<string, object> otherconfigs) {
        map<string, object> configs = kafkaproducerconfigs(servers, otherconfigs);
        producerfactory<string, string> producerfactory = new defaultkafkaproducerfactory<>(configs);
        return new kafkatemplate<>(producerfactory);
    }
    @predestroy
    public void destroy() {
        for (kafkatemplate<string, string> template : templatecache.values()) {
            template.destroy();
        }
    }
}

怎么用

1.正常的autowired就行
2.自己查配置还有是否需要认证的信息
3.获取template去发送数据
4.处理结果

map<string,object> configs = new hashmap<>();
//自己给configs 填值
kafkatemplate kafkatemplate = kafkautil.getkafkatemplate("ip:port",configs);
kafkatemplate.send("topic名称", "消息内容");