kafka 生产和消费信息入门

启动生产者

kafka-console-producer.sh \
--broker-list mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

启动消费者

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet

默认情况下,消费者获取的是启动后生产者生产的信息,如果想要获取启动之前的所有消息,需要加参数--from-beginning

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic pet \
--from-beginning

指定offset获取信息

如果想从指定位置获取信息,就可以指定offset信息.

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic cat \
--partition 0 \
--group group1 \
--offset earliest

使用offset参数必须同时指定partition参数,否走会报错,如下所示

[root@mypc01 /]# kafka-console-consumer.sh \
> --bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
> --topic cat \
> --offset earliest
The partition is required when offset is specified.

在 Kafka 中，每当消费者组内的消费者查找不到所记录的消费位移或发生位移越界时，就会根据消费者客户端参数 auto.offset.reset 的配置来决定从何处开始进行消费，这个参数的默认值为 “latest” 。

auto.offset.reset 的值可以为 earliest、latest 和 none 。
各含义在真实情况如下所示：
earliest ：
当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；
无提交的 offset 时，从头开始消费。
latest ：
当各分区下存在已提交的 offset 时，从提交的 offset 开始消费；
无提交的 offset 时，消费该分区下新产生的数据。
none ：
topic 各分区都存在已提交的 offset 时，从 offset 后开始消费；
只要有一个分区不存在已提交的offset，则抛出异常。

kafka从头开始消费所有信息

（1）使用一个全新的"group.id"（就是之前没有被任何消费者使用过）;（2）指定"auto.offset.reset"参数的值为earliest；

比如

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic cat \
--partition 0 \
--group group1 \
--offset earliest

或者也可以

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic catt \
--partition 0 \
--group group1 \
--offset 0

或者

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic cat \
--partition 0 \
--group group1 \
--from-beginning

以上的全部消息指的是某分区的全部消息,并不是所有分区的全部消息,如果要消费所有分区的所以消息,可以用这个,不指定分区,然后用from-beginning参数

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--topic cat \
--from-beginning

查看某个消费组的消费情况

比如你想知道现在的offset是多少,就是消费到什么位置了.可以如下操作

[root@mypc01 /]# kafka-consumer-groups.sh \
> --bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
> --describe \
> --group group1
Note: This will not show information about old Zookeeper-based consumers.
Consumer group 'group1' has no active members.TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
pet             0          16474           16474           0               -               -               -

查看总得消费条数

ctrl+c

Processed a total of 16474 messages

消息信息的无序说明

就是你获取的信息和发送消息的顺序是不一样的~
1 指的是分区间数据无序,分区内有序
2 可以使用–from-beginning来显示效果
获取所有分区的全部消息,不一定是先获取哪一个分区的数据.

消费者组与Partition

1 消费者是可以分组的
2 生产者的信息可以被多个组的消费者同时消费,组与组之间不受影响
3 同一个组内的消费者,一个消费者只能消费一个分区的数据. 前提分区数==组内消费者数量;
如果消费者数量大于分区数量,多余的消费者闲着;
如果消费者的数量<分区数量,某些消费者可以消费多个分区的数据.

kafka消费者在消费数据的时候，都是分组别的。不同组的消费不受影响，相同组内的消费，需要注意，

如果partition有3个，消费者有3个，那么便是每一个消费者消费其中一个partition对应的数据；

-如果有2个消费者，此时一个消费者消费其中一个partition数据，另一个消费者消费2个partition的数据。

如果有超过3个的消费者，同一时间只能最多有3个消费者能消费得到数据。

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--group g1 \
--topic pet

kafka-console-consumer.sh \
--bootstrap-server mypc01:9092,mypc02:9092,mypc03:9092 \
--group g2 \
--topic pet

报错

如何报如下错误,说明有一台kafka没有启动

 WARN [Consumer clientId=consumer-1, groupId=console-consumer-98098] Connection to node -3 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)