kafka生产者和消费者

目录

生产者命令行操作

消费者命令行操作

Kafka生产者

生产者消息发送流程：发送原理

生产者重要参数列表

生产者命令行操作

1、查看操作生产者命令参数

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh

参数	描述
--bootstrap-server	连接的Kafka Broker主机名称和端口号
--topic	操作的topic名称

2、发送消息 

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello
>atguigu
>hyf
>ljj

消费者命令行操作

查看操作消费者命令参数

1、[atguigu@hadoop102 kafka[$ bin/kafka-console-consumer.sh

参数	描述
--bootstrap-server	连接的Kafka Broker主机名称和端口号
--topic	操作的topic名称
--from-beginning	从头开始消费
--group	指定消费组名称

消费信息 

1、消费first主题中的数据

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

2、把主题中所有的数据都读取出来（包括历史数据）

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first

注意：生产者和消费者需同时建立连接才可以接收数据。

Kafka生产者

生产者消息发送流程：发送原理

在消息发送的过程中，涉及到了两个线程——main线程和Sender线程。在main线程中创建了一个双端队列RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka Broker。

发送流程图：

（批次大小） batch.size：只有数据积累到batch.size之后，sender才会发送数据。默认16k。

（时间）linger.time：如果数据迟迟未到batch.size，sender等待linger.size设置的时间到了之后就会发送数据。单位ms，默认值是0ms（可以设置），表示没有延迟

应答acks级别：

0：生产者发送过来的数据，不需要等数据罗盘应答。

1：生产者发送过来的数据，Leader收到数据后应答。

-1（all）：生产者发送过来的数据，Leader和SRI队列里面的所有节点收齐数据后应答。-1和all等价。

 生产者重要参数列表

参数名称	描述
bootstrap.server	生产者连接集群所需的broker地址清单。例如hadoop102：9092，hadoop103：9092，hadoop104：9092，可以设置1个或者多个，中间用逗号隔开。注意这里并非需要所有的broker地址，因为生产者从给定的broker里查找到其他broker信息。
key.serializer和value.serializer	指定发送消息的key和value的序列化类型。一定要写全类名
buffer.memory	RecordAccumulator缓冲区总大小，默认32m。
batch.size	缓冲区一批数据最大值，默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.time	如果数据迟迟未到batch.size,sender等待linger.time之后就会发送数据。单位ms，默认值是0ms，表示没有延迟。生产环境建议该值大小为5-100ms之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。 1：生产者发送过来的数据，Leader 收到数据后应答。 -1（all）：生产者发送过来的数据，Leader+和 isr 队列 里面的所有节点收齐数据后应答。默认值是-1，-1 和 all 是等价的。
max.in.flight.requests.per.connection	允许最多没有返回ack的次数，默认为5，开启幂等性要保证该值是1-5的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries 表示重试次数。默认是 int 最大值，2147483647。 如果设置了重试，还想保证消息的有序性，需要设置 MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1 否则在重试此失败消息的时候，其他的消息可能发送 成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是100ms。
enable.idempotence	是否开启幂等性，默认true，开启幂等性。
compression.type	生产者发送的所有数据的压缩方式。默认是none，也就是不压缩。 支持压缩类型：none、gzip、snappy、lz4和zstd。