- 1Charles-代理工具_charles proxy
- 2Python | Package | Python的三种包安装方式（pip/whl/tar.gz）_pip tar.gz
- 32023年项目管理软件排行榜发布_中科院《互联网周刊》2023上半年排行
- 42024广东省职业技能大赛云计算赛项实战——Minio服务搭建
- 5Android Studio 模仿美团界面_android仿美团外卖首页
- 6【机器学习】聚类代码练习
- 7撤销git commit的方法_vs撤回commit
- 8谈谈 tcp 慢启动与拥塞控制_tcp慢启动
- 9IntelliJ IDEA自动部署项目至远程服务器与传统部署项目至远程服务器的区别_idea本地文件与远程服务器文件比较
- 10Mobile ALOHA——斯坦福炒虾机器人的关键技术：动作分块算法ACT的原理解析_mobile aloha关键技术探讨
KAFKA_kafka手动ack
赞
踩
【KAFKA】
名词解释
（1）Producer：消息生产者，就是向 Kafka broker 发消息的客户端。
（2）Consumer：消息消费者，向 Kafka broker 取消息的客户端。
（3）Consumer Group（CG）：消费者组，由多个 consumer 组成。消费者组内每个消
费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不
影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
（4）Broker：一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个
broker 可以容纳多个 topic。
（5）Topic：可以理解为一个队列，生产者和消费者面向的都是一个 topic。
（6）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服
务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。
（7）Replica：副本。一个 topic 的每个分区都有若干个副本，一个 Leader 和若干个
Follower。
（8）Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数
据的对象都是 Leader。
（9）Follower：每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和
Leader 数据的同步。Leader 发生故障时，某个 Follower 会成为新的 Leader
MessageQueue是什么
消息队列是一种在分布式和大数据开发中不可或缺的中间件。在分布式开发或者大数据开发中通常使用消息队列进行缓冲、系统间解耦和削峰填谷等业务场景，常见的消息队列工作模式大致会分为两大类：
- 至多一次：消息生产者将数据写入消息系统，然后由消费者负责去拉去消息服务器中的消息，一旦消息被确认消费之后 ，由消息服务器主动删除队列中的数据，这种消费方式一般只允许被一个消费者消费，并且消息队列中的数据不允许被重复消费。
- 没有限制:同上诉消费形式不同，生产者发不完数据以后，该消息可以被多个消费者同时消费，并且同一个消费者可以多次消费消息服务器中的同一个记录。主要是因为消息服务器一般可以长时间存储海量消息。
Kafka 基础架构
Kafka集群以Topic形式负责分类集群中的Record每一个Record属于一个Topic。每个Topic底层都会对应一组分区的日志用于持久化Topic中的Record。同时在Kafka集群中，Topic的每一个日志的分区都一定会有1个Borker担当该分区的Leader，其他的Broker担当该分区的follower，Leader负责分区数据的读写操作，follower负责同步改分区的数据。这样如果分区的Leader宕机，改分区的其他follower会选取出新的leader继续负责该分区数据的读写。其中集群的中Leader的监控和Topic的部分元数据是存储在Zookeeper中.
Topics and Logs
Kafka中所有消息是通过Topic为单位进行管理，每个Kafka中的Topic通常会有多个订阅者，负责订阅发送到改Topic中的数据。Kafka负责管理集群中每个Topic的一组日志分区数据。
生产者将数据发布到相应的Topic。负责选择将哪个记录分发送到Topic中的哪个Partition。例如可以round-robin方式完成此操作，然而这种仅是为了平衡负载。也可以根据某些语义分区功能（例如基于记录中的Key）进行此操作。
每组日志分区是一个有序的不可变的的日志序列，分区中的每一个Record都被分配了唯一的序列编号称为是offset，Kafka 集群会持久化所有发布到Topic中的Record信息，改Record的持久化时间是通过配置文件指定,默认是168小时。
log.retention.hours=168
Kafka底层会定期的检查日志文件，然后将过期的数据从log中移除，由于Kafka使用硬盘存储日志文件，因此使用Kafka长时间缓存一些日志文件是不存在问题的。
在消费者消费Topic中数据的时候，每个消费者会维护本次消费对应分区的偏移量，消费者会在消费完一个批次的数据之后，会将本次消费的偏移量提交给Kafka集群，因此对于每个消费者而言可以随意的控制改消费者的偏移量。因此在Kafka中，消费者可以从一个topic分区中的任意位置读取队列数据，由于每个消费者控制了自己的消费的偏移量，因此多个消费者之间彼此相互独立。
Kafka中对Topic实现日志分区的有以下目的：
- 首先，它们允许日志扩展到超出单个服务器所能容纳的大小。每个单独的分区都必须适合托管它的服务器，但是一个Topic可能有很多分区，因此它可以处理任意数量的数据。
- 其次每个服务器充当其某些分区的Leader，也可能充当其他分区的Follwer，因此群集中的负载得到了很好的平衡。
生产者&消费者
消费者使用Consumer Group名称标记自己，并且发布到Topic的每条记录都会传递到每个订阅Consumer Group中的一个消费者实例。如果所有Consumer实例都具有相同的Consumer Group，那么Topic中的记录会在改ConsumerGroup中的Consumer实例进行均分消费；如果所有Consumer实例具有不同的ConsumerGroup，则每条记录将广播到所有Consumer Group进程。
更常见的是，我们发现Topic具有少量的Consumer Group，每个Consumer Group可以理解为一个“逻辑的订阅者”。每个Consumer Group均由许多Consumer实例组成，以实现可伸缩性和容错能力。这无非就是发布-订阅模型，其中订阅者是消费者的集群而不是单个进程。这种消费方式Kafka会将Topic按照分区的方式均分给一个Consumer Group下的实例，如果ConsumerGroup下有新的成员介入，则新介入的Consumer实例会去接管ConsumerGroup内其他消费者负责的某些分区，同样如果一下ConsumerGroup下的有其他Consumer实例宕机，则由改ConsumerGroup其他实例接管。
由于Kafka的Topic的分区策略，因此Kafka仅提供分区中记录的有序性，也就意味着相同Topic的不同分区记录之间无顺序。因为针对于绝大多数的大数据应用和使用场景， 使用分区内部有序或者使用key进行分区策略已经足够满足绝大多数应用场景。但是，如果您需要记录全局有序，则可以通过只有一个分区Topic来实现，尽管这将意味着每个ConsumerGroup只有一个Consumer进程。
高性能之道-顺序写&mmap
Kafka的特性之一就是高吞吐率，但是Kafka的消息是保存或缓存在磁盘上的，一般认为在磁盘上读写数据是会降低性能的，但是Kafka即使是普通的服务器，Kafka也可以轻松支持每秒百万级的写入请求，超过了大部分的消息中间件，这种特性也使得Kafka在日志处理等海量数据场景广泛应用。Kafka会把收到的消息都写入到硬盘中，防止丢失数据。为了优化写入速度Kafka采用了两个技术顺序写入和MMFile 。
因为硬盘是机械结构，每次读写都会寻址->写入，其中寻址是一个“机械动作”，它是最耗时的。所以硬盘最讨厌随机I/O，最喜欢顺序I/O。为了提高读写硬盘的速度，Kafka就是使用顺序I/O。这样省去了大量的内存开销以及节省了IO寻址的时间。但是单纯的使用顺序写入，Kafka的写入性能也不可能和内存进行对比，因此Kafka的数据并不是实时的写入硬盘中 。
Kafka充分利用了现代操作系统分页存储来利用内存提高I/O效率。Memory Mapped Files(后面简称mmap)也称为内存映射文件，在64位操作系统中一般可以表示20G的数据文件，它的工作原理是直接利用操作系统的Page实现文件到物理内存的直接映射。完成MMP映射后，用户对内存的所有操作会被操作系统自动的刷新到磁盘上，极大地降低了IO使用率。
网络IO
一般方案
1、文件在磁盘中数据被copy到内核缓冲区
2、从内核缓冲区copy到用户缓冲区
3、用户缓冲区copy到内核与socket相关的缓冲区。
4、数据从socket缓冲区copy到相关协议引擎发送出去
Zero拷贝
1、文件在磁盘中数据被copy到内核缓冲区
2、从内核缓冲区copy到内核与socket相关的缓冲区。
3、数据从socket缓冲区copy到相关协议引擎发送出去
第二章: 安装&部署
环境搭建-单机
u安装JDK1.8+，配置JAVA_HOME (CentOS 6.10 64bit)
u配置主机名和IP映射
u关闭防火墙&防火墙开机自启动
u安装&启动Zookeeper
u安装&启动|关闭Kafka
环境搭建-集群
u安装JDK，配置JAVA_HOME (CentOS 6.10 64bit)
u配置主机名和IP映射
u关闭防火墙&防火墙开机自启动
u同步时钟 ntpdate cn.pool.ntp.org | ntp[1-7].aliyun.com
u安装&启动Zookeeper
u安装&启动|关闭Kafka
启动
启动zookeeper
进入bin文件夹
# cd /usr/local/zookeeper/bin
# ./zkServer.sh start
- 1
- 2
通过status查看启动状态。稍微有个等待时间
# ./zkServer.sh status
启动kafka
启动kafka命令
cd /usr/local/kafka_2.2.0/
bin/kafka-server-start.sh -daemon config/server.properties
- 1
- 2
关闭命令
bin/kafka-server-stop.sh
Topic管理
创建
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --create --topic topic02 --partitions 3 --replication-factor 3
- 1
查看
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --list
- 1
详情
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --describe --topic topic01
Topic:topic01 PartitionCount:3 ReplicationFactor:3 Configs:segment.bytes=1073741824
Topic: topic01 Partition: 0 Leader: 0 Replicas: 0,2,3 Isr: 0,2,3
Topic: topic01 Partition: 1 Leader: 2 Replicas: 2,3,0 Isr: 2,3,0
Topic: topic01 Partition: 2 Leader: 0 Replicas: 3,0,2 Isr: 0,2,3
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
修改
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --create --topic topic03 --partitions 1 --replication-factor 1
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --alter --topic topic03 --partitions 2
- 1
- 2
- 3
删除
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --delete --topic topic03
- 1
订阅
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic topic01 --group g1 --property print.key=true --property print.value=true --property key.separator=,
- 1
生产
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic topic01
- 1
消费组
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --list g1
[root@node1 kafka_2.11-2.2.0]# ./bin/kafka-consumer-groups.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --describe --group g1
TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
topic01 1 0 0 0 consumer-1-** /192.168.52.130 consumer-1
topic01 0 0 0 0 consumer-1-** /192.168.52.130 consumer-1
topic01 2 1 1 0 consumer-1-** /192.168.52.130 consumer-1
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
第三章: Kafka 基础API
1、Topic基本操作 DML管理
2、生产者
3、消费者 sub/assign
4、自定义分区
5、序列化
6、拦截器
POM
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.2.0</version> </dependency> <!-- https://mvnrepository.com/artifact/log4j/log4j --> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.25</version> </dependency> <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.25</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.9</version> </dependency>
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
日志文件 log4j.properties
log4j.rootLogger = info,console
log4j.appender.console = org.apache.log4j.ConsoleAppender
log4j.appender.console.Target = System.out
log4j.appender.console.layout = org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern = %p %d{yyyy-MM-dd HH:mm:ss} %c - %m%n
- 1
- 2
- 3
- 4
- 5
- 6
- 7
1、Topic基本操作 DML管理
需在本地配上 主机和IP的映射
192.168.1.111 node1
192.168.1.112 node2
192.168.1.113 node3
C:\Windows\System32\drivers\etc
public class TopicDMLDemo { public static void main(String[] args) throws ExecutionException, InterruptedException { //配置连接参数 Properties props = new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092,node2:9092,node3:9092"); KafkaAdminClient adminClient= (KafkaAdminClient) KafkaAdminClient.create(props); //查询topics KafkaFuture<Set<String>> nameFutures = adminClient.listTopics().names(); for (String name : nameFutures.get()) { System.out.println(name); } //创建Topics List<NewTopic> newTopics = Arrays.asList(new NewTopic("topic02", 2, (short) 3)); createTopicsResult.all().get();//同步创建 //删除Topic // adminClient.deleteTopics(Arrays.asList("topic02")); //查看Topic详情 DescribeTopicsResult describeTopics = adminClient.describeTopics(Arrays.asList("topic01")); Map<String, TopicDescription> tdm = describeTopics.all().get(); for (Map.Entry<String, TopicDescription> entry : tdm.entrySet()) { System.out.println(entry.getKey()+"\t"+entry.getValue()); } //关闭连接 adminClient.close(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
2、生产者
生产者端
public class ProducerQuickStart { public static void main(String[] args) { //1、创建生产者 Properties prop = new Properties(); prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); KafkaProducer<String, String> producer = new KafkaProducer<>(prop); //2、发送消息 for (int i = 0; i < 10; i++) { ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "key"+i, "value"+i); producer.send(record); } //关闭 producer.close(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
3、消费者 sub/assign
subscribe订阅模式下，必须设置消费者组
只能保证分区有序性
subscribe 模式 public class ConsumerQuickStart1 { public static void main(String[] args) { //1.创建Kafka链接参数 Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,"g2");//kafka中消费者通常以组的方式消费 //2.创建Topic消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop); //3.订阅topic开头的消息队列 consumer.subscribe(Pattern.compile("^topic.*")); while (true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(1));//每隔一秒取数据 //有数据 while (true){ ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, String> record = recordIterator.next(); String key = record.key(); String value = record.value(); long offset = record.offset(); int partition = record.partition(); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } } } assign 模式 public class ConsumerQuickStart1 { public static void main(String[] args) { //1.创建Kafka链接参数 Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,"g2");//kafka中消费者通常以组的方式消费 //2.创建Topic消费者 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop); //3. 订阅相关的topics，手动指定消费分区，失去组管理特性 List<TopicPartition> partitions = Arrays.asList(new TopicPartition("topic01", 0)); //3.订阅topic开头的消息队列 consumer.assign(partitions); //指定消费分区的位置 consumer.seekToBeginning(partitions); while (true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(1));//每隔一秒取数据 //有数据 while (true){ ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, String> record = recordIterator.next(); String key = record.key(); String value = record.value(); long offset = record.offset(); int partition = record.partition(); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
4、自定义分区
默认分区策略：
1、如果指定策略，会按照我们的策略 比如设置了发送给分区1
2、如果没指定策略，如果有key 会按hash key的方式选择分区
3、如果也没有key ，会采用round-robin 的方式轮训
如果需要自定义分区 ，需要实现Partitioner接口，
configure() ：加载配置项
partition()：核心处理逻辑
public class UserDefinePartitioner implements Partitioner { private AtomicInteger atomicInteger=new AtomicInteger(0); @Override public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { //获取所有分区 int numPartitions = cluster.partitionsForTopic(topic).size(); if(keyBytes==null || keyBytes.length==0){ int andIncrement = counter.getAndIncrement(); //与上int最大值，保证是个正数 return (andIncrement & Integer.MAX_VALUE) % numPartitions; } else { return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions; } } @Override public void close() { System.out.println("close"); } @Override public void configure(Map<String, ?> configs) { System.out.println("configure"); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
5、自定义序列化器
实现Serializer 接口
public class ObjectSerializer implements Serializer<Object> { @Override public void configure(Map<String, ?> configs, boolean isKey) { System.out.println("configure"); } @Override public byte[] serialize(String topic, Object data) { return SerializationUtils.serialize((Serializable) data); } @Override public void close() { System.out.println("close"); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
生产者
public class ProducerQuickStart { public static void main(String[] args) { //1、创建生产者 Properties prop = new Properties(); prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ObjectSerializer.class.getName()); KafkaProducer<String, User> producer = new KafkaProducer<>(prop); //2、发送消息 for (int i = 0; i < 10; i++) { ProducerRecord<String, User> record = new ProducerRecord<>("topic02", "key"+i, new User(i,"name"+i,new Date())); producer.send(record); } //关闭 producer.close(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
消费者
public class ConsumerQuickStart { public static void main(String[] args) { //1.创建Kafka链接参数 Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ObjectDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,"g2");//kafka中消费者通常以组的方式消费 //2.创建Topic消费者 KafkaConsumer<String, User> consumer = new KafkaConsumer<>(prop); //3.订阅topic开头的消息队列 consumer.subscribe(Arrays.asList("topic02")); while (true){ ConsumerRecords<String, User> records = consumer.poll(Duration.ofNanos(1));//每隔一秒取数据 //有数据 while (true){ ConsumerRecords<String, User> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, User>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, User> record = recordIterator.next(); String key = record.key(); User value = record.value(); long offset = record.offset(); int partition = record.partition(); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
6、自定义拦截器
实现ProducerInterceptor 接口
public class UserDefineInterceptor implements ProducerInterceptor { @Override public ProducerRecord onSend(ProducerRecord producerRecord) { ProducerRecord wrapRecord = new ProducerRecord(producerRecord.topic(), producerRecord.key(), producerRecord.value()+ "---mashibing"); // wrapRecord.headers().add("user","mashibing".getBytes()); return wrapRecord; } @Override public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) { System.out.println("metadata:"+recordMetadata+",exception:"+e); } @Override public void close() { } @Override public void configure(Map<String, ?> map) { } } 生产者端 public class ProducerQuickStart { public static void main(String[] args) { //1、创建生产者 Properties prop = new Properties(); prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,UserDefineInterceptor.class.getName()); KafkaProducer<String, String> producer = new KafkaProducer<>(prop); //2、发送消息 for (int i = 0; i < 5; i++) { ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "key"+i, "value"+i); producer.send(record); } //关闭 producer.close(); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
第四章、Kafka 高级API
Offset自动控制
Kafka消费者默认对于未订阅的topic的offset的时候，也就是系统并没有存储该消费者的消费分区的记录信息，默认Kafka消费者的默认首次消费策略：latest
auto.offset.reset=latest
uearliest - 自动将偏移量重置为最早的偏移量
ulatest - 自动将偏移量重置为最新的偏移量
unone - 如果未找到消费者组的先前偏移量，则向消费者抛出异常
Kafka消费者在消费数据的时候默认会定期的提交消费的偏移量，这样就可以保证所有的消息至少可以被消费者消费1次,用户可以通过以下两个参数配置：
enable.auto.commit = true 默认
auto.commit.interval.ms = 5000 默认
如果用户需要自己管理offset的自动提交，可以关闭offset的自动提交，手动管理offset提交的偏移量，注意用户提交的offset偏移量永远都要比本次消费的偏移量+1，因为提交的offset是kafka消费者下一次抓取数据的位置。
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);//开启自动提交
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,10000);//十秒后自动提交
- 1
- 2
Offset手动管理
public class KafkaConsumerDemo_02 { public static void main(String[] args) { //1.创建Kafka链接参数 Properties props=new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092"); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName()); props.put(ConsumerConfig.GROUP_ID_CONFIG,"group01"); //关闭自动提交 props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false); //2.创建Topic消费者 KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props); //3.订阅topic开头的消息队列 consumer.subscribe(Pattern.compile("^topic.*$")); while (true){ ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, String> record = recordIterator.next(); String key = record.key(); String value = record.value(); long offset = record.offset(); int partition = record.partition(); Map<TopicPartition, OffsetAndMetadata> offsets=new HashMap<TopicPartition, OffsetAndMetadata>(); offsets.put(new TopicPartition(record.topic(),partition),new OffsetAndMetadata(offset)); consumer.commitAsync(offsets, new OffsetCommitCallback() { @Override public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) { System.out.println("完成："+offset+"提交！"); } }); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
Acks & Retries
Kafka生产者在发送完一个的消息之后，要求Broker在规定的额时间Ack应答答，如果没有在规定时间内应答，Kafka生产者会尝试n次重新发送消息。
acks=1 默认
acks=1 - Leader会将Record写到其本地日志中，但会在不等待所有Follower的完全确认的情况下做出响应。在这种情况下，如果Leader在确认记录后立即失败，但在Follower复制记录之前失败，则记录将丢失。
acks=0 - 生产者根本不会等待服务器的任何确认。该记录将立即添加到套接字缓冲区中并视为已发送。在这种情况下，不能保证服务器已收到记录。
acks=all - 这意味着Leader将等待全套同步副本确认记录。这保证了只要至少一个同步副本仍处于活动状态，记录就不会丢失。这是最有力的保证。这等效于acks = -1设置。
如果生产者在规定的时间内，并没有得到Kafka的Leader的Ack应答，Kafka可以开启reties重试机制。
request.timeout.ms = 30000 默认
retries = 2147483647 默认
//超时设置
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1);
//ack 设置
props.put(ProducerConfig.ACKS_CONFIG,"-1");
//重试次数
props.put(ProducerConfig.RETRIES_CONFIG,10);
- 1
- 2
- 3
- 4
- 5
- 6
幂等性
HTTP/1.1中对幂等性的定义是：一次和多次请求某一个资源对于资源本身应该具有同样的结果（网络超时等问题除外）。也就是说，其任意多次执行对资源本身所产生的影响均与一次执行的影响相同。
Methods can also have the property of “idempotence” in that (aside from error or expiration issues) the side-effects of N > 0 identical requests is the same as for a single request.
Kafka在0.11.0.0版本支持增加了对幂等的支持。幂等是针对生产者角度的特性。幂等可以保证上生产者发送的消息，不会丢失，而且不会重复。实现幂等的关键点就是服务端可以区分请求是否重复，过滤掉重复的请求。要区分请求是否重复的有两点：
唯一标识：要想区分请求是否重复，请求中就得有唯一标识。例如支付请求中，订单号就是唯一标识
记录下已处理过的请求标识：光有唯一标识还不够，还需要记录下那些请求是已经处理过的，这样当收到新的请求时，用新请求中的标识和处理记录进行比较，如果处理记录中有相同的标识，说明是重复记录，拒绝掉。
幂等性 需设置以下几个参数
//ack 设置 必须设置为all
prop.put(ProducerConfig.ACKS_CONFIG,"all");
//重试次数 要求大于0
prop.put(ProducerConfig.RETRIES_CONFIG,10);
//开启kafka的幂等性
prop.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
//设置为1 严格保证有序 要求小于5
prop.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,1);
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
事务控制
Kafka的幂等性，只能保证一条记录的在分区发送的原子性，但是如果要保证多条记录（多分区）之间的完整性，这个时候就需要开启kafk的事务操作。
在Kafka0.11.0.0除了引入的幂等性的概念，同时也引入了事务的概念。通常Kafka的事务分为 *生产者事务Only、**消费者&生产者事务。一般来说默认消费者消费的消息的级别是read_uncommited数据，这有可能读取到事务失败的数据，所有在开启生产者事务之后，需要用户设置消费者的事务隔离级别。
isolation.level = read_uncommitted 默认
该选项有两个值read_committed|read_uncommitted，如果开始事务控制，消费端必须将事务的隔离级别设置为read_committed
开启的生产者事务的时候，只需要指定transactional.id属性即可，一旦开启了事务，默认生产者就已经开启了幂等性。但是要求"transactional.id"的取值必须是唯一的，同一时刻只能有一个"transactional.id"存储在，其他的将会被关闭。
public class ProducerTransactionProducerOnly { public static void main(String[] args) { KafkaProducer<String, String> producer = buildProducer(); //1、初始化事务 producer.initTransactions(); try { //2.开始事务 producer.beginTransaction(); for (int i = 0; i < 10; i++) { if (i==8){ int j = 10/0; } ProducerRecord<String, String> record = new ProducerRecord<>("topic01", "transaction"+i, "value"+i); producer.send(record); producer.flush(); //3.提交事务 } producer.commitTransaction(); }catch (Exception e){ System.out.println("消息出错了" + e.getMessage()); //4.终止事务 producer.abortTransaction(); }finally { producer.close(); } } public static KafkaProducer<String,String> buildProducer(){ //1、创建生产者 Properties prop = new Properties(); prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //必须配置事务ID ，必须唯一 prop.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-id"+ UUID.randomUUID().toString()); //配置kafka批处理大小 prop.put(ProducerConfig.BATCH_SIZE_CONFIG,1024); //等待5ms 如果batch中的数据还不够1024大小 也会发送 prop.put(ProducerConfig.LINGER_MS_CONFIG,5); //配置kafka重试机制和幂等性 prop.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true); prop.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,20000); prop.put (ProducerConfig.ACKS_CONFIG,"all"); return new KafkaProducer<>(prop); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
读已提交
public class ConsumerTransactionReadCommited { public static void main(String[] args) { Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,"g1");//kafka中消费者通常以组的方式消费 //设置消费者的消费事务的隔离级别read commited prop.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed "); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop); consumer.subscribe(Arrays.asList("topic01")); while (true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(1));//每隔一秒取数据 //有数据 while (true){ ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, String> record = recordIterator.next(); String key = record.key(); String value = record.value(); long offset = record.offset(); int partition = record.partition(); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
读未提交 出异常了也能读到消息
public class ConsumerTransactionReadUncommited { public static void main(String[] args) { Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,"g2");//kafka中消费者通常以组的方式消费 prop.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_uncommitted"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop); consumer.subscribe(Arrays.asList("topic01")); while (true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(1));//每隔一秒取数据 //有数据 while (true){ ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1)); Iterator<ConsumerRecord<String, String>> recordIterator = consumerRecords.iterator(); while (recordIterator.hasNext()){ ConsumerRecord<String, String> record = recordIterator.next(); String key = record.key(); String value = record.value(); long offset = record.offset(); int partition = record.partition(); System.out.println("key:"+key+",value:"+value+",partition:"+partition+",offset:"+offset); } } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
既是生产者又是消费者 用于做数据的转换。
public class ProducerTransactionProducerAndConsumer { public static void main(String[] args) { KafkaProducer<String, String> producer = buildProducer(); KafkaConsumer<String, String> consumer = buildConsumer("g1"); //1、初始化事务 producer.initTransactions(); consumer.subscribe(Arrays.asList("topic01")); while (true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1)); if (records.isEmpty()){ Map<TopicPartition, OffsetAndMetadata> offsets= new HashMap<>(); Iterator<ConsumerRecord<String, String>> iterator = records.iterator(); producer.beginTransaction(); try { while (iterator.hasNext()){ ConsumerRecord<String, String> record = iterator.next(); offsets.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset()+1)); ProducerRecord<String,String> pRecord = new ProducerRecord<String, String>("topic02",record.key(),record.value()+"mashibing--------"); producer.send(pRecord); } }catch (Exception e){ producer.abortTransaction(); } } } } public static KafkaProducer<String,String> buildProducer(){ //1、创建生产者 Properties prop = new Properties(); prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //必须配置事务ID ，必须唯一 prop.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-id"+ UUID.randomUUID().toString()); //配置kafka批处理大小 prop.put(ProducerConfig.BATCH_SIZE_CONFIG,1024); //等待5ms 如果batch中的数据还不够1024大小 也会发送 prop.put(ProducerConfig.LINGER_MS_CONFIG,5); //配置kafka重试机制和幂等性 prop.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true); prop.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,20000); prop.put (ProducerConfig.ACKS_CONFIG,"all"); return new KafkaProducer<>(prop); } public static KafkaConsumer<String,String> buildConsumer(String group){ //1、创建生产者 Properties prop = new Properties(); prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092"); prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); prop.put(ConsumerConfig.GROUP_ID_CONFIG,group);//kafka中消费者通常以组的方式消费 prop.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed "); //必须关闭消费者的自动提交 prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false); return new KafkaConsumer<String, String>(prop); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
数据同步机制
Kafka的Topic被分为多个分区，分区是是按照Segments存储文件块。分区日志是存储在磁盘上的日志序列，Kafka可以保证分区里的事件是有序的。其中Leader负责对应分区的读写、Follower负责同步分区的数据，0.11 版本之前Kafka使用highwatermarker机制保证数据的同步，但是基于highwatermarker的同步数据可能会导致数据的不一致或者是乱序。在Kafka数据同步有以下概念。
LEO：log end offset 标识的是每个分区中最后一条消息的下一个位置，分区的每个副本都有自己的 LEO.
HW: high watermarker称为高水位线，所有HW之前的的数据都理解是已经备份的,当所有节点都备 份成功，Leader会更新水位线。
ISR:In-sync-replicas,kafka的leader会维护一份处于同步的副本集和，如果在replica.lag.time.max.ms时间内系统没有发送fetch请求，或者已然在发送请求，但是在该限定时间内没有赶上Leader的数据就被剔除ISR列表。在Kafka-0.9.0版本剔除replica.lag.max.messages消息个数限定，因为这个会导致其他的Broker节点频繁的加入和退出ISR。
①：High Watermark Truncation followed by Immediate Leader Election（数据丢失）
②: Replica Divergence on Restart after Multiple Hard Failures（数据不一致）
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4mD5LRsQ-1658630995483)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\1657205123712.png)]](https://img-blog.csdnimg.cn/516637e0fe1d42f48cee8db599253028.png)
0.11版本之前Kafka的副本备份机制的设计存在问题。依赖HW的概念实现数据同步，但是存在数据不一致问题和丢失数据问题，因此Kafka-0.11版本引入了
Leader Epoch解决这个问题，不在使用HW作为数据截断的依据。而是已引入了Leader
epoch的概念，任意一个Leader持有一个LeaderEpoch。该LeaderEpoch这是一个由Controller管理的32位数字，存储在Zookeeper的分区状态信息中，并作为LeaderAndIsrRequest的一部分传递给每个新的Leader。Leader接受Producer请求数据上使用LeaderEpoch标记每个Message。然后，该LeaderEpoch编号将通过复制协议传播，并用于替换HW标记，作为消息截断的参考点。
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gQvS6UHv-1658630995484)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\1657205147757.png)]](https://img-blog.csdnimg.cn/b7877fcce7f94a768b9d1a5aeb658e6b.png)
改进消息格式，以便每个消息集都带有一个4字节的Leader Epoch号。在每个日志目录中，会创建一个新的Leader Epoch Sequence文件，在其中存储Leader Epoch的序列和在该Epoch中生成的消息的Start Offset。它也缓存在每个副本中，也缓存在内存中。
follower变成Leader
当Follower为Leader时，它首先将新的Leader Epoch和副本的LEO添加到Leader Epoch Sequence序列文件的末尾并刷新数据。给Leader产生的每个新消息集都带有新的“Leader Epoch”标记。
Leader变成Follower
如果需要需要从本地的Leader Epoch Sequence加载数据，将数据存储在内存中，给相应的分区的Leader发送epoch 请求，该请求包含最新的EpochID,StartOffset信息.Leader接收到信息以后返回该EpochID所对应的LastOffset信息。该信息可能是最新EpochID的StartOffset或者是当前EpochID的Log End Offset信息.**
情形1：Fllower的Offset比Leader的小
情形2：用户的Leader
Epoch的信息startOffset信息比Leader返回的LastOffset要大，Follower回去重置自己的Leader
Epoch文件，将Offset修改为Leader的LastOffset信息，并且截断自己的日志信息。
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-XdtAFjXz-1658630995484)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\1657205223328.png)]](https://img-blog.csdnimg.cn/f869648734084b2480ee3fb1bc15c893.png)
Follower在提取过程中，如果关注者看到的LeaderEpoch消息集大于其最新的LeaderEpoch，则会在其LeaderEpochSequence中添加新的LeaderEpoch和起始偏移量，并将Epoch数据文件刷新到磁盘。同时将Fetch的日志信息刷新到本地日志文件。
High Watermark Truncation followed by Immediate Leader Election（数据丢失）
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-HlBvkdOA-1658630995484)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\1657205238652.png)]](https://img-blog.csdnimg.cn/81b73e30081241239c8e5b3259405d75.png)
②: Replica Divergence on Restart after Multiple Hard Failures（数据不一致）
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g2nbzYPq-1658630995484)(C:\Users\admin\AppData\Roaming\Typora\typora-user-images\1657205257870.png)]](https://img-blog.csdnimg.cn/65879b66d0d94ddc91da65d3faa7729f.png)
第五章、进阶
Kafka-Eagle安装
这是一个监视系统，监视您的kafka群集以及可视的使用者线程，偏移量，所有者等。当您安装Kafka
Eagle时，用户可以看到当前的使用者组，对于每个组，他们正在消耗的Topic以及该组在每个主题中的偏移量，滞后，日志大小和位置。这对于了解用户从消息队列消耗的速度以及消息队列增加的速度很有用。
[root@CentOSB ~]# tar -zxf kafka-eagle-web-1.4.0-bin.tar.gz -C /usr/ [root@CentOSB ~]# mv /usr/kafka-eagle-web-1.4.0 /usr/kafka-eagle [root@CentOSB ~]# vi .bashrc KE_HOME=/usr/kafka-eagle JAVA_HOME=/usr/java/latest PATH=$PATH:$JAVA_HOME/bin:$KE_HOME/bin CLASSPATH=. export JAVA_HOME export PATH export CLASSPATH export KE_HOME [root@CentOSB ~]# source .bashrc [root@CentOSB ~]# cd /usr/kafka-eagle/ [root@CentOSB kafka-eagle]# vi conf/system-config.properties kafka.eagle.zk.cluster.alias=cluster1 cluster1.zk.list=CentOSA:2181,CentOSB:2181,CentOSC:2181 cluster1.kafka.eagle.offset.storage=kafka kafka.eagle.metrics.charts=true kafka.eagle.driver=com.mysql.jdbc.Driver kafka.eagle.url=jdbc:mysql://192.168.52.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull kafka.eagle.username=root kafka.eagle.password=root [root@CentOSB kafka-eagle]# chmod u+x bin/ke.sh [root@CentOSB kafka-eagle]# ./bin/ke.sh start vi /usr/local/kafka_2.2.0/bin/kafka-server-start.sh 如果需要检测Kafka性能指标需要修改Kafka启动文件 vi kafka-server-start.sh ... 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 flume启动 命令： ./bin/flume-ng agent -c conf/ -n a1 -f conf/kafka.properties -Dflume.root.logger=INFO,console
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
