第4章 Kafka API

4.1 Producer API

4.1.1 消息发送流程

Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

相关参数：

batch.size：只有数据积累到batch.size之后，sender才会发送数据。

linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。

4.1.2 异步发送API

1）导入依赖

 

org.apache.kafka

kafka-clients

0.11.0.0

 

2）编写代码

需要用到的类：

KafkaProducer：需要创建一个生产者对象，用来发送数据

ProducerConfig：获取所需的一系列配置参数

ProducerRecord：每条数据都要封装成一个ProducerRecord对象

1.不带回调函数的API

package com.atguigu.kafka;

 

import org.apache.kafka.clients.producer.*;

 

import java.util.Properties;

import java.util.concurrent.ExecutionException;

 

public class CustomProducer {

 

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties props = new Properties();

 

        //kafka集群，broker-list

        props.put("bootstrap.servers", "hadoop102:9092");

 

        props.put("acks", "all");

 

        //重试次数

        props.put("retries", 1);

 

        //批次大小

        props.put("batch.size", 16384);

 

        //等待时间

        props.put("linger.ms", 1);

 

        //RecordAccumulator缓冲区大小

        props.put("buffer.memory", 33554432);

 

        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 

        Producer producer = new KafkaProducer<>(props);

 

        for (int i = 0; i < 100; i++) {

            producer.send(new ProducerRecord("first", Integer.toString(i), Integer.toString(i)));

        }

 

        producer.close();

    }

}

2.带回调函数的API

回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是RecordMetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

package com.atguigu.kafka;

 

import org.apache.kafka.clients.producer.*;

 

import java.util.Properties;

import java.util.concurrent.ExecutionException;

 

public class CustomProducer {

 

public static void main(String[] args) throws ExecutionException, InterruptedException {

 

        Properties props = new Properties();

 

        props.put("bootstrap.servers", "hadoop102:9092");//kafka集群，broker-list

 

        props.put("acks", "all");

 

        props.put("retries", 1);//重试次数

 

        props.put("batch.size", 16384);//批次大小

 

        props.put("linger.ms", 1);//等待时间

 

        props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小

 

        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 

        Producer producer = new KafkaProducer<>(props);

 

        for (int i = 0; i < 100; i++) {

            producer.send(new ProducerRecord("first", Integer.toString(i), Integer.toString(i)), new Callback() {

 

                //回调函数，该方法会在Producer收到ack时调用，为异步调用

                @Override

                public void onCompletion(RecordMetadata metadata, Exception exception) {

                    if (exception == null) {

                        System.out.println("success->" + metadata.offset());

                    } else {

                        exception.printStackTrace();

                    }

                }

            });

        }

        producer.close();

    }

}