推广一个网站周期,网站制作什么语言最好,wordpress游戏支付宝,水产网站模板使用集 群最大的好处是可以跨服务器进行负载均衡#xff0c;再则就是可以使用复制功能来避免因单点故 障造成的数据丢失。在维护 Kafka 或底层系统时#xff0c;使用集群可以确保为客户端提供高可用 性。 需要多少个broker 一个 Kafka 集群需要多少个 broker 取决于以下几个因… 使用集 群最大的好处是可以跨服务器进行负载均衡再则就是可以使用复制功能来避免因单点故 障造成的数据丢失。在维护 Kafka 或底层系统时使用集群可以确保为客户端提供高可用 性。 需要多少个broker 一个 Kafka 集群需要多少个 broker 取决于以下几个因素。首先需要多少磁盘空间来保 留数据以及单个 broker 有多少空间可用。如果整个集群需要保留 10TB 的数据每个 broker 可以存储 2TB那么至少需要 5 个 broker。如果启用了数据复制那么至少还需要 一倍的空间不过这要取决于配置的复制系数是多少 也就是说如 果启用了数据复制那么这个集群至少需要 10 个 broker。 第二个要考虑的因素是集群处理请求的能力。这通常与网络接口处理客户端流量的能力有 关特别是当有多个消费者存在或者在数据保留期间流量发生波动比如高峰时段的流量 爆发时。如果单个 broker 的网络接口在高峰时段可以达到 80% 的使用量并且有两个 消费者那么消费者就无法保持峰值除非有两个 broker。如果集群启用了复制功能则 要把这个额外的消费者考虑在内。因磁盘吞吐量低和系统内存不足造成的性能问题也可 以通过扩展多个 broker 来解决。 broker配置 要把一个 broker 加入到集群里只需要修改两个配置参数。首先所有 broker 都必须配 置相同的 zookeeper.connect该参数指定了用于保存元数据的 Zookeeper 群组和路径。其次每个 broker 都必须为 broker.id 参数设置唯一的值。如果两个 broker 使用相同的 broker.id那么第二个 broker 就无法启动。在运行集群时还可以配置其他一些参数特 别是那些用于控制数据复制的参数 操作系统调优 大部分 Linux 发行版默认的内核调优参数配置已经能够满足大多数应用程序的运行需求 不过还是可以通过调整一些参数来进一步提升 Kafka 的性能。这些参数主要与虚拟内存、 网络子系统和用来存储日志片段的磁盘挂载点有关。这些参数一般配置在 /etc/sysctl.conf 文件里 虚拟内存 一般来说 Linux 的虚拟内存会根据系统的工作负荷进行自动调整。我们可以对交换分区 的处理方式和内存脏页进行调整从而让 Kafka 更好地处理工作负载。 对于大多数依赖吞吐量的应用程序来说要尽量避免内存交换。内存页和磁盘之间的交换 对 Kafka 各方面的性能都有重大影响。Kafka 大量地使用系统页面缓存如果虚拟内存被 交换到磁盘说明已经没有多余内存可以分配给页面缓存了。 一种避免内存交换的方法是不设置任何交换分区。内存交换不是必需的不过它确实能够 在系统发生灾难性错误时提供一些帮助。进行内存交换可以防止操作系统由于内存不足而 突然终止进程。基于上述原因建议把 vm.swappiness 参数的值设置得小一点比如 1。该 参数指明了虚拟机的子系统将如何使用交换分区而不是只把内存页从页面缓存里移除。要优先考虑减小页面缓存而不是进行内存交换。 为什么不把 vm.swappiness 设为零 先前人们建议尽量把 vm.swapiness 设为 0它意味着“除非发生内存溢 出否则不要进行内存交换”。直到 Linux 内核 3.5-rc1 版本发布这个值的 意义才发生了变化。这个变化被移植到其他的发行版上包括 Red Hat 企业 版内核 2.6.32-303。在发生变化之后0 意味着“在任何情况下都不要发生交 换”。所以现在建议把这个值设为 1。 脏页会被冲刷到磁盘上调整内核对脏页的处理方式可以让我们从中获益。Kafka 依赖 I/O 性 能为生产者提供快速的响应。这就是为什么日志片段一般要保存在快速磁盘上不管是单个 快速磁盘如 SSD还是具有 NVRAM 缓存的磁盘子系统如 RAID。这样一来在后台刷 新进程将脏页写入磁盘之前可以减少脏页的数量这个可以通过将 vm.dirty_background_ ratio 设为小于 10 的值来实现。该值指的是系统内存的百分比大部分情况下设为 5 就可以 了。它不应该被设为 0因为那样会促使内核频繁地刷新页面从而降低内核为底层设备的 磁盘写入提供缓冲的能力。 通过设置 vm.dirty_ratio 参数可以增加被内核进程刷新到磁盘之前的脏页数量可以将它 设为大于 20 的值这也是系统内存的百分比。这个值可设置的范围很广60~80 是个比 较合理的区间。不过调整这个参数会带来一些风险包括未刷新磁盘操作的数量和同步刷 新引起的长时间 I/O 等待。如果该参数设置了较高的值建议启用 Kafka 的复制功能避 免因系统崩溃造成数据丢失。 为了给这些参数设置合适的值最好是在 Kafka 集群运行期间检查脏页的数量不管是在 生存环境还是模拟环境。可以在 /proc/vmstat 文件里查看当前脏页数量。 磁盘 除了选择合适的磁盘硬件设备和使用 RAID 外文件系统是影响性能的另一个重要因素。有很多种文件系统可供选择不过对于本地文件系统来说EXT4第四代可扩展文件系 统和 XFS 最为常见。 网络 默认情况下系统内核没有针对快速的大流量网络传输进行优化所以对于应用程序来 说一般需要对 Linux 系统的网络栈进行调优以实现对大流量的支持。实际上调整 Kafka 的网络配置与调整其他大部分 Web 服务器和网络应用程序的网络配置是一样的。首先可以对分配给 socket 读写缓冲区的内存大小作出调整这样可以显著提升网络的传 输性能。socket 读写缓冲区对应的参数分别是 net.core.wmem_default 和 net.core.rmem_ default合理的值是 131 072也就是 128KB。读写缓冲区最大值对应的参数分别是 net.core.wmem_max 和 net.core.rmem_max合理的值是 2 097 152也就是 2MB。要注 意最大值并不意味着每个 socket 一定要有这么大的缓冲空间只是说在必要的情况下 才会达到这个值。 Kafka生产者——向Kafka写入数据 我们从创建一个 ProducerRecord 对象开始ProducerRecord 对象需要包含目标主题和要发 送的内容。我们还可以指定键或分区。在发送 ProducerRecord 对象时生产者要先把键和 值对象序列化成字节数组这样它们才能够在网络上传输。 接下来数据被传给分区器。如果之前在 ProducerRecord 对象里指定了分区那么分区器 就不会再做任何事情直接把指定的分区返回。如果没有指定分区那么分区器会根据 ProducerRecord 对象的键来选择一个分区。选好分区以后生产者就知道该往哪个主题和 分区发送这条记录了。紧接着这条记录被添加到一个记录批次里这个批次里的所有消 息会被发送到相同的主题和分区上。有一个独立的线程负责把这些记录批次发送到相应的 broker 上。 服务器在收到这些消息时会返回一个响应。如果消息成功写入 Kafka就返回一个 RecordMetaData 对象它包含了主题和分区信息以及记录在分区里的偏移量。如果写入 失败则会返回一个错误。生产者在收到错误之后会尝试重新发送消息几次之后如果还 是失败就返回错误信息。 创建Kafka生产者 bootstrap.servers 该属性指定 broker 的地址清单地址的格式为 host:port。清单里不需要包含所有的 broker 地址生产者会从给定的 broker 里查找到其他 broker 的信息。不过建议至少要 提供两个 broker 的信息一旦其中一个宕机生产者仍然能够连接到集群上。 key.serializer broker 希望接收到的消息的键和值都是字节数组。生产者接口允许使用参数化类型因 此可以把 Java 对象作为键和值发送给 broker。这样的代码具有良好的可读性不过生 产者需要知道如何把这些 Java 对象转换成字节数组。key.serializer 必须被设置为一 个实现了 org.apache.kafka.common.serialization.Serializer 接口的类生产者会使 用这个类把键对象序列化成字节数组。Kafka 客户端默认提供了 ByteArraySerializer 这个只做很少的事情、StringSerializer 和 IntegerSerializer因此如果你只 使用常见的几种 Java 对象类型那么就没必要实现自己的序列化器。要注意key. serializer 是必须设置的就算你打算只发送值内容。 value.serializer 与 key.serializer 一样value.serializer 指定的类会将值序列化。如果键和值都是字 符串可以使用与 key.serializer 一样的序列化器。如果键是整数类型而值是字符串 那么需要使用不同的序列化器。 如何创建一个新的生产者 private Properties kafkaProps new Properties(); kafkaProps.put(bootstrap.servers, broker1:9092,broker2:9092); kafkaProps.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); kafkaProps.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); producer new KafkaProducer(kafkaProps); 我们把消息发送给服务器但并不关心它是否正常到达。大多数情况下消息会正常到 达因为 Kafka 是高可用的而且生产者会自动尝试重发。不过使用这种方式有时候 也会丢失一些消息。 我们使用 send() 方法发送消息它会返回一个 Future 对象调用 get() 方法进行等待 就可以知道消息是否发送成功。 我们调用 send() 方法并指定一个回调函数服务器在返回响应时调用该函数。 发送消息到Kafka 消息先是被放进缓冲区然后使用单独的线程发送到服务器端。send() 方法会返 回一个包含 RecordMetadata 的 Future 对象不过因为我们会忽略返回值所以无法知 道消息是否发送成功。如果不关心发送结果那么可以使用这种发送方式。比如记录 Twitter 消息日志或记录不太重要的应用程序日志。 我们可以忽略发送消息时可能发生的错误或在服务器端可能发生的错误但在发送消 息之前生产者还是有可能发生其他的异常。这些异常有可能是 SerializationException 说明序列化消息失败、BufferExhaustedException 或 TimeoutException说明缓冲区已 满又或者是 InterruptException说明发送线程被中断。 在这里producer.send() 方法先返回一个 Future 对象然后调用 Future 对象的 get() 方法等待 Kafka 响应。如果服务器返回错误get() 方法会抛出异常。如果没有发生错 误我们会得到一个 RecordMetadata 对象可以用它获取消息的偏移量。 如果在发送数据之前或者在发送过程中发生了任何错误比如 broker 返回了一个不允 许重发消息的异常或者已经超过了重发的次数那么就会抛出异常。我们只是简单地把 异常信息打印出来。 KafkaProducer 一般会发生两类错误。其中一类是可重试错误这类错误可以通过重发消息 来解决。比如对于连接错误可以通过再次建立连接来解决“无主no leader”错误则可 以通过重新为分区选举首领来解决。KafkaProducer 可以被配置成自动重试如果在多次重 试后仍无法解决问题应用程序会收到一个重试异常。另一类错误无法通过重试解决比如 “消息太大”异常。对于这类错误KafkaProducer 不会进行任何重试直接抛出异常。 异步发送消息 假设消息在应用程序和 Kafka 集群之间一个来回需要 10ms。如果在发送完每个消息后都 等待回应那么发送 100 个消息需要 1 秒。但如果只发送消息而不等待响应那么发送 100 个消息所需要的时间会少很多。大多数时候我们并不需要等待响应——尽管 Kafka 会把目标主题、分区信息和消息的偏移量发送回来但对于发送端的应用程序来说不是必 需的。不过在遇到消息发送失败时我们需要抛出异常、记录错误日志或者把消息写入 “错误消息”文件以便日后分析。 为了在异步发送消息的同时能够对异常情况进行处理生产者提供了回调支持。 生产者的配置 1. acks acks 参数指定了必须要有多少个分区副本收到消息生产者才会认为消息写入是成功的。这个参数对消息丢失的可能性有重要影响。该参数有如下选项。 • 如果 acks0生产者在成功写入消息之前不会等待任何来自服务器的响应。也就是说 如果当中出现了问题导致服务器没有收到消息那么生产者就无从得知消息也就丢 失了。不过因为生产者不需要等待服务器的响应所以它可以以网络能够支持的最大 速度发送消息从而达到很高的吞吐量。 • 如果 acks1只要集群的首领节点收到消息生产者就会收到一个来自服务器的成功 响应。如果消息无法到达首领节点比如首领节点崩溃新的首领还没有被选举出来 生产者会收到一个错误响应为了避免数据丢失生产者会重发消息。不过如果一个 没有收到消息的节点成为新首领消息还是会丢失。这个时候的吞吐量取决于使用的是 Kafka生产者——向Kafka写入数据 37 同步发送还是异步发送。如果让发送客户端等待服务器的响应通过调用 Future 对象 的 get() 方法显然会增加延迟在网络上传输一个来回的延迟。如果客户端使用回 调延迟问题就可以得到缓解不过吞吐量还是会受发送中消息数量的限制比如生 产者在收到服务器响应之前可以发送多少个消息。 • 如果 acksall只有当所有参与复制的节点全部收到消息时生产者才会收到一个来自 服务器的成功响应。这种模式是最安全的它可以保证不止一个服务器收到消息就算 有服务器发生崩溃整个集群仍然可以运行不过它的 延迟比 acks1 时更高因为我们要等待不只一个服务器节点接收消息。 2. buffer.memory 该参数用来设置生产者内存缓冲区的大小生产者用它缓冲要发送到服务器的消息。如果 应用程序发送消息的速度超过发送到服务器的速度会导致生产者空间不足。这个时候 send() 方法调用要么被阻塞要么抛出异常取决于如何设置 block.on.buffer.full 参数 3. compression.type 默认情况下消息发送时不会被压缩。该参数可以设置为 snappy、gzip 或 lz4它指定了 消息被发送给 broker 之前使用哪一种压缩算法进行压缩。snappy 压缩算法由 Google 发明 它占用较少的 CPU却能提供较好的性能和相当可观的压缩比如果比较关注性能和网 络带宽可以使用这种算法。gzip 压缩算法一般会占用较多的 CPU但会提供更高的压缩 比所以如果网络带宽比较有限可以使用这种算法。使用压缩可以降低网络传输开销和 存储开销而这往往是向 Kafka 发送消息的瓶颈所在。 4. retries 生产者从服务器收到的错误有可能是临时性的错误比如分区找不到首领。在这种情况 下retries 参数的值决定了生产者可以重发消息的次数如果达到这个次数生产者会 放弃重试并返回错误。默认情况下生产者会在每次重试之间等待 100ms不过可以通过 retry.backoff.ms 参数来改变这个时间间隔。建议在设置重试次数和重试时间间隔之前 先测试一下恢复一个崩溃节点需要多少时间比如所有分区选举出首领需要多长时间 让总的重试时间比 Kafka 集群从崩溃中恢复的时间长否则生产者会过早地放弃重试。不 过有些错误不是临时性错误没办法通过重试来解决比如“消息太大”错误。一般情 况下因为生产者会自动进行重试所以就没必要在代码逻辑里处理那些可重试的错误。你只需要处理那些不可重试的错误或重试次数超出上限的情况。 5. batch.size 当有多个消息需要被发送到同一个分区时生产者会把它们放在同一个批次里。该参数指 定了一个批次可以使用的内存大小按照字节数计算而不是消息个数。当批次被填满 批次里的所有消息会被发送出去。不过生产者并不一定都会等到批次被填满才发送半满 的批次甚至只包含一个消息的批次也有可能被发送。所以就算把批次大小设置得很大 也不会造成延迟只是会占用更多的内存而已。但如果设置得太小因为生产者需要更频 繁地发送消息会增加一些额外的开销。 6. linger.ms 该参数指定了生产者在发送批次之前等待更多消息加入批次的时间。KafkaProducer 会在 批次填满或 linger.ms 达到上限时把批次发送出去。默认情况下只要有可用的线程生 产者就会把消息发送出去就算批次里只有一个消息。把 linger.ms 设置成比 0 大的数 让生产者在发送批次之前等待一会儿使更多的消息加入到这个批次。虽然这样会增加延 迟但也会提升吞吐量因为一次性发送更多的消息每个消息的开销就变小了。 7. client.id 该参数可以是任意的字符串服务器会用它来识别消息的来源还可以用在日志和配额指 标里。 8. max.in.flight.requests.per.connection 该参数指定了生产者在收到服务器响应之前可以发送多少个消息。它的值越高就会占用 越多的内存不过也会提升吞吐量。把它设为 1 可以保证消息是按照发送的顺序写入服务 器的即使发生了重试。 9. timeout.ms、request.timeout.ms 和 metadata.fetch.timeout.ms request.timeout.ms 指定了生产者在发送数据时等待服务器返回响应的时间metadata. fetch.timeout.ms 指定了生产者在获取元数据比如目标分区的首领是谁时等待服务器 返回响应的时间。如果等待响应超时那么生产者要么重试发送数据要么返回一个错误 抛出异常或执行回调。timeout.ms 指定了 broker 等待同步副本返回消息确认的时间与 asks 的配置相匹配——如果在指定时间内没有收到同步副本的确认那么 broker 就会返回 一个错误。 10. max.block.ms 该参数指定了在调用 send() 方法或使用 partitionsFor() 方法获取元数据时生产者的阻塞 时间。当生产者的发送缓冲区已满或者没有可用的元数据时这些方法就会阻塞。在阻 塞时间达到 max.block.ms 时生产者会抛出超时异常。 11. max.request.size 该参数用于控制生产者发送的请求大小。它可以指能发送的单个消息的最大值也可以指 单个请求里所有消息总的大小。例如假设这个值为 1MB那么可以发送的单个最大消 息为 1MB或者生产者可以在单个请求里发送一个批次该批次包含了 1000 个消息每 个消息大小为 1KB。另外broker 对可接收的消息最大值也有自己的限制message.max. bytes所以两边的配置最好可以匹配避免生产者发送的消息被 broker 拒绝。 12. receive.buffer.bytes 和 send.buffer.bytes 这两个参数分别指定了 TCP socket 接收和发送数据包的缓冲区大小。如果它们被设为 -1 就使用操作系统的默认值。如果生产者或消费者与 broker 处于不同的数据中心那么可以 适当增大这些值因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。 顺序保证 Kafka 可以保证同一个分区里的消息是有序的。也就是说如果生产者按照 一定的顺序发送消息broker 就会按照这个顺序把它们写入分区消费者也 会按照同样的顺序读取它们。在某些情况下顺序是非常重要的。例如往 一个账户存入 100 元再取出来这个与先取钱再存钱是截然不同的 不过 有些场景对顺序不是很敏感。如果把 retries 设为非零整数同时把 max.in.flight.requests.per.connection 设为比 1 大的数那么如果第一个批次消息写入失败而第二个批次写入 成功broker 会重试写入第一个批次。 如果此时第一个批次也写入成功那 么两个批次的顺序就反过来了。一般来说如果某些场景要求消息是有序的那么消息是否写入成功也是 很关键的所以不建议把 retries 设为 0。可以把 max.in.flight.requests. per.connection 设为 1这样在生产者尝试发送第一批消息时就不会有其 他的消息发送给 broker。不过这样会严重影响生产者的吞吐量所以只有在 对消息的顺序有严格要求的情况下才能这么做。 序列化框架 Protobuf Protobuf Customer对象被序列化成表示customerID的4字节整数 表示customerName长度的4字节整数如果customerName为空则长度为0 表示customerName的N个字节 分区 ProducerRecord 对象包含了目标主题、键和值。Kafka 的消息是一个个 键值对ProducerRecord 对象可以只包含目标主题和值键可以设置为默认的 null不 过大多数应用程序会用到键。键有两个用途可以作为消息的附加信息也可以用来 决定消息该被写到主题的哪个分区。拥有相同键的消息将被写到同一个分区。也就是 说如果一个进程只从一个主题的分区读取数据 那么具有相 同键的所有记录都会被该进程读取。要创建一个包含键值的记录只需像下面这样创建 ProducerRecord 对象 如果键值为 null并且使用了默认的分区器那么记录将被随机地发送到主题内各个可用 的分区上。分区器使用轮询Round Robin算法将消息均衡地分布到各个分区上。 如果键值为 null并且使用了默认的分区器那么记录将被随机地发送到主题内各个可用 的分区上。分区器使用轮询Round Robin算法将消息均衡地分布到各个分区上。 如果键不为空并且使用了默认的分区器那么 Kafka 会对键进行散列使用 Kafka 自己 的散列算法即使升级 Java 版本散列值也不会发生变化然后根据散列值把消息映射 到特定的分区上。这里的关键之处在于同一个键总是被映射到同一个分区上所以在进 行映射时我们会使用主题所有的分区而不仅仅是可用的分区。这也意味着如果写入 数据的分区是不可用的那么就会发生错误。但这种情况很少发生。 只有在不改变主题分区数量的情况下键与分区之间的映射才能保持不变。举个例子在 分区数量保持不变的情况下可以保证用户 045189 的记录总是被写到分区 34。在从分 区读取数据时可以进行各种优化。不过一旦主题增加了新的分区这些就无法保证 了——旧数据仍然留在分区 34但新的记录可能被写到其他分区上。如果要使用键来映射 分区那么最好在创建主题的时候就把分区规划好 Kafka消费者——从Kafka读取数据 消费者和消费者群组 假设我们有一个应用程序需要从一个 Kafka 主题读取消息并验证这些消息然后再把它们 保存起来。应用程序需要创建一个消费者对象订阅主题并开始接收消息然后验证消息 并保存结果。过了一阵子生产者往主题写入消息的速度超过了应用程序验证数据的速 度这个时候该怎么办如果只使用单个消费者处理消息应用程序会远跟不上消息生成 的速度。显然此时很有必要对消费者进行横向伸缩。就像多个生产者可以向相同的主题 写入消息一样我们也可以使用多个消费者从同一个主题读取消息对消息进行分流。 Kafka 消费者从属于消费者群组。一个群组里的消费者订阅的是同一个主题每个消费者 接收主题一部分分区的消息。 如果我们往群组里添加更多的消费者超过主题的分区数量那么有一部分消费者就会被 闲置不会接收到任何消息 往群组里增加消费者是横向伸缩消费能力的主要方式。Kafka 消费者经常会做一些高延迟 的操作比如把数据写到数据库或 HDFS或者使用数据进行比较耗时的计算。在这些情 况下单个消费者无法跟上数据生成的速度所以可以增加更多的消费者让它们分担负 载每个消费者只处理部分分区的消息这就是横向伸缩的主要手段。我们有必要为主题 创建大量的分区在负载增长时可以加入更多的消费者。不过要注意不要让消费者的数 量超过主题分区的数量多余的消费者只会被闲置。 简而言之为每一个需要获取一个或多个主题全部消息的应用程序创建一个消费者群组 然后往群组里添加消费者来伸缩读取能力和处理能力群组里的每个消费者只处理一部分 消息。 消费者群组和分区再均衡 管理员添加了新的分区会发生分区重分配 分区的所有权从一个消费者转移到另一个消费者这样的行为被称为再均衡。再均衡非常 重要它为消费者群组带来了高可用性和伸缩性 在再均衡期间消费者无法读取消 息造成整个群组一小段时间的不可用。另外当分区被重新分配给另一个消费者时消 费者当前的读取状态会丢失它有可能还需要去刷新缓存在它重新恢复状态之前会拖慢 应用程序。 消费者通过向被指派为群组协调器的 broker不同的群组可以有不同的协调器发送心跳 来维持它们和群组的从属关系以及它们对分区的所有权关系。只要消费者以正常的时间 间隔发送心跳就被认为是活跃的说明它还在读取分区里的消息。消费者会在轮询消息 为了获取消息或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长会 话就会过期群组协调器认为它已经死亡就会触发一次再均衡。 如果一个消费者发生崩溃并停止读取消息群组协调器会等待几秒钟确认它死亡了才 会触发再均衡。在这几秒钟时间里死掉的消费者不会读取分区里的消息。在清理消费者 时消费者会通知协调器它将要离开群组协调器会立即触发一次再均衡尽量降低处理 停顿。 分配分区是怎样的一个过程 当消费者要加入群组时它会向群组协调器发送一个 JoinGroup 请求。第一 个加入群组的消费者将成为“群主”。群主从协调器那里获得群组的成员列 表列表中包含了所有最近发送过心跳的消费者它们被认为是活跃的 并负责给每一个消费者分配分区。它使用一个实现了 PartitionAssignor 接 口的类来决定哪些分区应该被分配给哪个消费者。 分配 完毕之后群主把分配情况列表发送给群组协调器协调器再把这些信息发 送给所有消费者。每个消费者只能看到自己的分配信息只有群主知道群组 里所有消费者的分配信息。这个过程会在每次再均衡时重复发生。 订阅主题 轮询就会处理所有的细节包括群组协调、分区再均衡、发送心跳和获取数据 消费者实际上是一个长期运行的应用程序它通过持续轮询向 Kafka 请求数据。 消费者的配置 1. fetch.min.bytes 该属性指定了消费者从服务器获取记录的最小字节数。broker 在收到消费者的数据请求时 如果可用的数据量小于 fetch.min.bytes 指定的大小那么它会等到有足够的可用数据时 才把它返回给消费者。这样可以降低消费者和 broker 的工作负载因为它们在主题不是很 活跃的时候或者一天里的低谷时段就不需要来来回回地处理消息。如果没有很多可用 数据但消费者的 CPU 使用率却很高那么就需要把该属性的值设得比默认值大。如果 消费者的数量比较多把该属性的值设置得大一点可以降低 broker 的工作负载。 2. fetch.max.wait.ms 我们通过 fetch.min.bytes 告诉 Kafka等到有足够的数据时才把它返回给消费者。而 feth. max.wait.ms 则用于指定 broker 的等待时间默认是 500ms。如果没有足够的数据流入 Kafka消费者获取最小数据量的要求就得不到满足最终导致 500ms 的延迟。如果要降低 潜在的延迟为了满足 SLA可以把该参数值设置得小一些。如果 fetch.max.wait.ms 被设 为 100ms并且 fetch.min.bytes 被设为 1MB那么 Kafka 在收到消费者的请求后要么返 回 1MB 数据要么在 100ms 后返回所有可用的数据就看哪个条件先得到满足。 3. max.partition.fetch.bytes 该属性指定了服务器从每个分区里返回给消费者的最大字节数。它的默认值是 1MB 它的默认值是 1MB也 就是说KafkaConsumer.poll() 方法从每个分区里返回的记录最多不超过 max.partition. fetch.bytes 指定的字节。如果一个主题有 20 个分区和 5 个消费者那么每个消费者需要 至少 4MB 的可用内存来接收记录。在为消费者分配内存时可以给它们多分配一些因 为如果群组里有消费者发生崩溃剩下的消费者需要处理更多的分区。max.partition. fetch.bytes 的值必须比 broker 能够接收的最大消息的字节数 大否则消费者可能无法读取这些消息导致消费者一直挂起重试。 4. session.timeout.ms 该属性指定了消费者在被认为死亡之前可以与服务器断开连接的时间默认是 3s。如 果消费者没有在 session.timeout.ms 指定的时间内发送心跳给群组协调器就被认为 已经死亡协调器就会触发再均衡把它的分区分配给群组里的其他消费者。该属性与 heartbeat.interval.ms 紧密相关。heartbeat.interval.ms 指定了 poll() 方法向协调器 发送心跳的频率session.timeout.ms 则指定了消费者可以多久不发送心跳。所以一 般需要同时修改这两个属性heartbeat.interval.ms 必须比 session.timeout.ms 小一 般是 session.timeout.ms 的三分之一。如果 session.timeout.ms 是 3s那么 heartbeat. interval.ms 应该是 1s。把 session.timeout.ms 值设得比默认值小可以更快地检测和恢 复崩溃的节点不过长时间的轮询或垃圾收集可能导致非预期的再均衡。把该属性的值设 置得大一些可以减少意外的再均衡不过检测节点崩溃需要更长的时间。 5. auto.offset.reset 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下因消费者长 时间失效包含偏移量的记录已经过时并被删除该作何处理。它的默认值是 latest意 思是说在偏移量无效的情况下消费者将从最新的记录开始读取数据在消费者启动之 后生成的记录。另一个值是 earliest意思是说在偏移量无效的情况下消费者将从 起始位置读取分区的记录。 6. enable.auto.commit 该属性指定了消费者是否自动提交偏移 量默认值是 true。为了尽量避免出现重复数据和数据丢失可以把它设为 false由自 己控制何时提交偏移量。如果把它设为 true还可以通过配置 auto.commit.interval.ms 属性来控制提交的频率。 7. partition.assignment.strategy 分区会被分配给群组里的消费者。PartitionAssignor 根据给定的消费者和主 题决定哪些分区应该被分配给哪个消费者。Kafka 有两个默认的分配策略。 Range 该策略会把主题的若干个连续的分区分配给消费者。假设消费者 C1 和消费者 C2 同时 订阅了主题 T1 和主题 T2并且每个主题有 3 个分区。那么消费者 C1 有可能分配到这 两个主题的分区 0 和分区 1而消费者 C2 分配到这两个主题的分区 2。因为每个主题 拥有奇数个分区而分配是在主题内独立完成的第一个消费者最后分配到比第二个消 费者更多的分区。只要使用了 Range 策略而且分区数量无法被消费者数量整除就会 出现这种情况。 8. client.id 该属性可以是任意字符串broker 用它来标识从客户端发送过来的消息通常被用在日志、 度量指标和配额里。 9. max.poll.records 该属性用于控制单次调用 call() 方法能够返回的记录数量可以帮你控制在轮询里需要处 理的数据量。 10. receive.buffer.bytes 和 send.buffer.bytes socket 在读写数据时用到的 TCP 缓冲区也可以设置大小。如果它们被设为 -1就使用操 作系统的默认值。如果生产者或消费者与 broker 处于不同的数据中心内可以适当增大这 些值因为跨数据中心的网络一般都有比较高的延迟和比较低的带宽。 加群联系作者vxxiaoda0423 仓库地址https://webvueblog.github.io/JavaPlusDoc/
