实习项目用到了 Kafka，系统学习一下
参考链接
一 概述
common.yml放在节点中，系统 A、B、C 监听节点数据有无变更，如变更及时响应
| 组件 | 作用 |
|---|---|
| Producer | 消息生产者 |
| Consumer | 消息消费者 |
| Consumer Group | 消费者组 |
| Broker | Kafka 实例 |
| Topic | 消息主题（逻辑概念） |
| Partition | Topic 分区（物理概念），一个 Topic 可以包含多个分区，单分区内消息有序；每个分区对应一个 Leader 和多个 Follower，仅 Leader 与生产者、消费者交互；Partition 在物理上对应一个文件夹 |
| Segment | Partition 物理上被分成多个 Segment，每个 Segment 对应一个物理文件 |
| Zookeeper | 保存元信息，现已废除 |
同一 Broker 对同一个分区也只能存放一个副本，所以分区副本数不能超过 Broker 数
消费者组内的消费者，与分区的关系
分区数 >= 一组内的Consumer数，以实现系统的可伸缩性，否则有一些 Consumer 是无法消费的
| 属性值 | 值 | 含义 |
|---|---|---|
enable_auto_commit | false | 自动提交偏移量，当一个Group在一个Topic上提交偏移量时，下次再使用该Group读取该Topic的消息时，就会从偏移量的位置开始读取 |
session_timeout_ms | … | 检测Consumer发生崩溃所需的最长时间。超过该时间Consumer未汇报心跳，则认为Consumer失效，将其移出group |
auto_offset_reset | earliest | 决定当Group在某Topic上无偏移时，开始读取的位置。设置为earliest使得每次抽样都从Topic的开始位置进行抽样，如果设置为latest就只能抽样那些正在写入消息的Topic |
max_poll_records | … | 单次poll()的大消息数 |
group_id | … | Group名 |
max_poll_interval_ms | … | 两次poll()的大间隔时间，超过该时间则认为Consumer失效，将其移出Group |
heartbeat_interval_ms | … | Consumer向Cooperator汇报心跳的间隔时间 |
只有完成所有流程的消息才可以被消费
| acks | 行为 |
|---|---|
| 0 | 生产者发起消息写入请求后，不会等待任何来自 Broker 器的响应（最不安全） |
| 1 | 生产者发起消息写入请求后，分区的 Leader 成功落盘后，Broker 即向生产者返回成功响应 |
| -1 | 生产者发起消息写入请求后，ISR 集合中的所有副本都落盘，Broker 才向生产者返回成功响应（最安全） |
3 数据不重复：幂等性Kafka 副本备份策略——如何保证消息不丢失
AR（Assigned Repllicas）：一个分区的所有副本
ISR（In-Sync Replicas）：能够和 Leader 保持同步的 Follower + Leader本身 组成的集合
OSR（Out-Sync Relipcas）：不能和 Leader 保持同步的 Follower 集合
AR = ISR + OSR
- Kafka 只保证对 ISR 集合中的所有副本保证完全同步
- ISR 集合是动态调整的，如果一些副本**和 Leader 完全同步两次时间差超过阈值
replica.lag.time.max.ms**则被移出 ISR（因为生产者可以批量发送消息，所以不能指定未同步的消息条数作为检测标准）- 要使消息不丢失，需要满足
(acks = -1) && (replication.factor>=2) && (min.insync.replicas>=2)
(acks = -1) && (replication.factor>=2) && (min.insync.replicas>=2)(acks = 0)(幂等性) && (至少一次)enable.idempotence = trueOffset 存放于内置 Topic__consumer_offsets，由 Coordinator 管理
Consumer 的偏移量是按照 组 + Topic + 分区 进行维护的
偏移量相关概念
偏移量的提交方式
enable_auto_commit = true和auto_commit_interval_ms设置自动提交间隔consumer.commitSync()提交失败的时候一直尝试提交，直到遇到无法重试的情况下才会结束consumer.commitAsync()消费者线程不会阻塞，提交失败的时候也不会进行重试，可以配合回调函数记录错误信息KafkaConsumerconsumer = new KafkaConsumer(configs);
consumer.subscribe(Collections.singletonList("topic_0"));
try {while (true){ConsumerRecordsrecords = consumer.poll(3000);
for (ConsumerRecordrecord : records) {System.out.println(record.value());
}
consumer.commitAsync(); // 异步提交
}
} catch (Exception exception){// ...
} finally {consumer.commitSync(); // 消费者关闭前，或者异步提交发生异常时，使用同步阻塞式提交
consumer.close();
} partition_assignment_strategy_configRange：计算每个消费者要消费的分区数，多余的分区分配给前几个消费者（Topic 增加时容易造成消费不均衡）RoundRobin：轮询向消费者分配分区Sticy：尽量均匀地分配分区，根据上次的分配结果尽量减少变动页缓存
零拷贝
参考链接
作用是在数据报从网络设备到用户程序空间传递的过程中，减少数据拷贝次数，减少系统调用，实现 CPU 的零参与
网络数据持久化到磁盘 (Producer 到 Broker)
磁盘文件通过网络发送 (Broker 到 Consumer)
磁盘顺序写入
批量操作
分区并行处理
你是否还在寻找稳定的海外服务器提供商？创新互联www.cdcxhl.cn海外机房具备T级流量清洗系统配攻击溯源，准确流量调度确保服务器高可用性，企业级服务器适合批量采购，新人活动首月15元起，快前往官网查看详情吧
