【kafka专栏】生产者同步及异步发送数据三种方式

文章目录

一、新建一个项目

本文我们为大家介绍apache kafka生产者同步及异步发送数据三种方式，基于java项目实现。为了方便应用，我们新建一个java的maven项目


引入kafka 的Java客户端依赖，同时假如JUnit5单元测试依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.6.0</version></dependency><dependency><groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter</artifactId><version>RELEASE</version><scope>test</scope></dependency>

二、设置Producer客户端参数

ProducerConfig：设置生产者客户端的一系列配置参数。

publicclass kafkaProducerTest {privateProperties props;@BeforeEachpublicvoidpreTest(){
        props =newProperties();//kafka broker服务列表
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.137.10:9092");//可靠性确认应答参数
        props.put(ProducerConfig.ACKS_CONFIG,"all");//发送失败，重新尝试的次数
        props.put(ProducerConfig.RETRIES_CONFIG,1);//生产者数据key的格式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());//生产者数据value的格式
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());}}

三、构建消息对象

kafka 的ProducerRecord就是我们要发送的消息，里面包含了很多的信息，主要有以下几个构造函数，可以更具自己的需求选择合适的构造函数。

 public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {}
 public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value) {}
 public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers) {}
 public ProducerRecord(String topic, Integer partition, K key, V value) {}
 puic ProducerRecord(String topic, K key, V value) {}
 public ProducerRecord(String topic, V value) {}

• topic：Topic主题的的名称
• partition：主题的分区编号，编号从0开始。表示消息数据指定发送到该分区中
• timestamp：时间戳，一般默认当前时间戳
• key：消息的键，可以是不同数据类型，但是通常是String。具有相同key的消息被发往同一个分区，也就是说具有相同key的消息可以保证数据有序性。
• data：消息的数据，可以是不同数据类型
• headers：消息的自定义标签信息

四、三种数据发送方式

Kafka中有三种发送消息的方式：

1. 只发不管结果（fire-and-forget）：只调用接口发送消息到Kafka服务器，但不管成功写入与否。由于Kafka是高可用的，因此大部分情况下消息都会写入，但在异常情况下会丢消息。
2. 同步发送（Synchronous send）：调用send()方法返回一个Future对象，我们可以使用它的get()方法来判断消息发送成功与否。会阻塞当前线程，执行效率低，但可靠性最高。
3. 异步发送（Asynchronous send）：调用send()时提供一个回调方法，不会阻塞当前线程，当接收到broker结果后回调此方法。可靠性不如同步发送，可能因为等问题不保证100%被回调。

4.1.不带回调函数

KafkaProducer#send(ProducerRecord<K,V>)

• KafkaProducer：生产者对象，用来发送数据
• ProducerRecord：每条数据都要封装成一个ProducerRecord 对象

//没有回调函数的调用方法@TestpublicvoidnoCallback(){KafkaProducer<String,String> producer =newKafkaProducer<>(props);for(int i =0; i <20; i++){
        producer.send(newProducerRecord<>("producer_test",Integer.toString(i),"noCallback value:"+ i));}
    producer.close();}

4.2.带回调函数

KafkaProducer#send(ProducerRecord<K,V>, Callback)

使用方法

@TestpublicvoidwithCallback(){KafkaProducer<String,String> producer =newKafkaProducer<>(props);

    producer.send(newProducerRecord<>("producer_test","withCallback value:"),(metadata, exception)->{//回调函数if(null== exception){System.out.println("消息数据发送成功 ->"+ metadata.offset());}else{
                    exception.printStackTrace();}});

    producer.close();}

回调函数会在producer收到ack之前异步调用，该方法有两个参数：RecordMetadata、Exception。这两个参数是互斥的关系，即：

• 如果Exception为null，则消息发送成功，此时RecordMetadata必定不为null。
• 消息发送异常时，RecordMetadata为null，而exception不为null。消息发送失败会自动重试，不需在回调函数中手动重试。重试次数由参数retries设置来设定，重试次数达到上线后，仍然发送失败才会有exception的存在。

4.3.同步发送

KafkaProducer的send方法都是异步的，由上面异步的例子可以看出，producer的send方法返回对象是Future类型，因此可以通过调用Future对象的get()方法阻塞，等待发送结果的响应，从而达到同步发送消息的效果。这里的同步是指，一条消息发送后会阻塞当前线程，直至返回ack消息。

因此与异步代码区别只有一处（get方法）：

//同步数据send
@Test
public void syncSend() throws ExecutionException, InterruptedException {
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    RecordMetadata metadata = producer.send(new ProducerRecord<>("producer_test",
            "syncSend value"))
            .get();
    if(metadata != null && metadata.hasOffset()){
        System.out.println("数据同步发送确认成功 ->" + metadata.offset());}
    producer.close();}