桥下の民 - Kafka
type: channel
link: https://www.wordplay.work/index.php/category/Kafka
items: Kafka 幂等和事务
items: Kafka 分区副本重分配、数据迁移、副本扩缩容
items: Kafka 数据迁移
items: Kafka 安全(SASL+ACL)
items: Kafka 命令整合
items: MM2 的跨 IDC kafka 热备多活
items: Kafka Tool
items: Kafka有序性
items: @KafkaListener 注解
items: Kafka 生产调优

Kafka 幂等和事务
type: item
date: 2024-08-16T01:26:00+00:00
description: 一、幂等性 1.1、Exactly-Once 保证在消息重发的时候，消费者不会重复处理。即使在消费者收到重复消息的时候，重复处理，也要保证最终结果的一致性。 所谓幂等性，数学概念就是：f(f(x)) = f(x) 。f 函数表示对消息的处理。 比如，银行转账，如果失败，需要重试。不管重试多少次，都要保证最终结果一定是一致的。 Exactly-Once：即精准一次，来保证幂等性。Exactly once 是 Kafka 从版本 0.11 之后提供的高级特性。 1.2、为什么需要幂等性 在使用 Kafka 时，需要确保 Exactly-Once 语义。分布式系统中，一些不可控因素有很多，比如网络、OOM、FullGC 等。在 Kafka Broker 确认 Ack 前，有可能出现网络异常、FullGC、OOM 等问题时导致 Ack 超时，Producer 会进行重复发送。注，在未达到最大重试次数前，会自动重试（非应用程序代码写的重试）。 Kafka 在引入幂等性之前，Producer 向 Broker 发送消息，然后 Broker 将消息追加到消息流中后给 Producer 返回 Ack 信号值。实现流程如下： 生产中，会出现各种不确定的因素，比如在 Producer 在发送给 Broker 的时候出现网络异常。比如以下这种异常情况的出现： 上图这种情况，当 Producer 第一次发送消息给 Broker 时，Broker 将消息 (x2,y2) 追加到了消息流中，但是在返回 Ack 信号给 Producer 时失败了（比如网络异常） 。此时，Producer 端触发重试机制，将消息 (x2,y2) 重新发送给 Broker，Broker 接收到消息后，再次将该消息追加到消息流中，然后成功返回 Ack 信号给 Producer。这样下来，消息流中就被重复追加了两条相同的 (x2,y2) 的消息。 因此需要保证幂等性保证即使多次发送也要让最终的结果一样。 1.3、如何实现幂等性 Kafka 在 0.11.0.0 之后加入的幂等性。他是通过添加唯一 ID，类似于数据库的主键，用于唯一标记一个消息。它在底层设计架构中引入了 ProducerID 和 SequenceNumber。 PID：ProducerID，每个生产者启动时，Kafka 都会给它分配一个 ID，ProducerID 是生产者的唯一标识，需要注意的是，Kafka 重启也会重新分配 PID 。 SequenceNumber：对于每个 ProducerID，Producer 发送数据的每个 Topic 和 Partition 都对应一个从 0 开始单调递增的 SequenceNumber 值。 同样，下图一种理想状态下的发送流程。实际情况下，会有很多不确定的因素，比如 Broker 在发送 Ack 信号给 Producer 时出现网络异常，导致发送失败。异常情况如下图所示： 当 Producer 发送消息 (x2,y2) 给 Broker 时，Broker 接收到消息并将其追加到消息流中。此时，Broker 返回 Ack 信号给 Producer 时，发生异常导致 Producer 接收 Ack 信号失败。对于 Producer 来说，会触发重试机制，将消息 (x2,y2) 再次发送，但是，由于引入了幂等性，在每条消息中附带了 PID（ProducerID）和 SequenceNumber。相同的 PID 和 SequenceNumber 发送给 Broker，而之前 Broker 缓存过之前发送的相同的消息，那么在消息流中的消息就只有一条 (x2,y2)，不会出现重复发送的情况。 1.4、使用幂等 通过如下配置使用幂等： enable.idempotence=true。enable.idempotence 配置项表示是否使用幂等性。当 enable.idempotence 配置为 true 时，acks 必须配置为 all。并且建议 max.in.flight.requests.per.connection 的值小于 5。 acks=all。 1.5、幂等性的限制条件 单独只使用 Producer 的幂等性是存在一些限制条件的： 只能保证 Producer 在单个会话内不丟不重 ，如果 Producer 出现意外挂掉再重启是无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不丢不重）; 幂等性不能跨多个 Topic-Partition，只能保证单个 partition 内的幂等性 ，当涉及多个 Topic-Partition 时，这中间的状态并没有同步。 如果需要跨会话、跨多个 topic-partition 的情况，需要使用 Kafka 的…

