背景

今天需要通过代码连接kafka拨测kafka的主题是否正常，查了些资料，用java写了个示例demo。

由于分内网和外网的访问，外网访问需要通过ssl证书，因此两种验证方式都测试了一下。

主要测试步骤如下：

1、三个java类

ConsumerClient 消费类
ProducerClient   生产类
KafkaUtil   连接工具类

2、添加pom.xml添加依赖

<dependency>   <groupId>org.apache.kafka</groupId>   <artifactId>kafka-clients</artifactId>   <version>2.5.1</version></dependency><dependency>   <groupId>org.apache.zookeeper</groupId>   <artifactId>zookeeper</artifactId>   <version>3.6.3</version></dependency>

3、将证书放到一个有读权限的目录下

例如：

/home/kafka/client.truststore.jks

4、修改类：KafkaUtil 的参数clientTruststoreJKS

例如：

public static final String clientTruststoreJKS="/home/kafka/client.truststore.jks";

5、启动消费端

执行ConsumerClient类的main方法

public static void main(String[] args) {    fecthKafka("kn");//外网验证测试测试ssl    fecthKafka(null);//内网测试不需要ssl}

6、启动生产端

执行ProducerClient类的main方法

public static void main(String[] args) {    sendToKafka("kn");//外网 ssl    sendToKafka("kn");//内网  非ssl}

详细代码如下：

  package com.example.demo.Utils;

import java.util.Properties;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.common.config.SaslConfigs;import org.apache.kafka.common.config.SslConfigs;import org.springframework.stereotype.Component;

/** * kafka有用户名验证的配置 */@Componentpublic class KafkaUtil {
    //服务器地址 没密码使用PLAINTEXT前缀 有密码了使用SASL_PLAINTEXT前缀    public static final String servers="192.36.39.38:9094,192.36.39.39:9094,192.36.39.40:9094,192.36.39.41:9094";
    public static final String serversSSL="kafka.nginx:9004,kafka.nginx:9005,kafka.nginx:9006";        public static final String clientTruststoreJKS="/home/kafka/client.truststore.jks";
    //kafka集群生产者配置    public static KafkaProducer<String, String> getProducer(String username,String passwd,String flag) {        Properties props = new Properties();        if("kn".equals(flag)){            props.put("bootstrap.servers",serversSSL);            props.put("security.protocol", "SASL_SSL");            props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,"");            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,clientTruststoreJKS);            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,"证书密码");        }else{            props.put("bootstrap.servers",servers);            props.put("security.protocol", "SASL_PLAINTEXT");        }        props.put("acks", "1");        props.put("retries", 0);        props.put("batch.size", 0);//16384        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");        props.setProperty("sasl.mechanism", "SCRAM-SHA-512");        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username="+username+" password= "+ passwd+";");        //props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + username + "' password='" + passwd + "';");        /*String jassc = "org.apache.kafka.common.security.plain.PlainLoginModule required\n"                + "username=\"" + username + "\"\n"                + "password=\"" + passwd + "\";";*/       // props.setProperty("sasl.jaas.config", jassc);        KafkaProducer<String, String> kp = new KafkaProducer<String, String>(props);
        return kp;    }

    public static KafkaConsumer<String, String> getConsumer(String groupId,String username,String passwd,String flag) {        Properties props = new Properties();        if("kn".equals(flag)){            props.put("bootstrap.servers", serversSSL);            props.setProperty ("security.protocol", "SASL_SSL");            props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG,"");            props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,clientTruststoreJKS);            props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,"FjBd#2202!");        }else{            props.put("bootstrap.servers", servers);            props.setProperty ("security.protocol", "SASL_PLAINTEXT");        }                      props.put("auto.offset.reset", "earliest"); //必须要加，如果要读旧数据        props.put("group.id", groupId);         props.put("enable.auto.commit", "false");        props.put("auto.commit.interval.ms", "100");        props.put("max.partition.fetch.bytes", "10240");//每次拉取的消息字节数，10K？，每次取回20条左右        props.put("session.timeout.ms", "30000");        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        props.setProperty ("sasl.mechanism", "SCRAM-SHA-512");        //props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + username + "' password='" + passwd + "';");                props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.scram.ScramLoginModule required username="+username+" password= "+ passwd+";");       /* String jassc = "org.apache.kafka.common.security.plain.PlainLoginModule required\n"                + "username = \"" + username + "\"\n"                + "password =\"" + passwd + "\";";        props.setProperty("sasl.jaas.config", jassc);*/        KafkaConsumer<String, String> kc = new KafkaConsumer<String, String>(props);
        return kc;    }}

package com.example.demo.Utils;
import java.util.concurrent.Future;import org.apache.kafka.clients.producer.Callback;import org.apache.kafka.clients.producer.Producer;import org.apache.kafka.clients.producer.ProducerRecord;import org.apache.kafka.clients.producer.RecordMetadata;import org.apache.kafka.common.protocol.types.Field;import org.springframework.stereotype.Component;

/* * 消费者 */@Componentpublic class ProducerClient {
    /**     * @param args     */    public static void main(String[] args) {        // TODO Auto-generated method stub        sendToKafka("kn");    }

    public static String sendToKafka(String flag) {          Producer<String, String> producer = KafkaUtil.getProducer("用户名","密码",flag);        for(int i=0;i<10;i++){            try {                final ProducerRecord<String, String> record = new ProducerRecord<String, String>("主题topic",                        "d+key-" + i, "{\"name\":\"哈哈\",\"id\":\"218CF4630C2F8795\"}");                Future<RecordMetadata> send = producer.send(record, new Callback() {                    @Override                    public void onCompletion(RecordMetadata metadata, Exception e) {                        if (e != null) {                            e.printStackTrace();                        }                    }                });
                System.out.println("sendToKafka-发送至Kafka:" + "d+key-" + i);                return "sendToKafka-发送至Kafka:" + "d+key-" + i;
            } catch (Exception e) {                e.printStackTrace();
            }        }        producer.close();        return  null;    }}

package com.example.demo.Utils;
import java.util.Arrays;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.springframework.stereotype.Component;
/*消费者*/@Componentpublic class ConsumerClient {
    public  static KafkaConsumer<String, String> consumer = null;
    public static void main(String[] args) {        fecthKafka("kn");    }
    public static void fecthKafka(String flag) {        consumer = KafkaUtil.getConsumer("消费组ID，随意填写","用户名","密码",flag); //group           consumer.subscribe(Arrays.asList("主题topic"));//topics          System.out.println("开始准备消费");        int i=0;        while (true) {            System.out.println("开始消费。。。。。。");            ConsumerRecords<String, String> records ;            try {                records = consumer.poll(Long.MAX_VALUE);//毫秒                System.out.println("records>>>");            }catch (Exception e){                System.out.println("消费异常。。。。。。");                e.printStackTrace();                continue;            }
            for (ConsumerRecord<String, String> record : records) {                System.out.println("fetched from partition " + record.partition() + ", offset: " + record.offset() + ",key: " + record.key() + ",value:" + record.value() );                i++;                System.out.println("i======="+i);            }
            try {                consumer.commitSync();            } catch (Exception e) {                e.printStackTrace();                continue;            }


        }    }}

ITPub博客

关于JAVA拨测kafka的一些细节

最新文章