kafka笔记1--集群搭建：https://blog.csdn.net/u010452388/article/details/84674454

本文主要讲解kafka生产者和消费者的API使用，以及部分配置说明

目录

一 引入依赖

二 生产者

2.1 代码

2.2 生产者配置说明

2.3 结果-生产者

三 消费者

3.1 代码

3.2 消费者配置说明

3.3 结果-消费者

一 引入依赖

这里引入客户端依赖的时候尽量保持与服务端版本一致，不然会出现奇怪的错误，看下服务端版本

从上面的图可以看出，服务端版本为1.1.0版本，前面的2.12是Scala版本，所以客户端引入下面的依赖

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>1.1.0</version>
        </dependency>

下图为生产者和消费者的工程结构图

二 生产者

2.1 代码

这里的生产者采用的是每隔1秒钟发送一条消息，总共发送19条

public class KafkaProducerDemo {

    public static Properties kafkaProperties() {

        Properties properties = new Properties();

        /*设置集群kafka的ip地址和端口号*/
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.2.231:9092,192.168.2.232:9092");
        /*发送的消息需要leader确认*/
        properties.put(ProducerConfig.ACKS_CONFIG, "1");
        /*用户id*/
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerDemo");
        /*对key进行序列化*/
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.IntegerSerializer");
        /*对value进行序列化*/
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        return properties;
    }

    public static void main(String[] args) throws InterruptedException {
        /*创建一个kafka生产者*/
        KafkaProducer<Integer, String> kafkaProducer =
                new KafkaProducer<Integer, String>(kafkaProperties());
        /*主题*/
        String topic = "test";
        /*循环发送数据*/
        for (int i = 0; i < 20; i++) {
            /*发送的消息*/
            String message = "我是一条信息" + i;
            /*发出消息*/
            kafkaProducer.send(new ProducerRecord<>(topic, message));
            System.out.println(message + "->已发送");
            Thread.sleep(1000);
        }

    }
}

2.2 配置-生产者

官网生产者配置详情：http://kafka.apache.org/11/documentation.html#producerconfigs

下面是根据官网的配置进行翻译的，如果有翻译的不对的，可以留言沟通

1.acks（默认为1）
    "0":消息发送给broker以后，不需要确认（性能高，但是会出现数据丢失）
    "1":只需要获得kafka集群中leader节点的确认即可返回（leader/follower）
    all或者"-1" 需要ISR中的所有Replica(副本)进行确认（需要集群中所有节点确认）

2.batch.size(默认16KB) 调优的重要参数
    producer对于同一个分区来说，会按照batch.size的大小进行统一收集，然后批量发送
    就是说我们如果发送的消息，不会直接发出去，等达到batch.size之后，再发出去

3.linger.ms(默认0毫秒)
    延迟发送，假如设置的值是1000的话，就是每隔1秒钟积累之前的信息，然后再发送
    
4.max.request.size(默认1M)
    消息最大发送的字节数，超过默认值1M的话，就会拒绝，抛异常

2.3 结果展示

发送的结果如下图：

三 消费者

3.1 代码

public class KafkaConsumerDemo extends Thread {

    private final KafkaConsumer<Integer,String> consumer;

    public KafkaConsumerDemo(String topic) {
        Properties props = new Properties();
        /*设置集群kafka的ip地址和端口号*/
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "192.168.2.231:9092,192.168.2.232:9092");
        /*设置消费者的group.id*/
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaConsumerDemo1");
        /*消费信息以后自动提交*/
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        /*控制消费者提交的频率*/
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
        /*key反序列化*/
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.IntegerDeserializer");
        /*value反序列化*/
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        /*创建kafka消费者对象*/
        consumer = new KafkaConsumer<Integer, String>(props);
        /*订阅主题*/
        consumer.subscribe(Collections.singleton(topic));

    }

    @Override
    public void run() {
        while (true) {
            /*每隔一段时间到服务器拉取数据*/
            ConsumerRecords<Integer, String> consumerRecords = consumer.poll(1000);
            for (ConsumerRecord record : consumerRecords) {
                System.out.println(record.value());
            }
        }
    }
    public static void main(String[] args) {
        new KafkaConsumerDemo("test").start();
    }
}

3.2 配置说明

1.group.id
    同一个gropu.id中的消费者，只能有一个消费者可以消费到信息
    但是不同的group.id都会去消费消息（消息是持久化的）
    
2.enable.auto.commit 
    如果位true的话，消费者消费消息以后自动提交，只有当消息提交以后，该消息才不会被再次接收到

3.auto.commit.interval.ms
    控制消费者提交的频率，默认单位是毫秒，一般配合enable.auto.commit

4.auto.offset.reset
    这个参数是针对新的 groupid 中的消费者而言的，当有新 groupid 的消费者来
    消费指定的 topic 时，对于该参数的配置，会有不同的语义

    auto.offset.reset=latest 情况下，新的消费者将会从其他消费者最后消费的offset 处开始消费 Topic 下的消息
    auto.offset.reset= earliest 情况下，新的消费者会从该 topic 最早的消息开始消费
    auto.offset.reset=none 情况下，新的消费者加入以后，由于之前不存在offset，则会直接抛出异常。
    
5.max.poll.records
    设置限制每次消费者poll返回的消息数，通过调整此值，可以减少poll的间隔

官网消费者配置详情：http://kafka.apache.org/11/documentation.html#consumerconfigs

3.3 结果展示

消费的结果如下图：