Kafka 分区副本重分配、数据迁移、副本扩缩容
type: item
date: 2024-06-12T02:30:00+00:00
description: 一、脚本的使用介绍 该脚本是 kafka 提供用来重新分配分区的脚本工具； 1.1、生成推荐配置脚本 关键参数 --generate。 在进行分区副本重分配之前，最好是用下面方式获取一个合理的分配文件； 编写 move-json-file.json 文件；这个文件就是告知想对哪些 Topic 进行重新分配的计算。 { "topics": [ {"topic": "test_create_topic1"} ], "version": 1 } 然后执行下面的脚本，--broker-list "0,1,2,3" 这个参数是你想要分配的 Brokers。 sh bin/kafka-reassign-partitions.sh --zookeeper xxx:2181 --topics-to-move-json-file config/move-json-file.json --broker-list "0,1,2,3" --generate 执行完毕之后会打印： Current partition replica assignment// 当前副本分配方式 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[3],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[2],"log_dirs":["any"]}]} Proposed partition reassignment configuration// 期望的重新分配方式 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]} 需求注意的是，此时分区移动尚未开始，它只是告诉你当前的分配和建议。保存当前分配，以防你想要回滚它。 1.2. 执行 Json 文件 关键参数 --execute。 将上面得到期望的重新分配方式文件保存在一个 json 文件里面 reassignment-json-file.json。 {"version":1,"partitions":[{"topic":"test_create_topic1","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":1,"replicas":[1],"log_dirs":["any"]},{"topic":"test_create_topic1","partition":0,"replicas":[0],"log_dirs":["any"]}]} 然后执行 sh bin/kafka-reassign-partitions.sh --zookeeper xxxxx:2181 --reassignment-json-file config/reassignment-json-file.json --execute 迁移过程注意流量陡增对集群的影响： Kafka 提供一个 broker 之间复制传输的流量限制，限制了副本从机器到另一台机器的带宽上限，当重新平衡集群，引导新 broker，添加或移除 broker 时候，这是很有用的。因为它限制了这些密集型的数据操作从而保障了对用户的影响、 例如上面的迁移操作加一个限流选项 --throttle 50000000。 在后面加上一个 --throttle 50000000 参数，那么执行移动分区的时候，会被限制流量在 50000000 B/s 加上参数后可以看到 The throttle limit was set to 50000000 B/s Successfully started reassignment of partitions. 需要注意的是，如果你迁移的时候包含副本跨路径迁移 (同一个 Broker 多个路径)…

