Kafka快速入門系列(9) | Kafka的Producer API操作
2021-09-02 21:29:03

??本篇博主帶來的是Kafka的Producer API操作。

---

1. 消息發(fā)送流程

??Kafka的Producer發(fā)送消息采用的是異步發(fā)送的方式。在消息發(fā)送的過程中，涉及到了兩個線程——main線程和Sender線程，以及一個線程共享變量——RecordAccumulator。main線程將消息發(fā)送給RecordAccumulator，Sender線程不斷從RecordAccumulator中拉取消息發(fā)送到Kafka broker。

下圖為KafkaProducer發(fā)送消息流程


相關參數：
batch.size：只有數據積累到batch.size之后，sender才會發(fā)送數據。
linger.ms：如果數據遲遲未達到batch.size，sender等待linger.time之后就會發(fā)送數據。

2. 無回調參數的API

• 1. 導入依賴

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>

    </build>


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

• 2. 編寫代碼

需要用到的類：
KafkaProducer：需要創(chuàng)建一個生產者對象，用來發(fā)送數據
ProducerConfig：獲取所需的一系列配置參數
ProducerRecord：每條數據都要封裝成一個ProducerRecord對象

• 3. 完整代碼

package com.buwenbuhuo.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;


/**
 * @author 卜溫不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop002: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緩沖區(qū)大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("second", i + "", "my name is bu wen bu huo -" + i));
        }
        producer.close();
    }
}

• 4. 群起zookeeper和kafka

[bigdata@hadoop002 zookeeper-3.4.10]$ bin/start-allzk.sh 
[bigdata@hadoop002 kafka]$ bin/start-kafkaall.sh 

• 5. 啟動一個consumer控制臺

[bigdata@hadoop002 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop002:9092 --topic first

• 6. 跑代碼進行測試
  

為什么是無序的，這是因為我們只能保證分區(qū)內有序。其他無序。

3. 帶回調函數的API

??回調函數會在producer收到ack時調用，為異步調用，該方法有兩個參數，分別是RecordMetadata和Exception，如果Exception為null，說明消息發(fā)送成功，如果Exception不為null，說明消息發(fā)送失敗。
??注意：消息發(fā)送失敗會自動重試，不需要我們在回調函數中手動重試。

• 1. 代碼

package com.buwenbuhuo.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


/**
 * @author 卜溫不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop002:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);


        //1.創(chuàng)建1個生產者對象
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        //2.調用send方法
        for (int i = 0; i < 1000; i++) {
            producer.send(new ProducerRecord<String, String>("second", i + "", "message-" + i), (metadata,exception)->{

                if(exception==null){
                    System.out.println("success"+metadata.topic()+"-"+metadata.partition()+"-"+metadata.offset());
                }else{
                    exception.printStackTrace();
                }
            });
        }

        //3.關閉生產者
        producer.close();
    }
}

• 2. 結果

• 3. 大致流程圖
  

4. 同步發(fā)送API / 只是比異步多了一個.get()

??同步發(fā)送的意思就是，一條消息發(fā)送之后，會阻塞當前線程，直至返回ack。
??由于send方法返回的是一個Future對象，根據Futrue對象的特點，我們也可以實現同步發(fā)送的效果，只需在調用Future對象的get方發(fā)即可。

• 1. 代碼：

package com.buwenbuhuo.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author 卜溫不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop002: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緩沖區(qū)大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<String, String>("second", i + "", "my name is bu wen bu huo -" + i)).get();
        }
        producer.close();
    }
}

• 2. 結果
  
  ??本次的分享就到這里了,

---

?? 看 完 就 贊 ， 養(yǎng) 成 習 慣 ！ ！ ！ color{#FF0000}{看完就贊，養(yǎng)成習慣?。?！} 看完就贊，養(yǎng)成習慣！！！^ _ ^ ?? ?? ??
??碼字不易，大家的支持就是我堅持下去的動力。點贊后不要忘了關注我哦！