【大数据学习 | kafka】producer的参数与结构

1. producer的结构

producer：生产者

它由三个部分组成

interceptor：拦截器，能拦截到数据，处理完毕以后发送给下游，它和过滤器不同并不是丢弃数据，而是将数据处理完毕再次发送出去，这个默认是不存在的

serialiazer：序列化器，kafka中存储的数据是二进制的，所以数据必须经过序列化器进行处理，这个是必须要有的，将用户的数据转换为byte[]的工具类，其中k和v要分别指定

partitioner: 分区器，主要是控制发送的数据到topic的哪个分区中，这个默认也是存在的

record accumulator

本地缓冲累加器 默认32M

producer的数据不能直接发送到kafka集群中，因为producer和kafka集群并不在一起，远程发送的数据不是一次发送一条这样太影响发送的速度和性能，所以我们发送都是攒一批数据发一次，record accumulator就是一个本地缓冲区，producer将发送的数据放入到缓冲区中，另外一个线程会去拉取其中的数据，远程发送给kafka集群，这个异步线程会根据linger.ms和batch-size进行拉取数据。如果本地累加器中的数据达到batch-size或者是linger.ms的大小阈值就会拉取数据到kafka集群中，这个本地缓冲区不仅仅可以适配两端的效率，还可以批次形式执行任务，增加效率

batch-size 默认16KB

linger.ms 默认为0

生产者部分的整体流程

首先producer将发送的数据准备好

经过interceptor的拦截器进行处理，如果有的话

然后经过序列化器进行转换为相应的byte[]

经过partitioner分区器分类在本地的record accumulator中缓冲

sender线程会自动根据linger.ms和batch-size双指标进行管控，复制数据到kafka

2. producer的简单代码

2.1 准备：

引入maven依赖：

<dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.3.2</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.30</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.17</version></dependency>
</dependencies>

在resources文件中创建log4j.properties

log4j.rootLogger=info,console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c %M(): %m%n

2.2 生产者中的设定参数

参数	含义
bootstrap.servers	kafka集群的地址
key.serializer	key的序列化器，这个序列化器必须和key的类型匹配
value.serializer	value的序列化器，这个序列化器必须和value的类型匹配
batch.size	批次拉取大小默认是16KB
linger.ms	拉取的间隔时间默认为0，没有延迟
partitioner	分区器存在默认值
interceptor	拦截器选的

2.3 全部代码

public class producer_test {public static void main(String[] args) {Properties pro = new Properties();pro.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop106:9092");//设定集群地址pro.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());pro.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//设定两个序列化器，其中StringSerializer是系统自带的序列化器，要和数据的类型完全一致pro.put(ProducerConfig.BATCH_SIZE_CONFIG, 16*1024);//batch-size默认是16KB，参数的单位是bytepro.put(ProducerConfig.LINGER_MS_CONFIG, 0);//默认等待批次时长是0KafkaProducer<String, String> producer = new KafkaProducer<String, String>(pro);ProducerRecord<String, String> record = new ProducerRecord<>("topic_a", "this is hainiu");//发送数据的时候有kv两个部分，但是一般k我们什么都不放，只放value的值producer.send(record);producer.close();}
}

在x-shell中观察消费的数据