Kafka 数据迁移
type: item
date: 2024-04-19T05:47:00+00:00
description: 两种迁移场景，分别是：同集群数据迁移、跨集群数据迁移。 一、同集群迁移 同集群之间数据迁移，比如在已有的集群中新增了一个 Broker 节点，此时需要将原来集群中已有的 Topic 的数据迁移部分到新的集群中，缓解集群压力。 将新的节点添加到 Kafka 集群很简单，只需为它们分配一个唯一的 Broker ID，并在新服务器上启动 Kafka。但是，这些新服务器节点不会自动分配任何数据分区，因此除非将分区移动到新增的节点，否则在创建新 Topic 之前新节点不会执行任何操作。因此，通常在将新服务器节点添加到 Kafka 集群时，需要将一些现有数据迁移到这些新的节点。 迁移数据的过程是手动启动的，执行过程是完全自动化的。在 Kafka 后台服务中，Kafka 将添加新服务器作为其正在迁移的分区的 Follower，并允许新增节点完全复制该分区中的现有数据。当新服务器节点完全复制此分区的内容并加入同步副本（ISR）时，其中一个现有副本将删除其分区的数据。 Kafka 系统提供了一个分区重新分配工具 kafka-reassign-partitions.sh，该工具可用于在 Broker 之间迁移分区。理想情况下，将确保所有 Broker 的数据和分区均匀分配。分区重新分配工具无法自动分析 Kafka 群集中的数据分布并迁移分区以实现均匀的负载均衡。因此，管理员在操作的时候，必须弄清楚应该迁移哪些 Topic 或分区。 分区重新分配工具可以在 3 种互斥模式下运行： --generate：在此模式下，给定 Topic 列表和 Broker 列表，该工具会生成候选重新​​分配，以将指定 Topic 的所有分区迁移到新 Broker 中。此选项仅提供了一种方便的方法，可在给定 Topic 和目标 Broker 列表的情况下生成分区重新分配计划。 --execute：在此模式下，该工具将根据用户提供的重新分配计划启动分区的重新分配。 （使用 --reassignment-json-file 选项）。由管理员手动制定自定义重新分配计划，也可以使用 --generate 选项提供。 --verify：在此模式下，该工具将验证最后一次 --execute 期间列出的所有分区的重新分配状态。状态可以有成功、失败或正在进行等状态。 1.1、迁移过程实现 分区重新分配工具可用于将一些 Topic 从当前的 Broker 节点中迁移到新添加的 Broker 中。这在扩展现有集群时通常很有用，因为将整个 Topic 移动到新的 Broker 变得更容易，而不是一次移动一个分区。当执行此操作时，用户需要提供已有的 Broker 节点的 Topic 列表，以及到新节点的 Broker 列表（源 Broker 到新 Broker 的映射关系）。然后，该工具在新的 Broker 中均匀分配给指定 Topic 列表的所有分区。在迁移过程中，Topic 的复制因子保持不变。 现有如下实例，将 Topic 为 ke01，ke02 的所有分区从 Broker1 中移动到新增的 Broker2 和 Broker3 中。由于该工具接受 Topic 的输入列表作为 JSON 文件，因此需要明确迁移的 Topic 并创建 json 文件，如下所示： > cat topic-to-move.json {"topics": [{"topic": "ke01"}, {"topic": "ke02"}], "version":1 } 准备好 JSON 文件，然后使用分区重新分配工具生成候选分配，命令如下： > bin/kafka-reassign-partitions.sh --zookeeper dn1:2181 --topics-to-move-json-file topics-to-move.json --broker-list "1,2" --generate 执行命名之前，Topic（ke01、ke02）的分区如下图所示： 执行完成命令之后，控制台出现如下信息： 该工具生成一个候选分配，将所有分区从 Topic ke01，ke02 移动到 Broker1 和 Broker2。需求注意的是，此时分区移动尚未开始，它只是告诉你当前的分配和建议。保存当前分配，以防你想要回滚它。新的赋值应保存在 JSON 文件（例如 expand-cluster-reassignment.json）中，以使用 --execute 选项执行。JSON 文件如下： {"version":1,"partitions":[{"topic":"ke02","partition":0,"replicas":[2]},{"topic":"ke02","partition":1,"replicas":[1]},{"t…

Kafka 安全(SASL+ACL)
type: item
date: 2024-03-07T08:23:00+00:00
description: 一、Security 在 Kafka 的官方文档中介绍了如下几种安全方式，以满足用户不同场景对安全的需求。 SASL/GSSAPI (Kerberos) - starting at version 0.9.0.0。主要是为 Kerberos 使用，如果当前已有 Kerberos 认证，只需要为集群中每个 Broker 和访问用户申请 Principle ，然后在 Kafka 配置文件中开启 Kerberos 的支持即可。官方文档详见 Authentication using SASL/Kerberos。 SASL/PLAIN - starting at version 0.10.0.0。一个简单的用户名和密码身份认证机制，通常与 TLS/SSL 一起用于加密，以实现身份验证。是一种比较容易使用的方式，但是也有一个很明显的缺点，这种方式会把用户账户文件配置到静态文件中，每次想要添加新的账户都需要重启 Kafka 去加载静态文件，才能生效，十分不方便。官方文档详见 Authentication using SASL/PLAIN。 SASL/SCRAM-SHA-256 和 SASL/SCRAM-SHA-512 - starting at version 0.10.2.0。通过将认证信息保存在 ZooKeeper 里面，从而动态的获取用户信息，相当于把 ZK 用作一个认证中心使用。这种认证可以在使用过程中，使用 Kafka 提供的命令动态地创建和删除用户，无需重启整个集群，十分方便。官方文档详见 Authentication using SASL/SCRAM 。 SASL/OAUTHBEARER - starting at version 2.0。Kafka 引入的新认证机制，主要是为了实现与 OAuth2 框架的集成，Kafka 不提倡单纯使用 OAUTHBEARER，因为它生成的不安全 Json Web Token，必须配以 SSL 加密才能在生产环境中使用。官方文档详见 Authentication using SASL/OAUTHBEARER 。 二、SASL+ACL 实现用户及权限认证 2.1、下载 wget https://repo.huaweicloud.com/apache/kafka/3.3.1/kafka_2.12-3.3.1.tgz -P /data cd /data tar -zxf kafka_2.12-3.3.1.tgz ln -s kafka_2.12-3.3.1 kafka cd /data/kafka 2.2、Kafka 服务配置 Kafka server 的 jaas 文件如下： /data/kafka/config/jaas/kafka_server_jaas.conf 这里创建五个用户，密码分别如下，后会会对其进行分别授权。 用户 密码 admin admin123 kwriter kwriter123 kreader kreader123 sharga sharga123 shargb shargb123 创建文件： mkdir /data/kafka/config/jaas/ vim /data/kafka/config/jaas/kafka_server_jaas.conf 配置如下： KafkaServer { org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin123" user_admin="admin123" user_kwriter="kwriter123" user_kreader="kreader123" user_sharga="sharga123" user_shargb="shargb123"; }; 2.3、修改 Kafka 服务启动脚本 1、复制一份作为带安全的启动脚本： cp /data/kafka/bin/kafka-server-start.sh/data/kafka/bin/sasl_kafka-server-start.sh 2、修改大概 44 行为如下，引入上面的 jaas.conf 文件，修改为如下 export LOG_DIR=/var/log/kafka-logs export KAFKA_OPTS="-Djava.security.auth.login.config=/data/kafka/config/jaas/kafka_server_jaas.conf" exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@" 2.4、配置 serv…

Kafka 命令整合
type: item
date: 2024-01-17T09:19:00+00:00
description: 一、Topic Command 1.0、Kafka 启停 启动 /opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties 停止 /opt/module/kafka/bin/kafka-server-stop.sh 1.1、Topic 创建 ​​bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test​​ 1.2、删除 Topic ​bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test​​ 1.3、Topic 分区扩容 单个 Topic 扩容 ​bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic test_create_topic1 --partitions 4​​ 批量扩容 (将所有正则表达式匹配到的 Topic 分区扩容到 4 个) ​​sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server localhost:9092 --alter --partitions 4​​ 1.4、查询 Topic 描述 1、查询单个 Topic bin/kafka-topics.sh --topic test --bootstrap-server localhost:9092 --describe --exclude-internal​​ 2、批量查询 Topic (正则表达式匹配，下面是查询所有 Topic) ​​sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server localhost:9092 --describe --exclude-internal​​ 支持正则表达式匹配 Topic, 只需要将 topic 用双引号包裹起来。 1.5、查询 Topic 列表 1、查询所有 Topic 列表 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list --exclude-internal​​ 2、查询匹配 Topic 列表 (正则表达式) 查询​​test_create_​​开头的所有 Topic 列表。 bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --list --exclude-internal --topic "user.*" 二、Config Command 2.1、查询配置 1、查询单个 Topic 配置 (只列举动态配置) ​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --topic test​​ 或者 ​​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --entity-type topics --entity-name test​​ 2、查询所有 Topic 配置 (包括内部 Topic)(只列举动态配置) ​​sh bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1.85:9092 --entity-type topics​​ 3、查询 Topic 的详细配置 (动态 + 静态) 只需要加上一个参数​​--all​​ 其他配置 /clients/users/brokers/broker-loggers 的查询 同理 ；只需要将​​--entity-type​​ 改成对应的类型就行了 (topics/clients/users/brokers/broker-loggers) 查询 kafka 版本信息 bin/kafka-configs.sh --describe --bootstrap-server 127.0.0.1:9092 --version​​ 2.2、增删改配置 ​​--alter​​ 删除配置: ​​--delete-config​​ k1=v1,k2=v2 添加 / 修改配置: ​​--add-config​​ k1,k2 选择类型: ​…

MM2 的跨 IDC kafka 热备多活
type: item
date: 2023-05-12T09:39:00+00:00
description: 一、MM2 简介 在介绍 MM2 之前先谈一下 MM1, 这个是 kafka 很早之前就有的组件，本质来说就是实现了 consumer + producer， 从集群 A 将数据同步到集群 B， 使用的是 kafka client 的 high level api。如果网络不稳定，会有频繁 rebalance 问题。一个 MM1 实例是解决部分 topic 从集群 A—>B 的同步问题。所以如果我们有 N 个 kafka 集群， 那么为了满足所有同步需求，理论上我们要维护 N (N-1) 个实例， 如果算上每个实例只处理部分 topic 同步的话，那么需要维护的实例会更多。 MM2 是 MM1 的升级替代品，用于 kafka 集群间的数据同步，解决了 MM1 的很多问题，同时部署维护成本大大降低，以 N 个 kafka 集群为例，只需要部署 N 个 MM2 集群即可满足需求。基于更高抽象的 Kafka Connect framework 解决容错和水平扩展问题。内部基于 low level (assign) 来实现 topic 订阅， 没有频繁 rebalance 的问题， 对同步有更精准的控制。 MM2 具有如下特性: 基于 Kafka Connect framework 和生态 自动探测新 topic, partition 自动同步 topic 配置，自动同步 topic acl 支持 active-active 集群对，以及任意数量的 active 集群 支持跨 IDC 同步， aggregation 和其他复杂拓扑 consumer offset 等 meta 信息的同步和翻译 no rebalance，减少同步波动 提供广泛的指标，例如跨多个数据中心 / 集群的端到端复制延迟 容错和水平可扩展 MM2 架构如下: MM2 基于 kafka connect framework 实现集群部署，通过 Connect Worker 来同步数据， 容错与水平扩展等由 Connect 框架保障。 MM2 主要有两种运行模式：Connector cluster 和 Driver, 我们采用的是 Driver 模式， 因为这种模式封装得更好，配置简洁，也是官方推荐的模式。 MM2 对于同步 topic 有前缀命名约定，默认情况下是 source clusterName, 当然也可以自定义，如下图所有，所有同步 topic 都包含有 source clusterName 前缀，这样做有两个好处：一是可以清楚知晓数据来源， 二是可以避免循环同步的问题。 下面介绍几个比较重要的内部 topic，可以说 MM2 的主要特性都依赖于这几个 topic: checkpoint topic: 同步集群间 consumer group state，其中记录了 consumer offset 的映射关系，在 2.7.0 + 版本后还支持自动提交到 target cluster 的__consumer_offsets， 也可通过 translateOffsets () 接口手动翻译后提交 offset_sync topic: 同步集群间 broker 端 offset 映射关系，分为 upstreamOffset 和 downstreamOffset, 其中 upstreamOffset 是上游 topic 的 offset (source 端)， downstreamOffset 是下游 topic 的 offset (target 端) heartbeat topic: 用于监控 replication flows, 可以动态发现 replication topology，便于 MM2 动态分配同步任务 关于 MM2 的部署我们采用的方案是：每个 kafka 集群 (作为 target cluster) 部署一个 MM2 集群实例，部署在 k8s 上方便运维管理， 通过 jmx_prometheus_javaagent 将 metrics 暴露成 exporter 供 prometheus server 抓取， 然后通过 grafana 展示。 二、跨 IDC kafka 热备多活 在没有跨 IDC kafka 热备方案之前， 我们只有冷备方案，就是在集群 A 出现问题的情况下，手动将业务迁移到集群 B 上， 这种操作的代价是巨大的， 而且短时间内无法恢复，离线业务可能还好，实时业务是完全无法接受的。 下面列举了在冷备的情况下业务迁移面临的问题: 业务上下游的依赖问题，因为涉及到生产和消费的先后问题，各业务必须协调好顺序 沟通上下游各个业务做迁移，沟通成本很高 数据安全得不到保障，在生产端没有及时切换的情况下可能会丢数据，下游 pipeline 是…

Kafka Tool
type: item
date: 2023-03-08T12:35:00+00:00
description: 一、下载安装 下载地址：Offset Explorer 根据不同的系统下载对应的版本。 [atguigu@hadoop102 kafka]$ find ./libs/ -name *kafka_* | head -1 | grep -o 'kafka[^ ]*' kafka_2.12-3.0.0.jar 二、简单使用 2.1、连接kafka 设置完了，点击Test测试是否能连接，连接通了，然后点击Add，添加完成设置。出现如下界面。 2.2、简单使用 配置以字符串的形式显示kafka消息体，Tools -> Setting -> Topics，把Key、Message的Byte Array改成String，点ok。 或者通过如下界面配置。 注释：更改完Content Types，要点击Update和Refresh按钮。 再次查看kafka的数据：

Kafka有序性
type: item
date: 2023-01-02T11:12:00+00:00
description: 如何保障kafka消费的顺序性，宗旨就是通过将消息绑定到定向的分区或者队列来保证顺序性，通过增加分区或者线程来提升消费能力。 一、生产者 要保证生产者发消息发的是顺序性的消息，发消息的时候指定一下key相同的key会发送到一个分区中，而分区时有序的在发消息的时候多个操作（下单，支付）保证顺序的话，保证这些操作在一个topic下的同一个分区，发送的时候指定key相同的key会去到同一个分区中，kafkaTemplate.send（topic,partition,key,data）有参数可以指定key。意思就是保证发消息的时候有序，发到同一个分区中，那么消费的时候也是有序的，因为分区可以理解为一个队列，消息先进先出。 当有消息被生产出来的时候，如果没有指定分区或者指定 key ，那么消费会按照【轮询】的方式均匀地分配到所有可用分区中，但不一定按照分区顺序来分配。 在 Kafka 中消费者可以订阅一个或多个主题，并被分配一个或多个分区。 如果一个消费者消费了多个分区，某些场景下消费者需要顺序地消费消息，但消息并不是按照顺序分配给分区的，所以就不一定能够保证消息消费的全局顺序性。 如何尽可能地保证消息消费的全局顺序性？（即这些消息具有因果关系）要想消费消息 B 必须先消费消息 A。 要注意的是，Kafka 的设计目标是提供高吞吐量和低延迟，而不是强制保证全局有序性。 所以该文探讨的是需要强调全局顺序性场景下的 Kafka 应用。 二、单分区 最简单粗暴的方法，虽然 Kafka 不能保证全局消费顺序性，但是能够保证分区内的消息顺序性。 可以只创建一个分区，并让消费者消费这个分区，这样就能够保证消费的消息是有序的。 但是这样做大大降低了吞吐量和处理效率，容易使得性能出现瓶颈。 三、基于key 在 Kafka 中，基于 key 的消息分配策略是通过消息中的键（key）来确定消息发送到哪个分区。 当生产者发送消息时，可以指定一个键（key），Kafka 使用这个键通过哈希算法来确定消息被发送到哪个分区。 四、代码 实现顺序性原理：消息发送的时候保证有序，设置相同的key会把消息投递到同一个分区的topic中，再由一个消费者来消费该分区topic。 topic: "topic_query_p3r1" 分配了三个partition分区。 4.1、producer 投递顺序消息 同一组行为设置相同的key，会把这组数据投递到同一分区topic中。 /** * 投递顺序性消息，根据用户id做取模推送到不同分区的topic中 * 相同的key推送到同一分区中 * 第一个参数：topic * 第二个参数：key * 第三个参数：发送的消息内容 * 三个参数全部是sring类型 */ @RequestMapping("/kafka2") public String testKafka2() { for (int userId = 0; userId < 3; userId++) { kafkaTemplate.send("topic_query_p3r1", userId + "", "insert" + userId); kafkaTemplate.send("topic_query_p3r1", userId + "", "update" + userId); kafkaTemplate.send("topic_query_p3r1", userId + "", "delete" + userId); } return null; 4.2、consumer 消费顺序消息 前提是生产者发消息的时候指定key了。 1、直接进行消费 因为投递的相同行为的消息是有序的，所以直接消费也不会有问题。 /** * 消费topic_query_p3r1主题，ConsumerGroupId1消费组 */ @KafkaListener(topics = "topic_query_p3r1", groupId = "ConsumerGroupId1") public void p3r2ConsumerGroupId0(ConsumerRecord<?, ?> consumer) throws InterruptedException { System.out.println("消费者A topic名称:" + consumer.topic() + ", key:" + consumer.key() + ", value:" + consumer.value() + ", 分区位置:" + consumer.partition() + ", 下标" + consumer.offset()+" "+Thread.currentThread().getId()); Thread.…

@KafkaListener 注解
type: item
date: 2022-09-08T13:55:00+00:00
description: 一、简介 Kafka 目前主要作为一个分布式的发布订阅式的消息系统使用，也是目前最流行的消息队列系统之一。本文集成的框架 spring-kafka。 源码解读： @Target ({ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE}) @Retention (RetentionPolicy.RUNTIME) @MessageMapping @Documented @Repeatable (KafkaListeners.class) public @interface KafkaListener { /** * 消费者的 id，当 GroupId 没有被配置的时候，默认 id 为 GroupId */ String id () default ""; /** * 监听容器工厂，当监听时需要区分单数据还是多数据消费需要配置 containerFactory 属性 */ String containerFactory () default ""; /** * 需要监听的 Topic，可监听多个，和 topicPattern 属性互斥 */ String [] topics () default {}; /** * 需要监听的 Topic 的正则表达。和 topics,topicPartitions 属性互斥 */ String topicPattern () default ""; /** * 可配置更加详细的监听信息，必须监听某个 Topic 中的指定分区，或者从 offset 为 200 的偏移量开始监听，可配置该参数，和 topicPattern 属性互斥 */ TopicPartition [] topicPartitions () default {}; /** * 侦听器容器组 */ String containerGroup () default ""; /** * 监听异常处理器，配置 BeanName */ String errorHandler () default ""; /** * 消费组 ID */ String groupId () default ""; /** * id 是否为 GroupId */ boolean idIsGroup () default true; /** * 消费者 Id 前缀 */ String clientIdPrefix () default ""; /** * 真实监听容器的 BeanName，需要在 BeanName 前加 "__" */ String beanRef () default "__listener"; } 二、案例 2.1、ConsumerRecord 类消费 使用 ConsumerRecord 类接收有一定的好处，ConsumerRecord 类里面包含分区信息、消息头、消息体等内容，如果业务需要获取这些参数时，使用 ConsumerRecord 会是个不错的选择。如果使用具体的类型接收消息体则更加方便，比如说用 String 类型去接收消息体。 这里编写一个 Listener 方法，监听 topic1，并把 ConsumerRecord 里面所包含的内容打印到控制台中： @Component public class Listener { private static final Logger log = LoggerFactory.getLogger (Listener.class); @KafkaListener (id = "consumer", topics = "topic1") public void consumerListener (ConsumerRecord record) { log.info ("topic.quick.consumer receive :" + record.toString ()); } } 2.2、批量消费 批量消费在现实业务场景中是很有实用性的。因为批量消费可以增大 kafka 消费吞吐量，提高性能。 批量消费实现步骤： 1、重新创建一份新的消费者配置，配置为一次拉取 10 条消息。 2、创建一个监听容器工厂，命名为：batchContainerFactory，设置其为批量消费并设置并发量为 5，这个并发量根据分区数决定，必须小于等于分区数，否则会有线程一直处于空闲状态。 3、创建一个分区数为 8 的 Topic。 4、创建监听方法，设置消费 id 为 batchConsumer，clientID 前缀为 batch，监听 batch，使用 batchContainerFactory 工厂创建该监听容器。 @Compone…

Kafka 生产调优
type: item
date: 2022-03-07T06:38:00+00:00
description: 一、Kafka 硬件配置选择 1.1、场景说明 100 万日活，每人每天 100 条日志，每天总共的日志条数是 100 万 * 100 条 = 1 亿条。 1 亿 / 24 小时 / 60 分 / 60 秒 = 1150 条 / 每秒钟。 每条日志大小：0.5k - 2k（取 1k）。 1150 条 / 每秒钟 * 1k ≈ 1m/s 。 高峰期每秒钟：1150 条 * 20 倍 = 23000 条。 每秒多少数据量：20MB/s。 1.2、服务器台数选择 服务器台数 = 2 * （生产者峰值生产速率 * 副本 / 100） + 1 = 2 * （20m/s * 2 / 100） + 1 = 3 台 建议 3 台服务器。 1.3、磁盘选择 kafka 底层主要是顺序写，固态硬盘和机械硬盘的顺序写速度差不多。 建议选择普通的机械硬盘。 每天总数据量：1 亿条 * 1k ≈ 100g 100g * 副本 2 * 保存时间 3 天 / 0.7 ≈ 1T 建议三台服务器硬盘总大小，大于等于 1T。 1.4、内存选择 Kafka 内存组成：堆内存 + 页缓存 1）Kafka 堆内存建议每个节点：10g ~ 15g 在 kafka-server-start.sh 中修改 if ["x$KAFKA_HEAP_OPTS" = "x"]; then export KAFKA_HEAP_OPTS="-Xmx10G -Xms10G" fi （1）查看 Kafka 进程号。 [atguigu@hadoop102 kafka]$ jps 2321 Kafka 5255 Jps 1931 QuorumPeerMain （2）根据 Kafka 进程号，查看 Kafka 的 GC 情况。 参数说明： S0C：第一个幸存区的大小； S1C：第二个幸存区的大小 S0U：第一个幸存区的使用大小； S1U：第二个幸存区的使用大小 EC：伊甸园区的大小； EU：伊甸园区的使用大小 OC：老年代大小； OU：老年代使用大小 MC：方法区大小； MU：方法区使用大小 CCSC: 压缩类空间大小； CCSU: 压缩类空间使用大小 YGC：年轻代垃圾回收次数； YGCT：年轻代垃圾回收消耗时间 FGC：老年代垃圾回收次数； FGCT：老年代垃圾回收消耗时间 GCT：垃圾回收消耗总时间； （3）根据 Kafka 进程号，查看 Kafka 的堆内存。 [atguigu@hadoop102 kafka]$ jmap -heap 2321 Attaching to process ID 2321, please wait... Debugger attached successfully. Server compiler detected. JVM version is 25.212-b10 using thread-local object allocation. Garbage-First (G1) GC with 8 thread (s) Heap Configuration: MinHeapFreeRatio = 40 MaxHeapFreeRatio = 70 MaxHeapSize = 2147483648 (2048.0MB) NewSize = 1363144 (1.2999954223632812MB) MaxNewSize = 1287651328 (1228.0MB) OldSize = 5452592 (5.1999969482421875MB) NewRatio = 2 SurvivorRatio = 8 MetaspaceSize = 21807104 (20.796875MB) CompressedClassSpaceSize = 1073741824 (1024.0MB) MaxMetaspaceSize = 17592186044415 MB G1HeapRegionSize = 1048576 (1.0MB) Heap Usage: G1 Heap: regions = 2048 capacity = 2147483648 (2048.0MB) used = 246367744 (234.95458984375MB) free = 1901115904 (1813.04541015625MB) 11.472392082214355% used G1 Young Generation: Eden Space: regions = 83 capacity = 105906176 (101.0MB) used = 87031808 (83.0MB) free = 18874368 (18.0MB) 82.17821782178218% used Surv…