Kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理领域。 订阅式模式：一对多的关系，一个生产者，数据存储在消息队列中，多个消费者均可从这个消息对列中获取数据，消费者消费数据之后不会清除消息。 一般都是从命令行和API两个方面进行讲解。 数据处理框架需要从数据的安全性以及效率两个方面深入了解。 在/opt/software目录下 在/opt/module/kafka目录下 /opt/module/kafk路径下 修改如下三个参数，修改后的值如下： 增加如下配置： 生效配置文件： 注意：分发之后记得配置其他机器的环境变量 /opt/module/kafka/config/server.properties中的broker.id=3、broker.id=4 注：broker.id不得重复 首先启动zookeeper集群和hadoop集群 依次在hadoop102、hadoop103、hadoop104节点上启动kafka 选项说明： 选项说明： 选项说明： 说明： 选项说明： 选项说明： –from-beginning ： 加上：会把topic中以往所有的数据都读取出来 不加：此时只会消费最新的数据，原来topic中的数据不会被消费 说明：此框架步步引导，采取提出问题解决问题的方式阐述。 情况1 ：指定了partition； 情况2 ：未指定partition，封装key，则按照key的hashcode % 分区数量 得出在哪个分区； 情况3：未指定partition，也未封装key处理方式 : 参数1：producer发送的数据量：batch.size，默认值为16Kb； 条件2：linger.ms：两条数据发送的间隔时间 t ，默认值为0s； 当发送的数据量 < batch.size 并且 发送的数据时间间隔 < t 时，所有的数据在一个分区； 当发送的数据量 > batch.size 或者 发送的数据时间间隔 > t 时，则数据会进入下一个分区； 数据传输流程： producer —–> server（kafka） ———>消费者 消费者从server中读取数据的方式有两种：pull （拉）和 push（推） pull ： consumer向server拉取数据 【主动】 push ：server向consumer推送数据【被动】 kafka默认使用pull，拉取数据的方式。因为kafka是一对多的关系，同一个组内的不同消费者的消费速率不同，所以不好设定推送的速率。 当出现拉取的数据为空时，consumer会等待一段时间之后再拉取数据，这段时长即为timeout 三种方式：roundrobin 、 range 、sticky roundrobin ： 轮询的方式 ，理解为洗牌，一张一张的发，分区一个一个轮询的方式分配给消费者； 缺点：当有新的消费者加进来时，所有的分区需要重新分配分区，基本上大多数的消费者的消费分区都会发生改变。 range：理解斗地主把牌按数量平均分配； 缺点：订阅的话题过多时，存在分区数量不均等的情况。 sticky：是在第一种方式的基础上进行改进，解决新增消费者情况的缺点，此时不再是所有消费者的分区进行重新分配，而是新进的消费者取之前所有消费者最后一次分区的数据进行消费。 由于consumer在消费过程中可能会出现断电宕机等故障，consumer恢复后，需要从故障前的位置的继续消费，所以consumer需要实时记录自己消费到了哪个offset，以便故障恢复后继续消费。 优点： 说明： 内存是分级别的，读写数据时数据先经过内核内存再经过。 如果是数据的写出操作，则数据经过内核内存以后就直接往外写出，不需经过用户内存，用户内存只是负责调度的功能，减少了 数据的传输过程，这个过程称为零拷贝。 流程：一个kafka集群，多个broker，一个zk集群 温馨提示 消费者每次拉取数据的最大值为：1M，（ 1048576字节） 模式一：自动提交，默认每5s提交一次； 模式二：手动提交，两种方式：commitSync（同步提交）、commitAsync（异步提交）； 同步和异步的异同点： kafka：数据的中转站，主要功能由topic体现； flume：数据的采集，通过source和sink体现。 官网配置文件： 官方配置文件： 作用：将数据拉去到kafka的topic中。 官方配置文件：
文章目录
Kafka总结
一、kafka概述
1.1 kafka定义
1.2 框架说明
1.3 Kafka涉及的关键词
1. producer: 消息的生产者，即为向kafka broker发消息; 2. broker ： kafka集群的节点； 3. topic : 队列（话题），生产者和消费者面向的都是一个topic； 4. message：消息，队列中的一条消息； 5. partition: 分区，为方便扩展和提高吞吐量，将一个topic分为了多个partition； 6. index ： 消息数据在log文件中的索引； 7. log ：消息的具体数据； 8. timeindex： 时间索引，代表发送的数据时间索引； 9. offset ： 消息的偏移量，每一条消息都对应一个offset； 10. segment : 一个分片数据； 11. leader ：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是leader； 12. follower : 每个分区多个副本中的“从”，实时从leader中同步数据，保持和leader数据的同步。leader发生故障时，某个follower会成为新的leader 二、Kafka安装
2.1 集群部署
2.2.1 解压安装包
tar -zxvf kafka_2.11-2.4.1.tgz -C /opt/module/ 2.2.2 修改解压后的文件名称
mv kafka_2.11-2.4.1/ kafka 2.2.3 创建logs文件夹
mkdir logs 2.2.4 修改配置文件
vim config/server.properties broker.id=2； log.dirs=/opt/module/kafka/logs zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka 2.2.5 配置环境变量
sudo vim /etc/profile.d/my_env.sh #KAFKA_HOME KAFKA_HOME=/opt/module/kafka PATH=$PATH:$KAFKA_HOME/bin export KAFKA_HOME source /etc/profile 2.2.6 分发安装包
xsync kafka/ 2.2.7 修改其他机器的配置文件
2.2.8 启动集群
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
2.2.9 关闭集群
kafka-server-stop.sh 2.2.10 kafka群起群停脚本
#！bin/bash if [ $# -lt 1 ] then echo "No Args Input Error" exit fi case $1 in "start") for i in `cat /opt/module/hadoop-3.1.3/etc/hadoop/workers` do echo "==========start $i kafka==========" ssh $i '$KAFKA_HOME/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties' done ;; "stop") for i in `cat /opt/module/hadoop-3.1.3/etc/hadoop/workers` do echo "==========stop $i kafka==========" ssh $i '$KAFKA_HOME/bin/kafka-server-stop.sh' done ;; *) echo "Input Args Error" ;; esac 2.2 Kafka命令操作
2.2.1 查看当前服务器中的所有topic
kafka-topics.sh --bootstrap-server hadoop102:9092 --list
2.2.2 创建topic
kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic first --partitions 2 --replication-factor 2
2.2.3 查看某个Topic的详情
kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first
2.2.4 修改分区数
kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 6
2.2.5 发送消息
kafka-console-producer.sh --broker-list hadoop102:9092,hadoop103:9092,hadoop104:9092 --topic first >hello world >atguigu atguigu
2.2.6 消费消息
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first
2.2.7 删除topic
kafka-topics.sh --bootstrap-server hadoop102:9820 --delete --topic first 三、 Kafka深入流程
3.1 Kafka工作流程及文件存储
-- 灵魂拷问1： 产生的数据一直向log中进行追加，会有什么问题呢？ -- 答案： log中的数据会越来越大，查询和读取效率会变慢。 -- 解决方式： 数据达到一定程度以后（默认值为1G：log.segment.bytes = 1G），log会进行数据切分，生成多个segment切分文件。 切分后的文件依然包含index、log、timeindex 。所以三个文件是作为一个整体的。 --切分机制 -- 灵魂拷问2： 现在假如有两个切分的文件，当有一个消费者需要消费一条消息（假如是 offset = 3），怎么知道这个消息在哪个切分文件中，以及真实数据如何查询？ --答案： 1）log和index文件名说明： -- 牢记log、index、timeindex是一个整体 index：00000000000000000000.index log：00000000000000000000.log 前面的数字00000000000000000000：代表此log文件中第一条消息的offset。 2） 查询的方式： 根据消费消息的offset值 -->找到指定的index文件 --> 匹配此条消息在log文件中数据的偏移量（即该数据在log文件中起始位置）--> 找到待消费的数据 -- 灵魂拷问3： 为什么kafka要采取向一个log文件中追加数据呢？ -- 答案： 1）减少IO； 2）消费数据是连续进行消费，连续读取数据的效率高。 3.2 Kafka之生产者producer
3.2.1 分区策略
3.2.2 数据可靠性保证
-- 灵魂拷问1： 如何保证从producer发送数据server的过程中数据不丢失？ -- 答案： server收到数据以后会回执，发送ack（acknowledgement确认收到）给producer，producer收到ack以后，则确定数据传送的过程中没有丢失。
-- 灵魂拷问2： 如何确保数据在server中能够被妥善保管呢？ -- 答案： server向producer回执ack的时机： 模式1：leader收到消息以后立即回复ack； 模式2：leader收到消息并存储在本地以后，立即回复ack； 模式3：leader收到消息后，所有follow从leader中拉取数据，当所有的follower完成存储以后，leader向producer回复ack。 说明：情况1/2/3是通过acks参数进行配置。 acks=0 -- 表示模式1 acks=1 -- 表示模式2 acks=-1或all --表示模式3 -- 默认情况下是模式2； =============================================================================== -- 灵魂拷问3： leader与follower副本数据同步策略是什么呢？ -- 答案 两种副本同步策略。 第一种：半数以上完成同步，就发送ack 第二种：全部完成同步，才发送ack kafka选择第二种，对应问题2的模式3. ================================================================================ -- 灵魂拷问4： kafka选择第二种副本同步策略会有哪些问题呢？ -- 答案： 问题1：follower同步leader的数据时，当某一个follower迟迟未向leader回复备份成功时，出现阻塞的状态； 问题2：当leader回执给producer的ack丢失时，producer因为没有收到来自leader的ack，则默认数据没有发送成功，会重新向集群发送未收到ack的消息，导致数据的重复。 -- 数据的重复指：同一条消息重复发送。 -- 那如何解决这两个问题呢？
规则：leader完成消息的读取和写出操作，follower定时向leader拉取数据。 1. leader维护了一个动态的in-sync replicat set (ISR) 同步副本的列表，说明：即使是follower也有可能不在isr列表中。 2.。只要在isr列表中所有的follower均告知leader副本备份完成以后，则leader向producer回执ack，则不受限于出现故障的follower，因为出现故障，就被移除isr列表中。 -- 问题1： 那么什么情况下follower不在isr列表呢？ -- 答案： 如果follower没有在规定的时间与leader保持同步，则leader会将该follower从isr中踢出，同步最大时间通过replica.lag.time.max.ms参数设定。 -- 问题2： 那么从isr中踢出的follower怎么重新回到isr中呢？ --故障处理机制 -- 答案： 每个消息在follower的log文件中有： 1) 真实数据 :消息的真实数据 2) LEO(log end offset) : 消息的最后偏移量 3) HW(High Watermark) ：ISR列表中follower最小的LEO（偏移量） 说明： 1）每个follower中的LEO可能是不一样的，因副本同步的快慢有差异； 2）leader中log的LEO是最大的，因为数据源源不断的发送过来，它的落盘速度是最快的； 3）HW之前的数据对consumer可见； 4）HW是一个动态的数据，当leader回执ack一次HW就会更新一次。 follower发生故障后会被临时踢出ISR，待该follower恢复后，follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取掉，从HW开始向leader进行同步。等该follower的LEO大于等于该Partition的HW，即follower追上leader之后，就可以重新加入ISR了。 -- 问题3： 当leader挂掉以后怎么办？ -- 答案： 1） 重新选举leader； 2） 从isr列表中的follower中选取； 3） 随机选择。
-- 在0.11之前的kafka版本： 在消费者端进行去重，在producer传输数据时，对消息增加唯一的全局主键，然后在消费端根据该主键进行去重。 该方式导致消费者组所有的消费者都需要进行去重操作，重复。 -- 在0.11版本之后引进了 Exactly Once （幂等性）来解决数据重复的问题 1） Exactly Once （幂等性） ： 做n次和做一次的效果是一样的，就是指Producer不论向Server发送多少次重复数据（重复发送同一条数据），Server端都只会持久化一条，在server端完成去重操作。 2） 幂等性实现过程 初始化数据时，给消息分配一个pid，发往同一个分区的消息会附带sequence Number,broker端会将<pid,partition,sequence Number>和消息的真实数据一起存储到log文件中，当具有相同主键的消息提交时，Broker只会持久化一条。 -- 何为主键？ 由<pid,partition,sequence Number>三个参数构成的集合。重复发送的数据，这三个值不会变，数据是否重复与数据的内容无关，而是指为同一条数据多次发送。 -- 总结：重发的消息的主键是不会改变的，新发的消息seqnumber就会变化。 例如：消息A与消息B的数据内容完全一致 producer向集群发送消息A，集群收到以后返回的ack丢失，则消息A会被再次发送一次，此时消息A的主键是和第一次发送时相同，则集群认为数据是重复，不会进行存储； producer向集群发送消息B，虽然与消息A的数据相同，但是seqnumber是不同的，所以不是重复的数据，集群会进行数据存储。 说明： 1） sequence Number ：消息序列号，发往同一Partition的消息会附带Sequence Number，表示该producer向该分区发送的第几次消息； 2) pid : 生产者的id； 3) partition ： 分区号； 4） PID重启就会变化，同时不同的Partition也具有不同主键，所以幂等性无法保证跨分区跨会话的Exactly Once； 5） 开启幂等性会降低kafka的性能； 6） 幂等性的底层原理也还是通过给消息增加全局的唯一主键的方式； 7） 开启幂等性参数：enable.idompotence设置为true即可。 3.3 Kafka之消费者 consumer
3.3.1 消费模式
3.3.2 分区分配策略
3.3.3 offset的维护
Kafka 0.9版本之前，consumer默认将offset保存在Zookeeper中，从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets。-- 问题： 为什么要将offset从zookeeper中转移到kafka中？ -- 回答： zookeeper中维护offset的效率对于Kafka来说，不可控的，Kafka不能通过修改自己的代码来提升zookeeper维护offset的效率，所以将offset的维护迁移到kafka的会话中。 3.4 Kafka高效读写数据
3.4.1 顺序写磁盘
-- 问题1： kafka的producer生产的数据最终按照顺序存储到磁盘上，写入到磁盘中数据过程不是很慢吗？ -- 回答： 1） 多分区存储模式：kafka是采用多分区的存储方式，提高了高并发； 2） 顺序写模式：按照顺序写的速度能够减少大量磁头询地址的时间，使写数据速度和网络传输速度相当，所以基本上够用，但是还是比内存数据传输的速度要慢。 3.4.2 应用Pagecache
-- 说明： Pagecache(网页缓存)：是操作系统实现的一个功能，因为linux系统兼容这个功能，所以kafka能够使用，解决大量随机读写的过程。 -- 内存： 我们常说的内存可以分为两个模块，一是提供给系统的内核使用，此部分对于用户是不可见的，不能被用户使用，二是供用户使用的内存。 -- 原理： pagechache是在内核内存中开辟的一个内存空间，producer生产的数据，先会存储在该内存中，待达到一定的数据量以后，再统一进行落盘，当消费者消费的速率和生产者生产的速率相同时，读写的效率是最高的，因为此时生产的数据不需要落盘处理，consumer直接从内存中读取数据。 -- 交换区和pagecache的区别： 交换区：将磁盘当做内存使用； pagecache：将内存当做磁盘使用； 恰好是两个相反的过程。 -- 假如pagecache挂掉了怎么办？内存中的数据不是丢失了吗？ 首先当发生这个问题时，是不能够完全保证数据一定不丢失，但是由于kafka具有副本策略，所以有一定保证的。 1） I/O Scheduler 会将连续的小块写组装成大块的物理写从而提高性能 2） I/O Scheduler 会尝试将一些写操作重新按顺序排好，从而减少磁盘头的移动时间 3） 充分利用所有空闲内存（非 JVM 内存）。如果使用应用层 Cache（即 JVM 堆内存），会增加 GC 负担 4） 读操作可直接在 Page Cache 内进行。如果消费和生产速度相当，甚至不需要通过物理磁盘（直接通过 Page Cache）交换数据 5） 如果进程重启，JVM 内的 Cache 会失效，但 Page Cache 仍然可用 3.4.3 零拷贝技术
3.5 zookeeper在kafka中的作用
-- 步骤： 1） 首先所有的broker会竞选一个controller（随机竞选，谁厉害谁上），责管理集群broker的上下线，所有topic的分区副本分配和leader选举等工作； 2） 所有的broker将自己的id信息注册到zk集群的节点上； 3） controller监控zk的这个信息； 4） controller负责broker的leader选举工作； 5） broker将状态信息注册到zk集群上； 6） 此时分区的leader故障以后，controller从zk集群中获取isr中的follower信息，请负责从isr中follower选举出一个新的leader； 7） controller更新zk集群上broker的状态信息。 -- 假如故障的leader恰好也是controller怎么办？ 先从现存的follower中重新选举controller，再执行1-5步。 
3.6 Kafka事务
--问题： 事务用来解决什么问题？ --回答： kafka使用Exactly Once解决producer端生产数据重复的问题存在什么问题？ 问题1：不能跨分区； 问题2：producer不能重启pid。 则事务就是来解决上面问题的，事务可以保证Kafka在Exactly Once语义的基础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败。 -- 那具体是怎么做的呢？ 3.6.1 producer事务
--解决producer重启问题： 1) 引进全局唯一的Transaction ID，将producer的pid与Transaction ID进行绑定。当重启producer时，可以通过正在进行的Transaction ID获得原来的PID. 2) 为了管理Transaction，Kafka引入了一个新的组件Transaction Coordinator。Producer就是通过和Transaction Coordinator交互获得Transaction ID对应的任务状态。Transaction Coordinator还负责将事务所有写入Kafka的一个内部Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。 3.6.2 Consumer事务（精准一次性消费）
kafak对consumer事务的保证是非常弱的，尤其无法保证Commit的信息被精确消费。这是由于Consumer可以通过offset访问任意信息，而且不同的Segment File生命周期不同，同一事务的消息可能会出现重启后被删除的情况 四、 Kafka API
api的步骤： 第一步： new 对象; 第二步： 具体的操作; 第三步： 关闭资源。 -- 不知道要写哪些参数？不知道参数的意义？不知道参数取值？怎么办？ 请认准kafka官网：https://kafka.apache.org/documentation/ producer API ： 找Producer Configs consumer API： 找Consumer Configs 4.1 Producer API
4.1.1 消息发送流程
-- 问题： Kafka的Producer发送消息采用的是异步发送的方式，这种方式优点和缺点是什么呢？ -- 回答： 优点：高效率，生产者只要一直生产数据就可以，不需要等到ack回执后再进行生产数据； 缺点：不能实时知道数据是否发送成功，不过有ack机制和幂等性机制（保证数据的准确性）。 -- 发送数据的流程： 两线程一共享变量： 1. main线程：将消息发送给RecordAccumulator 2. Sender线程：Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker 3. 线程共享变量——RecordAccumulator：数据临时存储器。 第一步：生产者首先将数据包装成ProducerRecord 第二步：main线程中有一个send方法，producer将这个对象interceptors【拦截器】处理； 第三步：interceptors处理好后将数据传递给【序列化器】，将数据序列化； -- 在producer端序列化 第四步：将序列化好的数据传递给【分区器】，对数据进行分区； -- 在producer端序列化 第五步：将数据传递到内存的数据缓存区，在这里面，话题有多少个分区，在缓存区里面就有多少个分区，一一对应，对应的分区数据就会去到对应的缓存区的分区中； -- 此时的数据是已经分好区了，同时也是已经序列化，此时producer就不再管这里的数据了； 第六步：Sender线程就将数据发送给topic中的分区中。 -- 此时的数据，Sender线程是怎么向topic中发的呢？ batch.size：只有数据积累到batch.size之后，sender才会发送数据。 linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。 
4.1.2 异步发送API
package kafkaproducer; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; /** * @author lianzhipeng * @Description * @create 2020-05-08 14:58 */ public class Producer { public static void main(String[] args) throws ExecutionException, InterruptedException { //1.new 对象 Properties properties = new Properties(); properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); properties.setProperty("acks", "all"); properties.setProperty("bootstrap.servers", "hadoop102:9092"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties); //2.具体的操作 for (int i = 0; i < 100; i++) { Future<RecordMetadata> result = producer.send(new ProducerRecord<String, String>( "first", "Message" + i, "这是第" + i + "条信息" ), new Callback() {//回调函数，当producer发送的数据完成以后，返回告诉producer数据发送成功 public void onCompletion(RecordMetadata metadata, Exception exception) { int partition = metadata.partition(); String topic = metadata.topic(); long offset = metadata.offset(); System.out.println( topic + "话题" + partition + "分区" + offset + "消息发送成功"); } }); /* 如下一行代码产生同步回调和异同回调两种方式： 同步回调：加了此行代码，生产者收到ack以后再发第二条消息；类似打电话 异步回调：未加此行代码，生成者只要一直发送消息既可。类似发短信 */ RecordMetadata recordMetadata = result.get(); System.out.println("第" + i + "条消息发送结束"); } //3.关闭资源，资源关闭的时候会调用回调函数 producer.close(); } } 4.2 Consumer API
4.2.1 数据漏消费和重复消费
-- 问题1 数据漏消费 什么时候会出现数据漏消费呢？ 先提交offset后消费。 例如：消费者从kafka集群中获取了数据，此数据在消费的过程中出现故障延迟最后宕机，在故障期间offset已经提交至kafka集群，此时实际上数据并没有被使用，但是kafka集群上该消费者消费的数据偏移量已经更新了，重启消费者时，上一条数据不能被消费了，导致数据漏消费。 -- 问题2 数据重复消费 什么时候会出现数据库重复消费呢？ 当数据已经被消费以后，此时返回的offset时消费者出现了故障，则kafka集群中的_consumer_offset会话保存的offset则为上一次的数据，offset没有被更新，当消费者重新启动时，上一条数据则会被重新再消费一次。
-- 相同点： 提交本次poll的一批数据最高的偏移量. -- 不同点： commitSync（同步提交）：提交offset时，commitSync阻塞当前线程，一直到提交成功，并且会自动失败重试（（由不可控因素导致，也会出现提交失败））； commitAsync（异步提交）：则没有失败重试机制，故有可能提交失败。commitAsync（异步提交）：则没有失败重试机制，故有可能提交失败。 4.2.2 几个重要的参数
Type:
int
Default:
5000
Valid Values:
[0,…]
Importance:
low
Type:
string
Default:
latest
Valid Values:
[latest, earliest, none]
Importance:
medium
Type:
int
Default:
1048576
Valid Values:
[0,…]
Importance:
high
4.2.3 代码
package kafkaconsumer; import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import java.time.Duration; import java.util.Collections; import java.util.Map; import java.util.Properties; /** * @author lianzhipeng * @Description * @create 2020-05-08 21:04 */ public class MyConsumer { public static void main(String[] args) { //1 new 对象 Properties properties = new Properties(); properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); properties.setProperty("bootstrap.servers", "hadoop102:9092"); properties.setProperty("group.id", "group9"); properties.setProperty("auto.offset.reset", "earliest"); //自动提交offset properties.setProperty("enable.auto.commit","false"); KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties); //2 操作 //连接话题 consumer.subscribe(Collections.singleton("first")); //拉取数据 Duration duration = Duration.ofMillis(500); while (true){ ConsumerRecords<String, String> records = consumer.poll(duration); for (ConsumerRecord<String, String> record : records) { System.out.println(record); } //手动同步提交 // consumer.commitSync(); //手动异步提交 consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { offsets.forEach( (t, o) -> { System.out.println("分区：" + t + "nOffset：" + o); } ); } }); } //3 关闭资源 // consumer.close(); } }
//手动异步提交方式，形参里面为回调对象。 consumer.commitAsync(new OffsetCommitCallback() { /* 回调方式，当消费成功以后调用此方法并进行打印 */ @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { offsets.forEach( (t, o) -> { System.out.println("分区：" + t + "nOffset：" + o); } ); } }); 五、Kafka监控（Kafka Eagle）
修改kafka-server-start.sh命令中 --原文： if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" fi --改为： if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70" export JMX_PORT="9999" #export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" fi --注意：修改之后在启动Kafka之前要分发之其他节点
[atguigu@hadoop102 software]$ tar -zxvf kafka-eagle-bin-1.3.7.tar.gz
[atguigu@hadoop102 kafka-eagle-bin-1.3.7] $ tar -zxvf kafka-eagle-web-1.4.5-bin.tar.gz -C /opt/module/
[atguigu@hadoop102 module]$ mv kafka-eagle-we-1.4.5/ eagle
[atguigu@hadoop102 bin]$ chmod 777 ke.sh
###################################### # multi zookeeper&kafka cluster list ###################################### kafka.eagle.zk.cluster.alias=cluster1 cluster1.zk.list=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka ###################################### # kafka offset storage ###################################### cluster1.kafka.eagle.offset.storage=kafka ###################################### # kafka metrics, 30 days by default ###################################### kafka.eagle.metrics.charts=true kafka.eagle.metrics.retain=30 ###################################### # kafka sqlite jdbc driver address ###################################### kafka.eagle.driver=com.mysql.jdbc.Driver kafka.eagle.url=jdbc:mysql://hadoop102:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull kafka.eagle.username=root kafka.eagle.password=123456
export KE_HOME=/opt/module/eagle export PATH=$PATH:$KE_HOME/bin -- 注意：source /etc/profile
[atguigu@hadoop102 eagle]$ bin/ke.sh start ... ... ... ... ******************************************************************* * Kafka Eagle Service has started success. * Welcome, Now you can visit 'https://192.168.9.102:8048/ke' -- 这个网址就是登入的eagle的网址 * Account:admin ,Password:123456 -- 这是登入的密码 ******************************************************************* * <Usage> ke.sh [start|status|stop|restart|stats] </Usage> * <Usage> https://www.kafka-eagle.org/ </Usage> ******************************************************************* --注意：启动之前需要先启动ZK以及KAFKA
网址： http://192.168.9.102:8048/ke 账号： admin 密码： 123456 六、面试题
6.1 Kafka中的ISR、AR代表什么
ISR:分区leader维护的一个follower列表，在isr中的follower与leader同步。 AR:分区的所有副本。 6.2 Kafka中的HW、LEO等分别代表什么
LEO: leader维护的isr中所有follower的最后偏移量。 HW：所有followerleo最小的值。 6.3 Kafka中是怎么体现消息顺序性的
每次生产的数据是在一个上次生产数据的基础上追加，同时存储了消息的offset和数据的index索引，减少了数据存储时的磁头寻址的过程。 6.4 Kafka中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？
处理顺序： 拦截器 --> 序列化器 --> 分区器 拦截器：对数据进行简单处理，加一些标识。 序列化：对数据进行序列化，保证数据可用于传输； 分区器：给数据加上分区标签，指定数据应该去到哪个kafka集群中的分区。 以上三步骤均在producer端就完成了。 6.5 Kafka生产者客户端的整体结构是什么样子的？使用了几个线程来处理？分别是什么？
一共2个线程，一个数据缓存区。 线程 main线程：负责对数据进行包装、序列化、分区。 sender线程：负责将数据从数据缓冲区发送topic话题中。 6.6 消费者组中的消费者个数如果超过topic的分区，那么就会有消费者消费不到数据这句话是否正确
是的，正确。 了解一下分区分配的策略。 三种方式：roundrobin 、 range 、sticky。 6.7 消费者提交消费位移时提交的是当前消费到的最新消息的offset还是offset+1？
offset + 1 。 6.8 有哪些情形会造成重复消费？
先消费后提交offset。 6.9 那些情景会造成消息漏消费？
先提交offset后消费。 6.10 当你使用kafka-topics.sh创建（删除）了一个topic之后，Kafka背后会执行什么逻辑？
了解producer发送数据的过程。 6.11 topic的分区数可不可以增加？如果可以怎么增加？如果不可以，那又是为什么？
可以增加。 6.12 topic的分区数可不可以减少？如果可以怎么减少？如果不可以，那又是为什么？
不能减少，因为原分区中的数据没有地方去。 6.13 Kafka有内部的topic吗？如果有是什么？有什么作用？
会话：_consumer_offset，保存consumer消费的偏移量。 6.14 Kafka分区分配的概念？
一共有三种分区分配的策略。 三种方式： 1）roundrobin ： 轮询分配。 2）range ： 平均分配。 3）sticky ： 轮询分配 + 解决新增消费者的优化。 6.15 简述Kafka的日志目录结构？
一共有4个文件 1）log文件：记录真实数据，内部包含了真实数据 + hw + leo。 2）index文件 ： 存储消息的偏移量。 3) timeindex文件 ： 存储下消息的时间偏移量。 6.16 如果我指定了一个offset，Kafka Controller怎么查找到对应的消息？
通过offset，消息的偏移量，通过日志目录的文件顺序号，根据区间范围找到消息所在的inde和log目录。 其次根据在index表中的消息偏移量找到真实数据在log文件中该消息的起始索引位置。 6.17聊一聊Kafka Controller的作用？
1、负责leader的选举； 2、负责监控leader的状态； 3.负责更新集群在zookeeper中的状态。 6.18 Kafka中有那些地方需要选举？这些地方的选举策略又有哪些？
1.每个分区的leader选举；(isr)； 2.controller的选举（先到先得）。 6.19 失效副本是指什么？有那些应对措施？
follower不能与leader进行同步数据，暂时被leader踢出isr列表中。通过followe故障恢复重新备份，当leo达到了isr中的hw时，又重新会回到isr的列表中。 6.20 Kafka的那些设计让它有如此高的性能？
1. pagecache； 2.顺序读写机制； 3.零拷贝技术。 七 、flume与kafka融合技术
7.1 kafka source
-- 问题 ： fulme在kafka中的作用 -- 答案： 消费者 tier1.sources.source1.type = org.apache.flume.source.kafka.KafkaSource --source类型 tier1.sources.source1.channels = channel1 tier1.sources.source1.batchSize = 5000 tier1.sources.source1.batchDurationMillis = 2000 tier1.sources.source1.kafka.bootstrap.servers = localhost:9092 -- kafka的集群 tier1.sources.source1.kafka.topics = test1, test2 -- 订阅的话题 tier1.sources.source1.kafka.consumer.group.id = custom.g.id --消费者组 7.2 kakfa channel
-- 情况一： 有Flume source and sink -- 缓冲区 kakfa channel为事件提供了可靠且高可用的通道； -- 情况二： 有source and interceptor but no sink --生产者 it allows writing Flume events into a Kafka topic, for use by other app -- 情况三： 有 sink, but no source --消费者 it is a low-latency, fault tolerant way to send events from Kafka to Flume sinks such as HDFS, HBase or Solr a1.channels.channel1.type = org.apache.flume.channel.kafka.KafkaChannel a1.channels.channel1.kafka.bootstrap.servers = kafka-1:9092,kafka-2:9092,kafka-3:9092 a1.channels.channel1.kafka.topic = channel1 a1.channels.channel1.kafka.consumer.group.id = flume-consumer 7.3 kafka sink
a1.sinks.k1.channel = c1 a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink a1.sinks.k1.kafka.topic = mytopic a1.sinks.k1.kafka.bootstrap.servers = localhost:9092 a1.sinks.k1.kafka.flumeBatchSize = 20 a1.sinks.k1.kafka.producer.acks = 1 a1.sinks.k1.kafka.producer.linger.ms = 1 a1.sinks.k1.kafka.producer.compression.type = snappy
本网页所有视频内容由 imoviebox边看边下-网页视频下载, iurlBox网页地址收藏管理器 下载并得到。
ImovieBox网页视频下载器 下载地址: ImovieBox网页视频下载器-最新版本下载
本文章由: imapbox邮箱云存储,邮箱网盘,ImageBox 图片批量下载器,网页图片批量下载专家,网页图片批量下载器,获取到文章图片,imoviebox网页视频批量下载器,下载视频内容,为您提供.
阅读和此文章类似的: 全球云计算
官方软件产品操作指南 (170)
