微信公众号：大数据开发运维架构
关注可了解更多大数据相关的资讯。问题或建议，请公众号留言;
如果您觉得“大数据开发运维架构”对你有帮助，欢迎转发朋友圈
从微信公众号拷贝过来，格式有些错乱，建议直接去公众号阅读
概述：
上篇文章：Flink实战：写入Kafka自定义序列化类和自定义分区器，里面讲了如何自定义分区和自定义序列化类向Kafka集群写入数据，上一个代码实例中的Kafka集群并没有开启SSL加密认证，Kafka开启SSLr认证的我之前也写了一篇文章：Kafka实战:集群SSL加密认证和配置（最新版kafka-2.4.0），之前对这块东西不是很熟悉的同行，自己去看下我这边文章，里面详细讲了如何开启和配置SSL，这里就不在说了，很多粉丝问我对于SSL加密的Kafka集群如何集成Flink进行消息的写入，我这里我给大家详细讲解下实例代码,今天我这里软件版本Flink1.9.0、Kafka2.4.0,不同版本依赖不太一样，建议自己根据自己的版本参考Flink的官网教程进行修改：
1.上篇文章的代码我没有粘贴我的maven依赖，可能好多人对这个比较晕乎，我这里直接把pom.xml也给你贴出来，里面有冗余依赖你可以自己调试删除一些：
UTF-81.2.171.7.221.9.01.82.11${java.version}${java.version}1.2.5<!--<kafka.version>0.10.1.0</kafka.version>-->2.4.02.7.34.5.2junitjunit4.11testorg.apache.flinkflink-java${flink.version}<!--<scope>provided</scope>-->org.apache.flinkflink-streaming-java_${scala.binary.version}${flink.version}<!--<scope>provided</scope>-->org.scala-langscala-library2.11.8org.apache.kafkakafka-clients${kafka.version}compileorg.apache.flinkflink-connector-kafka_2.111.9.0org.apache.flinkflink-connector-kafka-0.10_2.111.9.0compileorg.apache.flinkflink-hadoop-fs${flink.version}org.apache.hadoophadoop-common${hadoop.version}org.apache.hadoophadoop-hdfs${hadoop.version}org.apache.httpcomponentshttpclient${httpclient.version}org.apache.flinkflink-streaming-java_2.111.9.1compile<!--<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> </exclusions> </dependency>-->
2.自定义序列化类CustomKeyedSerializationSchema，还是上篇文章一样，这里也贴出来吧：
packagecom.hadoop.ljs.flink.utils;importorg.apache.flink.streaming.util.serialization.KeyedSerializationSchema;importjava.util.Map;/***@author: Created By lujisen*@companyChinaUnicom Software JiNan*@date: 2020-02-24 20:57*@version: v1.0*@description: com.hadoop.ljs.flink.utils */publicclassCustomKeyedSerializationSchemaimplementsKeyedSerializationSchema{@Overridepublicbyte[] serializeKey(String s) {/*根据传过来的消息，自定义key*/String[] line=s.split(",");System.out.println("key::::"+line[0]);return("key--"+line[0]).getBytes(); }@Overridepublicbyte[] serializeValue(String s) {/*根据传过来的消息，自定义value*/String[] line=s.split(",");System.out.println("value::::"+line[1]);return("value--"+line[1]).getBytes(); }@OverridepublicStringgetTargetTopic(String topic){/*这里是目标topic，一般不需要操作*/returnnull; }}
3.主函数测试类：
packagecom.hadoop.ljs.flink.streaming;importcom.hadoop.ljs.flink.utils.CustomKeyedSerializationSchema;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;importorg.apache.kafka.clients.CommonClientConfigs;importorg.apache.kafka.clients.producer.ProducerConfig;importorg.apache.kafka.common.config.SslConfigs;importjava.util.Properties;/***@author: Created By lujisen*@companyChinaUnicom Software JiNan*@date: 2020-02-24 21:27*@version: v1.0*@description: com.hadoop.ljs.flink.utils */publicclassFlinkKafkaSSLProducer{publicstaticfinalString topic="topic1";publicstaticfinalString bootstrap_server="salver31.hadoop.unicom:9093,salver32.hadoop.unicom:9093";publicstaticfinalString client_truststore="D:\\kafkaSSL\\client.truststore.jks";publicstaticfinalString client_keystore="D:\\kafkaSSL\\client.keystore.jks";publicstaticfinalString client_ssl_password="123456";publicstaticfinalString hostname="localhost";publicstaticfinalintport=9000;publicstaticvoidmain(String[] args)throwsException{/*获取flink流式计算执行环境*/finalStreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();/*从Socket端接收数据*/DataStream dataSource = senv.socketTextStream(hostname, port,"\n");/*下面可以根据自己的需求进行自动的转换*//* SingleOutputStreamOperator<Map<String, String>> messageStream = dataSource.map(new MapFunction<String, Map<String, String>>() { @Override public Map<String, String> map(String value) throws Exception { System.out.println("接收到的数据："+value); Map<String, String> message = new HashMap<>(); String[] line = value.split(","); message.put(line[0], line[1]); return message; } });*//*接收的数据，中间可经过复杂的处理，最后发送到kafka端*/dataSource.addSink(newFlinkKafkaProducer(topic,newCustomKeyedSerializationSchema(), getProducerProperties(),FlinkKafkaProducer.Semantic.EXACTLY_ONCE));/*启动*/senv.execute("FlinkKafkaProducer"); }/*获取Kafka配置*/publicstaticPropertiesgetProducerProperties(){Properties props =newProperties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap_server);//configure the following three settings for SSL Encryptionprops.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,"SSL"); props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, client_truststore); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, client_ssl_password);// configure the following three settings for SSL Authentication props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, client_keystore); props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, client_ssl_password); props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, client_ssl_password);//这个选项必填，不然报错hostname ...not found， 这里取消主机名校验props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,"");props.put(ProducerConfig.ACKS_CONFIG,"all");props.put(ProducerConfig.RETRIES_CONFIG,0);//这里一定要改下序列化key.value类，不然会报错： Can't convert key of class [B to class org.apache.kafka.common.serialization.StringSerializer。。。。。。。。props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");/*props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");*/returnprops; }}
4.测试验证。我这里从window端socket的9000端口接收数据，发送至kafka2.4.0集群，测试命令：
发送数据window：
nc-lp9000
kafka端接收数据，由于我是通过消费者端看的，所以只能看到value值：
5.至此，Flink连接开启SSL认证Kafka2.4.0集群验证完成。
后面我会继续写两篇相应的文章：
1.如何连接开启Kerberos认证的Kafka集群；
2.如何通过Flink连接开启Kerberos认证的Kafka集群发送消息。
敬请关注！！！
