APP
本篇博主带来的是Kafka的Producer API操作。
1. 消息发送流程
Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。
2. 无回调参数的API下图为KafkaProducer发送消息流程
相关参数：
batch.size：只有数据积累到batch.size之后，sender才会发送数据。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。
- 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：需要创建一个生产者对象，用来发送数据
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缓冲区大小
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. 带回调函数的API为什么是无序的，这是因为我们只能保证分区内有序。其他无序。
回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是RecordMetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。
- 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.创建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. 大致流程图
同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回ack。
由于send方法返回的是一个Future对象，根据Futrue对象的特点，我们也可以实现同步发送的效果，只需在调用Future对象的get方发即可。
- 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缓冲区大小
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. 结果
本次的分享就到这里了,
看
完
就
赞
，
养
成
习
惯
！
！
！
color{#FF0000}{看完就赞，养成习惯！！！}
看完就赞，养成习惯！！！^ _ ^ ❤️ ❤️ ❤️
码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！
本文摘自 ：https://blog.51cto.com/u
