之前讨论轮询时就说过，不需要担心消费者会在一个无限循环里轮询消息，我们会告诉消费者如何优雅地退出循环。
 如果确定要退出循环，需要通过另一个线程调用 consumer.wakeup() 方法。如果循环运行在主线程里，可以在 ShutdownHook 里调用该方法。要记住， consumer.wakeup() 是消费者唯一一个可以从其他线程里安全调用的方法。调用 consumer.wakeup() 可以退出 poll(), 并抛出 WakeupException 异常，或者如果调用 consumer.wakeup() 时没有线程等待轮询， 那么异常将会在下一轮调用 poll() 时抛出。我们不需要处理 WakeupException，因为它只是用于跳出循环的一种方式。不过， 在退出线程之前调用 consumer.close() 是很有必要的，它会提交任何还没有提交的东西， 并向群组协调器(broker)发送消息，告知自己要离开群组，接下来就会触发再均衡 ，而不需要等待会话超时。
 示例：下面是运行在主线程上的消费者退出线程的代码

        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                System.out.println("Starting exit...");
                // 1、ShutdownHook运行在单独的线程里，所以退出循环最安全的方式
                // 只能是调用 wakeup() 方法。
                consumer.wakeup();
                try {
                    mainThread.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
消费者使用代码示例：
 
    public void processingRecordsTest() {
        try {
            consumer.subscribe(Collections.singleton("test_partition"), new SaveOffsetsOnRebalance());
            consumer.poll(Duration.ofMillis(0));
            // 这里是死循环，直到按下Ctrl + C 键，关闭的钩子会在退出时进行清理
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                System.out.println(" --- shutdown --- " + System.currentTimeMillis());
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("offset:" + record.offset() + ",key:" + record.key() +
                            ",value:" + record.value());
                for (TopicPartition partition: consumer.assignment()){
                    System.out.println("commit offset partition:" + consumer.position(partition));
                consumer.commitSync();
        } catch (Exception e) {
            // 2、在另一个线程里调用 wakeup() 方法，导致 poll() 抛出 WakeupException。
            // 你可以捕获异常以确保应用不会意外终止，但实际上这不是必需的。
            e.printStackTrace();
        } finally {
            try {
            } finally {
                // 3、在退出之前，确保彻底关闭了消费者。
                consumer.close();
                System.out.println("consumer close...");
                    一.发送消息 Kafka向 Broker 发送消息的方式，可以分为三种，分别是 Fire-and-forget、Synchronous send、Asynchronous send。示例代码：public class ProducerSendServiceTest {    /**     * 基本属性     */    private static KafkaProducer&...
Kafka消费者消费模式有两种：pull（拉）模式和push（推）模式。
consumer 默认采用 pull（拉）模式从 broker 中读取数据。pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据。针对这一点，Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有数据可供消费，consumer 会等待一段时间之后再返回，这段时长即为 timeout。
push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 br
				
三 kafka接收消费消息
本节教程在window下演示，如果是在linux上学习的同学，可以将命令的前缀进行替换即可，比如 window 下的 命令前缀 bin\windows\kafka-topics.bat ，则linux下的命令前缀为 bin\kafka-topics.sh;
3.1 创建topic
kafka生产消息使用producer生产者，其核心组件服务器为broker, 消费消息使用comsumer消费者, 消息接收需要使用到 topic; topic中又有分区和副本；
创建一个名为test
				
需求：创建一个独立消费者，消费主题中数据：
注意：在消费者 API 代码中必须配置消费者组 id。命令行启动消费者不填写消费者组 id 会被自动填写随机的消费者组 id
Springboot 自定义日志配置关闭Kafka消费者debug日志打印：在resource目录下添加文件 即可。
测试生产者发送消息：需求：创建一个独立消费者，消费主题 0 号分区的数据。① kafka 消费者消费主题0号分区的数据：
② kafka 生产者向主题的0号分区发送数据：
③ 测试：先启动消费者程序，再启动生产者程序需求：测
				
一：集群消息的发送
./kafka-console-producer.sh --broker-list 192.168.156.131:9092,192.168.156.131:9093,192.168.156.131:9094 --topic my-replicated-topic
[root@localhost bin]# ./kafka-console-producer.sh --broker-list 192.168.156.131:9092,192.168.156.131:9093,
				
文章目录消费者一 基本知识1.1 客户端开发1.2 重要参数1.3 消息的订阅1.4 消息的拉取二 原理解析2.1 反序列化2.2消费位移2.2.1 自动提交2.2.2 手动提交2.2.3  控制或关闭消费2.3.4 指定位移消费2.3 在均衡2.4 消费拦截器2.5 多线程
一 基本知识
在Kafka的消费理念中还有一层消费组（Consumer Group）的概念，每个消费者都有一个对应的消费组。当消息发布到主题后，只会被投递给订阅它的每个消费组中的一个消费者。
每个消费者只能消费所分配到的分区
consumer采用从broker中主动拉取数据。 Kafka采用这种方式。
pull模式不足之处是，如 果Kafka没有数 据，消费者可能会陷入循环中，一直返回 空数据。
不采用push（推）模式是因为：Kafka没有采用这种方式，因为由broker决定消息发送速率，很难适应所有消费者的 消费速率。例如推送的速度是50m/s，Consumer1、Consumer2就来不及处理消息。
1.2Kafka 消费者总体工作流程
1.3 消费者..
				
1、点对点模式(一对一，消费者主动拉取数据，消息收到后消息清除) 消息生产者生产消息发送到 Queue 中，然后消息消费者从 Queue 中取出并且消费消息。消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。 Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
2、发布/订阅模式(一对多，消费者消费数据之后不会清除消息) 消息生产者(发布)将消息发布到 topic 中，同时有多个消息消费者(订阅)消费该消息。和点对点方式不同，发布到 topi
Broker:    消息中间件的处理节点，一个kafka节点就是一个broker
Topic：    用于对消息进行归类，每条消息都要指定一个Topic
Producer： 消息生产者，向broker发送消息
Consumer： 消息消费者，从broker读取消息
发送消息
kafka自带一个producer命令客户端，可以从本地读取内容，也可以直接在命令行输入内容，输入内容以消息的形式发送到kafka集群中，输入的每一行内容当作一个独立的消息。使用kafka 发送消息的客户端