kafka 总结以及 JavaAPI 操作kafka生产者和消费者

矫情吗；* 2023-06-15 07:16 18阅读 0赞

[https://www.jianshu.com/p/4bf007885116  
][https_www.jianshu.com_p_4bf007885116]入门推荐阅读

# 总结 #

1、Segment的概念：  
一个分区被分成相同大小数据条数不相等的Segment，  
每个Segment有多个index文件和数据文件组成

2、数据的存储机制（就是面试题中kafka速度为什么如此之快）：  
首先是Broker接收到数据后，将数据放到操作系统的缓存里（pagecache），  
pagecache会尽可能多的使用空闲内存，  
使用sendfile技术尽可能多的减少操作系统和应用程序之间进行重复缓存，  
写入数据的时候使用顺序写入，写入数据的速度可达600m/s

3、Consumer怎么解决负载均衡？（rebalance）  
1）获取Consumer消费的起始分区号  
2）计算出Consumer要消费的分区数量  
3）用起始分区号的hashCode值模余分区数

4、数据的分发策略？  
Kafka默认调用自己分区器（DefaultPartitioner），  
也可以自定义分区器，需要实现Partitioner特质，实现partition方法

5、Kafka怎么保证数据不丢失？  
Kafka接收数据后会根据创建的topic指定的副本数来存储，  
也就是副本机制保证数据的安全性

6、Kafka的应用：  
①作为消息队列的应用在传统的业务中使用高吞吐、分布式、使得处理大量业务内容轻松自如。  
②作为互联网行业的日志行为实时分析,比如：实时统计用户浏览页面、搜索及其他行为，结合实时处理框架使用实现实时监控，或放到 hadoop/离线数据仓库里处理。  
③作为一种为外部的持久性日志的分布式系统提供服务。主要利用节点间备份数据，文件存储、日志压缩等功能。

——————
    其他应用场景：
    	① 企业内部指标
    		对于某些时效性要求较高的指标，如预警指标等，必须在数据变化时
    		及时计算并发送信息
    	② 通信服务运营商
    		 对于用户套餐中的剩余量进行监控，如流量，语音通话，短信
    	③ 电商行业
    		对于吞吐量特别大和数据变动频次较高的应用，如电商网站，必须使
    		用实时计算来捕捉用户偏好

7、Kafka组件：  
①每个partition在存储层面是append log文件。新消息都会被直接追加到log文件的尾部，每条消息在log文件中的位置称为offset（偏移量）。  
②每条Message包含了以下三个属性：  
1° offset 对应类型：long 此消息在一个partition中序号。可以认为offset是partition中Message的id  
2° MessageSize 对应类型：int。  
3° data 是message的具体内容。  
③越多的partitions意味着可以容纳更多的consumer,有效提升并发消费的能力。  
④总之：业务区分增加topic、数据量大增加partition (副本数<=broker节点数)。

8、实时流处理框架如Storm, Spark Streaming如何实现实时处理的，底层封装了Kafka Stream。  
若是手动实现实时处理的框架，需要自己使用Kafka Stream 库。  
  
org.apache.kafka  
kafka-streams  
1.0.2

9、维护消息订阅方消费的offset的方式有哪些？  
①zookeeper ,参数：–zookeeper  
②kafka集群来维护，参数：–bootstrap-server  
主题名：\_\_consumer\_offsets, 默认： 50个分区； 默认的副本数是：1  
若是达到默认的主题\_\_consumer\_offsets的分区的ha (高容错)，需要在server.properties文件中定制默认的副本数：  
default.replication.factor=3  
③手动维护偏移量 （一般使用redis存储偏移量）

10、几个问题：  
①每次启动一个消费者进程（kafka-console-consumer.sh）,是一个单独的进程  
②手动书写的消费者，可以通过参数来定制是从头开始消费，还是接力消费。需要指定flg (main: args\[\])  
③kafka-console-consumer.sh，每次开启一个消费者进程，有一个默认的消费者组。命名方式是：console-consumer-64328  
④查看消费者组的信息，详见： 4\_笔记\\查看消费者组.png  
⑤PachCache, SendFile

# 首先准备两个配置文件到resources #

**producer.properties**

