From 74d69ea9189bc22de95f469bcc8bec3c5d989a04 Mon Sep 17 00:00:00 2001 From: qyx <565485304@qq.com> Date: Sat, 17 Sep 2022 13:57:06 +0800 Subject: [PATCH] =?UTF-8?q?[=E6=96=87=E6=A1=A3=E4=BF=AE=E6=94=B9](master):?= =?UTF-8?q?=20Kafka=20=E5=8F=82=E6=95=B0=E9=85=8D=E7=BD=AE=E4=B8=8B=20?= =?UTF-8?q?=E5=91=A8=E5=85=AD=E5=AD=A6=E4=B9=A0=202022=E5=B9=B409=E6=9C=88?= =?UTF-8?q?17=E6=97=A513:56:58?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bigdata/kafka/README.md | 56 ++++++++++++++++++++++++++++++++++++----- 1 file changed, 50 insertions(+), 6 deletions(-) diff --git a/bigdata/kafka/README.md b/bigdata/kafka/README.md index f716a11..d40693c 100644 --- a/bigdata/kafka/README.md +++ b/bigdata/kafka/README.md @@ -138,12 +138,56 @@ zookeeper.connect=[你的ip地址]:2181 - gg.handler.kafkahandler.Mode = tx gg.handler.kafkahandler.Mode = op 这两个的差别。遇到时 dml 数据会丢失的情况。用的是 op 。 - 当设置成op单个数据库表的变更(插入、更新、删除) 会被当成一条Kafka消息发送;当设置成tx时，数据库事务 所做的所有变更统一被封装进一条Kafka消息，并在事务提 交后被发送。后者有事务性的保障，至少有原子性方面的保证，不会丢失部分 CDC 数据。 - - - - - - +- Topic 级别参数 + - Topic 级别参数会**覆盖全局 Broker 参数的值**，而每个 Topic 都能设置自己的参数值， 这就是所谓的 Topic 级别参数。 + - **retention.ms**: 规定了该 Topic 消息被保存的时长。 默认是 7 天，即该 Topic 只保存最近 7 天的消息。一旦 设置了这个值，它会覆盖掉 Broker 端的全局参数值。 + - **retention.bytes**:规定了要为该 Topic 预留多大的磁 盘空间。和全局参数作用相似，这个值通常在多租户的 Kafka 集群中会有用武之地。当前默认值是 -1，表示可以 无限使用磁盘空间。 + - Topic 级别 参数的设置就是这种情况，我们有两种方式可以设置: + - 创建 Topic 时进行设置 + - 修改 Topic 时设置 + - 如何在创建 Topic 时设置这些参数 + - 你的部门需要将交易数据发送到 Kafka 进行处理，需要保 存最近半年的交易数据，同时这些数据很大，通常都有几 MB，但一般不会超过 5MB。现在让我们用以下命令来创建 Topic: + - bin/kafka-topics.sh--bootstrap-serverlocalhost:9092--create--topictransaction--partitions1--replication-factor1--configretention.ms=15552000000--configmax.message.bytes=5242880 + - 请注意结尾处的--config设置，我们就是在 config 后面指定了想要设置的 Topic 级别参数。 + - 使用另一个自带的命令kafka-configs来修改 Topic 级别参数。假设我们现在要发送最大值是 10MB 的消息，该如何修改呢？命令如下： + - bin/kafka-configs.sh--zookeeperlocalhost:2181--entity-typetopics--entity-nametransaction--alter--add-configmax.message.bytes=10485760 + - [你最好始终坚持使用第二种方式来设置，并且在未来，Kafka 社区很有可能统一使用kafka-configs脚本来调整 Topic 级别参数。] + +- JVM 参数 + - Kafka 服务器端代码是用 Scala 语言编写的，但终归还是编译成 Class 文件在 JVM 上运行，因此 JVM 参数设置对于 Kafka 集群的重要性不言而喻。 + - 有条件的话至少使用 Java 8 吧, [Kafka 2.0已经不支持Java 7了，2.1版本开始初步支持Java 11，但不建议生产环境用11，所以还是使用Java 8吧。] + - 堆大小这个参数至关重要。无脑给出一个通用的建议：将你的 JVM 堆大小设置成 **6GB** 吧，这是目前业界比较公认的一个合理值。我见过很多人就是使用默认的 Heap Size 来跑 Kafka，说实话默认的 1GB 有点小， + 毕竟 Kafka Broker 在与客户端进行交互时会在 JVM 堆上创建大量的 ByteBuffer 实例，Heap Size 不能太小。 + - [没有通用的标准，只有一个最佳实践值：6GB。最好还是监控一下实时的堆大小，特别是**GC之后的live data大小，通常将heapsize设置成其1.5~2倍就足以了**] + - JVM 端配置的另一个重要参数就是垃圾回收器的设置，也就是平时常说的 GC 设置。如果你依然在使用 Java 7，那么可以根据以下法则选择合适的垃圾回收器： + - 如果 Broker 所在机器的 CPU 资源非常充裕，建议使用 CMS 收集器。启用方法是指定-XX:+UseCurrentMarkSweepGC。 + - 否则，使用吞吐量收集器。开启方法是指定-XX:+UseParallelGC。 + - 当然了，如果你已经在使用 Java 8 了[G1是jdk9中默认的，jdk8还是需要显式指定的]。在没有任何调优的情况下，G1 表现得要比 CMS 出色，主要体现在更少的 Full GC，需要调整的参数更少等，所以使用 G1 就好了。 + - Java8默认的新生代垃圾回收器是：UseParallelGC，可以用-XX:+PrintCommandLineFlags -version查看，如果显示指定 -XX:+UseCurrentMarkSweepGC 的话，会默认开启 -XX:+UseParallelGC + - 确定好了要设置的 JVM 参数，我们该如何为 Kafka 进行设置呢？ + - 只需要设置下面这两个环境变量即可： + - KAFKA_HEAP_OPTS：指定堆大小。 + - KAFKA_JVM_PERFORMANCE_OPTS：指定 GC 参数。 + - 比如你可以这样启动 Kafka Broker，即在启动 Kafka Broker 之前，先设置上这两个环境变量： + - $> **export KAFKA_HEAP_OPTS=--Xms6g --Xmx6g** + - $> **export KAFKA_JVM_PERFORMANCE_OPTS= -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true** + - $> **bin/kafka-server-start.sh config/server.properties** + +- 操作系统参数 + - Kafka 并不需要设置太多的 OS 参数，但有些因素最好还是关注一下，比如下面这几个： + - 文件描述符限制 + - 文件系统类型 + - Swappiness + - 提交时间 + - 首先是 **ulimit -n** + - 觉得任何一个 Java 项目最好都调整下这个值。实际上，文件描述符系统资源并不像我们想象的那样昂贵，你不用太担心调大此值会有什么不利的影响。通常情况下将它设置成一个超大的值是合理的做法，比如 **ulimit -n 1000000**。 + - 但不设置的话后果很严重，比如你会经常看到“Too many open files”的错误。 + - 其次是文件系统类型的选择。 + - 这里所说的文件系统指的是如 ext3、ext4 或 XFS 这样的日志型文件系统。根据官网的测试报告，**XFS** 的性能要强于 ext4，所以生产环境最好还是使用 XFS。对了，最近有个 Kafka 使用 **ZFS** 的数据报告，貌似性能更加强劲，有条件的话不妨一试。 + - 第三是 swap 的调优 + - 网上很多文章都提到设置其为 0，将 swap 完全禁掉以防止 Kafka 进程使用 swap 空间。我个人反倒觉得还是不要设置成 0 比较好，我们可以设置成一个较小的值。为什么呢？因为一旦设置成 0，当物理内存耗尽时，操作系统会触发 OOM killer 这个组件， + 它会随机挑选一个进程然后 kill 掉，即根本不给用户任何的预警。但如果设置成一个比较小的值，当开始使用 swap 空间时，你至少能够观测到 Broker 性能开始出现急剧下降，从而给你进一步调优和诊断问题的时间。基于这个考虑， + 我**个人建议将 swappniess 配置成一个接近 0 但不为 0 的值，比如 1**。 ### 1.5 kafka的基本操作