APP
??本篇博主帶來的是Kafka的Producer API操作。
1. 消息發(fā)送流程
??Kafka的Producer發(fā)送消息采用的是異步發(fā)送的方式。在消息發(fā)送的過程中，涉及到了兩個線程——main線程和Sender線程，以及一個線程共享變量——RecordAccumulator。main線程將消息發(fā)送給RecordAccumulator，Sender線程不斷從RecordAccumulator中拉取消息發(fā)送到Kafka broker。
2. 無回調(diào)參數(shù)的API下圖為KafkaProducer發(fā)送消息流程
相關參數(shù)：
batch.size：只有數(shù)據(jù)積累到batch.size之后，sender才會發(fā)送數(shù)據(jù)。
linger.ms：如果數(shù)據(jù)遲遲未達到batch.size，sender等待linger.time之后就會發(fā)送數(shù)據(jù)。
- 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)建一個生產(chǎn)者對象，用來發(fā)送數(shù)據(jù)
ProducerConfig：獲取所需的一系列配置參數(shù)
ProducerRecord：每條數(shù)據(jù)都要封裝成一個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);//重試次數(shù)
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. 跑代碼進行測試
3. 帶回調(diào)函數(shù)的API為什么是無序的，這是因為我們只能保證分區(qū)內(nèi)有序。其他無序。
??回調(diào)函數(shù)會在producer收到ack時調(diào)用，為異步調(diào)用，該方法有兩個參數(shù)，分別是RecordMetadata和Exception，如果Exception為null，說明消息發(fā)送成功，如果Exception不為null，說明消息發(fā)送失敗。
??注意：消息發(fā)送失敗會自動重試，不需要我們在回調(diào)函數(shù)中手動重試。
- 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個生產(chǎn)者對象
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//2.調(diào)用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.關閉生產(chǎn)者
producer.close();
}
}
- 2. 結果
-
3. 大致流程圖
??同步發(fā)送的意思就是，一條消息發(fā)送之后，會阻塞當前線程，直至返回ack。
??由于send方法返回的是一個Future對象，根據(jù)Futrue對象的特點，我們也可以實現(xiàn)同步發(fā)送的效果，只需在調(diào)用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);//重試次數(shù)
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)成習慣！！！^ _ ^ ?? ?? ??
??碼字不易，大家的支持就是我堅持下去的動力。點贊后不要忘了關注我哦！
本文摘自 ：https://blog.51cto.com/u