# Licensed to the Apache Software Foundation (ASF) under one or more
    # contributor license agreements.  See the NOTICE file distributed with
    # this work for additional information regarding copyright ownership.
    # The ASF licenses this file to You under the Apache License, Version 2.0
    # (the "License"); you may not use this file except in compliance with
    # the License.  You may obtain a copy of the License at
    #
    #    http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.
    # see org.apache.kafka.clients.producer.ProducerConfig for more details
    
    ############################# Producer Basics #############################
    
    
    # list of brokers used for bootstrapping knowledge about the rest of the cluster
    # format: host1:port1,host2:port2 ...
    # Kafka服务端的主机名和端口号
    bootstrap.servers=hadoop:9092
    
    # specify the compression codec for all data generated: none, gzip, snappy, lz4
    compression.type=none
    
    # 等待所有副本节点的应答 （follower角色的分区从leader角色的分区中同步完毕消息后，给leader反馈信息）
    acks=all
    
    #消息发送最大尝试次数
    retries=0
    
    #一批消息处理大小
    batch.size=16384
    
    # name of the partitioner class for partitioning events; default partition spreads data randomly
    # partitioner.class=com.l000phone.partition.MyPartition
    
    # the maximum amount of time the client will wait for the response of a request
    #request.timeout.ms=
    
    # how long `KafkaProducer.send` and `KafkaProducer.partitionsFor` will block for
    #max.block.ms=
    
    # the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together
    # 请求延时
    linger.ms=1
    
    # the maximum size of a request in bytes
    #max.request.size=
    
    # the default batch size in bytes when batching multiple records sent to a partition
    #batch.size=
    
    # the total bytes of memory the producer can use to buffer records waiting to be sent to the server
    #发送缓存区内存大小
    buffer.memory=33554432
    
    #消息的key对应的序列化类
    key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
    
    #消息的value对应的序列化类
    value.serializer=org.apache.kafka.common.serialization.StringSerializer

**consumer.properties**

# 生产者producer： #

public class MyMsgProducerDemo {
        
        public static void main(String[] args) {
        
            //步骤：
            Producer<Integer, String> producer = null;
    
            try {
        
                //①准备Properties的实例，并将资源目录下的配置文件producer.properties中定制的参数封装进去
                Properties properties = new Properties();
                properties.load(MyMsgProducerDemo.class.getClassLoader().getResourceAsStream("producer.properties"));
    
                //②KafkaProducer实例的创建
                producer = new KafkaProducer(properties);
    
                //③通过循环模拟发布多条消息
                for (int i = 1; i <= 10; i++) {
        
                    //a）准备消息
                    ProducerRecord<Integer, String> record = new ProducerRecord<>("test", i, i + "\t→ 老同学，最近可好？！呵呵哒哒...");
    
                    //b) 发布消息
                    producer.send(record, new Callback() {
        
                        /**
                         * 当前待发送的消息发送完毕后，下述方法会被回调执行
                         *
                         * @param metadata
                         * @param exception
                         */
    
                        public void onCompletion(RecordMetadata metadata, Exception exception) {
        
                            System.out.printf("当前的消息对应的主题是：%s，内容是：%s，所在的分区是：%d，偏移量是：%d%n",
                                    metadata.topic(), record.value(), metadata.partition(), metadata.offset());
                        }
                    });
                }
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                //⑤资源释放
                if (producer != null) {
        
                    producer.close();
                }
            }
    
    
        }
    }

# 消费者consumer #

import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.TopicPartition;
    
    import java.util.Arrays;
    import java.util.Collection;
    import java.util.Properties;
    
    public class MyMsgConsumerDemo {
        
        public static void main(String[] args) {
        
    
            //步骤：
            KafkaConsumer consumer = null;
            try {
        
                //①Properties的实例，将consumer.properties资源文件中的参数设置封装进去
                Properties properties = new Properties();
                properties.load(MyMsgConsumerDemo.class.getClassLoader().getResourceAsStream("consumer.properties"));
    
                //②KafkaConsumer
                consumer = new KafkaConsumer(properties);
    
                //③指定订阅的主题
                final Consumer<Integer, String> finalConsumer = consumer;
                consumer.subscribe(Arrays.asList("test", "test2"), new ConsumerRebalanceListener() {
        
    
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        
    
                    }
    
                    /**
                     * 从各个分区的开始位置进行订阅
                     * @param partitions
                     */
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        
                        finalConsumer.seekToBeginning(partitions);
                    }
                });
    
                //④循环接收消息
                while (true) {
        
                    //④正式开始进行订阅
                    ConsumerRecords<Integer, String> records = consumer.poll(1000);
    
                    //⑤分析订阅后的结果
                    for (ConsumerRecord<Integer, String> record : records) {
        
                        String topic = record.topic();
                        int partition = record.partition();
                        long offset = record.offset();
                        String value = record.value();
                        Integer key = record.key();
                        System.out.printf("当前消息的详情是：%n主题名→%s,分区编号→%d,偏移量→%d,消息的value→%s,消息的key→%d%n%n",
                                topic, partition, offset, value, key
                        );
    
                    }
    
    
                    //所有的消息订阅完毕，就退出
                    if (records.isEmpty()) {
        
                        break;
                    }
                }
    
            } catch (Exception e) {
        
                e.printStackTrace();
            } finally {
        
                //⑥资源释放
                if (consumer != null) {
        
                    consumer.close();
                }
            }
    
    
        }
    }

[https_www.jianshu.com_p_4bf007885116]: https://www.jianshu.com/p/4bf007885116