需要基于 Apache Kafka 做实时事件流…

适用于用 Kafka 构建事件驱动/实时流处理系统，覆盖主题分区、生产消费、Streams、Schema Registry、Connect、监控、部署与迁移。

适用场景

解决思路

1. Kafka Streams 并行度：每个 stream partition 对应一个 Kafka topic partition，每条数据记录对应一条 Kafka message，key 决定路由到哪个分区；基于输入流分区创建固定数量 task，task 是固定并行单元，分区到 task 的分配不变，实例失败后任务自动在其他实例重启。
2. 为应用规划最大并行实例数：最大并行度受 stream task 数限制，而 task 数由输入主题最大分区数决定；例如输入主题 5 个分区最多跑 5 个实例，多出的实例空闲但可用于故障接管。
3. 避免数据和处理热点：让数据均匀分布到各分区，让处理工作量均匀分布到各分区，避免单个分区 2 百万消息而另一个为空。
4. 创建主题：bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my_topic_name --partitions 20 --replication-factor 3 --config x=y。
5. 增加分区：bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my_topic_name --partitions 40；注意不能减少分区，增加分区不会重分布已有数据，可能影响使用自定义分区的消费者。
6. 理解分区特性：分区内消息严格有序、不可变、仅追加、可按时间或大小保留、跨 broker 复制。
7. 用户主题管理：输入主题通过 Source processor 指定（如 StreamsBuilder#stream、StreamsBuilder#table、Topology#addSource），输出主题通过 sink processor 指定（如 KStream#to、KTable.to、Topology#addSink），必须提前手动创建和管理。
8. 避免自动创建主题：自动创建可能使用集群默认配置和默认主题设置（例如 replication factor），生产环境不鼓励。
9. 内部主题管理：状态存储的 changelog 等内部主题由应用创建，仅该流应用使用；启用安全时底层客户端需要 Kafka broker admin 权限；命名通常为 <application.id>-<operatorName>-<suffix>，但未来版本不保证。
10. 生产测试消息：bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic streams-plaintext-input，然后输入消息如 all streams lead to kafka。
11. 选择投递语义：至少一次用生产者重试直到收到 ack、消费者处理完再提交 offset，适合可接受重复或幂等处理；至多用消费者处理前提交 offset、生产者不重试，适合允许丢数据如 metrics；精确一次用事务写入和幂等生产者、消费者只读已提交消息，适合金融交易等关键处理。
12. Producer 负载均衡：producer 直接把数据发到 partition leader broker，不经过路由层；Kafka 节点向 producer 提供 metadata 以定位正确 partition leader；可用自定义分区逻辑或随机分布。
13. Producer API 操作：send(record) 发送记录并返回 Future；flush() 强制发送缓冲记录；close() 释放资源；metrics() 获取生产者指标。
14. 配置 producer 基础项：bootstrap.servers 用多个 broker 如 host1:9092,host2:9092,host3:9092；key.serializer/value.serializer 设置序列化类如 StringSerializer、ByteArraySerializer 或自定义复杂类型。
15. 配置 producer acks：0 不等待 ack，1 只等 leader，all 或 -1 等所有 ISR，生产高可靠用 all/-1。
16. 配置 producer retries：默认 2147483647（Integer.MAX_VALUE）；设 0 禁用重试。
17. 开启幂等 producer：enable.idempotence=true 防止重复。
18. Producer 性能调优：linger.ms 默认 0，设 5-100ms 增加批处理吞吐但增加延迟；batch.size 默认 16384，增大可提高吞吐；compression.type 可选 none、gzip、snappy、lz4、zstd 降低网络和存储但增加 CPU；buffer.memory 默认 33554432，缓冲未发送记录，满时阻塞。
19. 编写 Java producer：设置 Properties，包括 BOOTSTRAP_SERVERS_CONFIG、KEY_SERIALIZER_CLASS_CONFIG、VALUE_SERIALIZER_CLASS_CONFIG、ACKS_CONFIG=all、ENABLE_IDEMPOTENCE_CONFIG=true、LINGER_MS_CONFIG=10、BATCH_SIZE_CONFIG=32768。
20. 用 try-with-resources 创建 KafkaProducer；构造 ProducerRecord('my-topic', 'key1', 'Hello Kafka!')；用 producer.send(record, callback) 异步发送；callback 中检查 exception，成功时打印 metadata.partition() 和 metadata.offset()。
21. Producer 最佳实践：使用 batching；启用 idempotence；在 callback 处理错误；启用压缩；使用有意义的 partition key 保证有序；在 finally 中关闭 producer；监控 record-send-rate、compression-rate 等指标；尽量复用 producer 实例。
22. 理解消费者组：group ID 唯一标识；组内每个分区只由一个 consumer 消费；consumer 加入/离开自动 rebalancing；组按分区跟踪已提交 offset。
23. 消费者组监控：bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 列出组；--describe --group my-group --members --verbose 查看成员和分区分配。
24. 配置 consumer 基础项：bootstrap.servers、group.id、key.deserializer/value.deserializer、enable.auto.commit、auto.offset.reset。
25. 设置 auto.offset.reset：earliest 从最早开始，latest 从末尾开始，none 找不到 offset 时抛异常。
26. 了解 Kafka Streams 的 consumer 默认：max.poll.records=100；client.id 默认 <application.id>-<random-UUID>；enable.auto.commit 默认 true。
27. 编写 Java consumer：设置 BOOTSTRAP_SERVERS_CONFIG、GROUP_ID_CONFIG、KEY_DESERIALIZER_CLASS_CONFIG、VALUE_DESERIALIZER_CLASS_CONFIG、AUTO_OFFSET_RESET_CONFIG=earliest、ENABLE_AUTO_COMMIT_CONFIG=false。
28. Consumer 循环：subscribe(Collections.singletonList('my-topic'))；poll(Duration.ofMillis(100))；遍历 records 并 processRecord(record)；处理批次后 commitSync()；用 try-with-resources 关闭 KafkaConsumer。
29. Consumer offset 提交策略：auto-commit 简单但对至少一次有风险，可能处理完前提交；manual synchronous commit 阻塞直到成功，保证提交后再继续但吞吐低；manual asynchronous commit 非阻塞吞吐高，在 callback 处理失败；hybrid 处理中用 async，rebalance/关闭前用 sync。
30. 使用 commitAsync 回调：consumer.commitAsync((offsets, exception) -> 如果 exception != null 打印 Commit failed)；使用 commitSync 捕获 CommitFailedException。
31. Consumer 最佳实践：为至少一次语义禁用 auto-commit；实现 ConsumerRebalanceListener 处理 rebalance；缩短 poll() 调用处理时间；优雅关闭 consumer；监控 consumer lag；理解分区分配策略；Kafka consumer 非线程安全，每个线程一个 consumer；对瞬时失败做重试，永久失败送 DLQ。
32. Kafka Streams 应用分解为子拓扑，由 repartition 主题连接，每个子拓扑可独立扩展。
33. Kafka Streams 拓扑：Source processor 无上游，从 Kafka 主题消费输入流并转发下游；Sink processor 无下游，把接收记录发送到指定 Kafka 主题；普通 processor 处理当前记录时也可访问外部系统，结果可写回 Kafka 或外部系统。
34. 选择 KStream：用 StreamsBuilder builder = new StreamsBuilder(); KStream<String, Long> wordCounts = builder.stream('word-counts-input-topic', Consumed.with(Serdes.String(), Serdes.Long())); 表示不可变记录流。
35. 选择 KTable：用 builder.table('input-topic') 表示 changelog 流，每个 key 的最新值。
36. 选择 GlobalKTable：每个应用实例获取所有分区数据；KTable 每个实例只获取 1 个分区数据。
37. 写流到 Kafka：stream.to('my-stream-output-topic') 使用默认 key/value serdes；或 stream.to('my-stream-output-topic', Produced.with(Serdes.String(), Serdes.Long())) 显式指定 serdes；输出可能根据情况重新分区。
38. 手动重新分区：KStream<byte[], String> repartitionedStream = stream.repartition(Repartitioned.numberOfPartitions(10)); Kafka Streams 将生成主题作为内部主题管理，确保清理并允许下游子拓扑扩展；在 key 变更操作后且未触发自动重新分区时使用。
39. Join co-partitioning：等值 join 输入数据必须 co-partitioned；左右输入主题分区数相同；所有写入输入主题的应用使用相同分区策略，包括 producer partitioner.class 和 KStream#to 的 StreamPartitioner；默认分区设置一般满足。
40. Join 例外：KStream-GlobalKTable join 不需要 co-partitioning，因为每个实例有 GlobalKTable changelog 所有分区，可用 KeyValueMapper 做非 key join；KTable-KTable foreign-key join Kafka Streams 内部保证 co-partitioning。
41. Stateful 操作使用状态存储：Key-Value Stores 用于聚合和 join；Window Stores 用于时间操作；Session Stores 用于会话聚合。
42. 配置内部主题：message.timestamp.type 对内部主题用 CreateTime；内部 repartition 主题 compaction.policy=delete、retention.time=-1（无限）；Key-Value 存储的 changelog 主题 compaction.policy=compact；Windowed Key-Value 存储 changelog compaction.policy=delete,compact、retention.time 为 24 小时 + 窗口存储设置；Versioned State Stores cleanup.policy=compact、min.compaction.lag.ms 为 24 小时 + store 的 historyRetentionMs。
43. Kafka Streams 并行度：由输入主题分区数决定，读单个 10 分区主题最多跑 10 个实例；更多实例空闲；输入主题分区数是并行上限。
44. Kafka Streams 配置前缀：相同值给 consumer/producer/admin 用 PARAMETER_NAME；不同值用 consumer.PARAMETER_NAME、producer.PARAMETER_NAME、admin.PARAMETER_NAME；也可用 StreamsConfig.consumerPrefix('PARAMETER_NAME')、producerPrefix、adminClientPrefix。
45. 特定 consumer 类型配置：consumer.PARAMETER_NAME 给所有 consumer；restore.consumer.PARAMETER_NAME 给 restore consumer；也可用 StreamsConfig.restoreConsumerPrefix('PARAMETER_NAME')。
46. 主题配置：topic.PARAMETER_NAME 覆盖 changelog 和 repartition 主题默认值；也可用 StreamsConfig.topicPrefix('PARAMETER_NAME')。
47. Kafka Streams EOS 配置：consumer isolation.level=READ_COMMITTED；producer enable.idempotence=true；producer transaction.timeout.ms=10000；producer delivery.timeout.ms=Integer.MAX_VALUE。
48. EOS producer client ID 命名：at-least-once 默认 [client.Id]-StreamThread-[sequence-number]；exactly-once EOS v1 为 [client.Id]-StreamThread-[sequence-number]-[taskId]；EOS v2 beta 为 [client.Id]-StreamThread-[sequence-number]；client.id 可设或默认 [application.id]-[processId]（processId 随机 UUID）。
49. 拓扑命名稳定：聚合 repartition 主题用 Grouped；KStream-KTable join repartition 主题用 Joined；KStream-KStream join repartition 主题用 StreamJoined；KStream-KTable join 状态存储用 Joined；KStream-KStream join 状态存储用 StreamJoined；聚合和 KTable-KTable join 的状态存储用 Materialized；Stream/Table 非状态操作用 Named。
50. 强制显式命名：props.put(StreamsConfig.ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG, true) 防止使用自动生成名称启动，保证拓扑更新稳定。
51. 拓扑优化：设置 topology.optimization=all 或 none；优化可复用 source topics 作为 changelog topics，在从 KStreamBuilder 迁移到 StreamsBuilder 时重要。
52. 运行 WordCount 示例：mvn clean package；mvn exec:java -Dexec.mainClass=myapps.WordCount；观察子拓扑、source、processor、sink 和 repartition 主题 counts-store-repartition。
53. Schema Registry 部署：在 Kafka 集群旁部署 schema registry；它管理事件 schema 并映射到主题，指导 producer 使用正确格式；Kafka 不自带 schema registry，使用第三方实现。
54. Schema Registry 收益：管理 schema 演进；执行 backward/forward/full/none 兼容性检查；集中管理 schema；类型安全；自动生成文档；跟踪每个 subject 的 schema 版本。
55. Schema 格式：Avro 紧凑二进制且支持丰富 schema 演进；JSON Schema 人类可读且有校验；Protobuf 使用 Google Protocol Buffers。
56. 兼容性模式：BACKWARD 新 schema 可读旧数据；FORWARD 旧 schema 可读新数据；FULL 两者兼容；NONE 不检查。
57. Avro producer 示例：设置 VALUE_SERIALIZER_CLASS_CONFIG=KafkaAvroSerializer；设置 schema.registry.url=http://localhost:8081；定义 Avro schema；用 Schema.Parser 解析；创建 GenericRecord，put name 和 age；构造 ProducerRecord<String, GenericRecord>('users', 'user1', user)；producer.send(record)。
58. Kafka Connect 用途：Source Connector 把外部数据导入 Kafka（数据库 CDC、文件系统、消息队列、云服务、API/webhook）；Sink Connector 把 Kafka 数据导出到数据库、数据仓库、对象存储、搜索引擎、分析平台。
59. Sink connector 输入主题配置：用 topics 逗号分隔列表或 topics.regex 正则指定输入主题。
60. JDBC source connector 配置：设置 connector.class=io.confluent.connect.jdbc.JdbcSourceConnector、tasks.max=1、connection.url、connection.user、connection.password、table.whitelist=users,orders、mode=incrementing、incrementing.column.name=id、topic.prefix=postgres-。
61. Elasticsearch sink connector 配置：设置 connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector、tasks.max=1、topics=user-events,order-events、connection.url=http://localhost:9200、type.name=_doc、key.ignore=false。
62. CDC 做法：用 Debezium（MySQL、PostgreSQL、MongoDB、SQL Server）、Oracle GoldenGate、Maxwell's Daemon 等捕获数据库变更并实时流入 Kafka，用于实时同步、从现有数据库做事件溯源、微服务数据集成、零停机迁移。
63. Topic 操作权限：DESCRIBE_PRODUCERS 需要 Topic Read；DESCRIBE_TOPIC_PARTITIONS 需要 Topic Describe。
64. 监控 producer metrics：record-send-rate、record-error-rate、compression-rate-avg、buffer-available-bytes。
65. 监控 consumer metrics：records-consumed-rate、fetch-latency-avg、records-lag-max、commit-latency-avg。
66. 监控 broker metrics：UnderReplicatedPartitions、OfflinePartitionsCount、ActiveControllerCount、RequestHandlerAvgIdlePercent。
67. 监控 Streams metrics：process-rate、process-latency-avg、commit-rate、poll-rate。
68. 通用节点指标：outgoing-byte-rate、outgoing-byte-total、request-rate、request-total、request-size-avg、request-size-max、incoming-byte-rate；Mbean 模式 kafka.[producer|consumer|connect]:type=[consumer|producer|connect]-node-metrics,client-id=([-.w]+),node-id=([0-9]+)。
69. 生产集群架构：通常 3+ broker 保证容错；生产 replication factor 3；更多分区更多并行；ZooKeeper 或 KRaft 用 3 或 5 节点保证 quorum。
70. Broker HA 配置：broker.id=1；listeners=PLAINTEXT://broker1:9092,SSL://broker1:9093；log.dirs=/data/kafka-logs-1,/data/kafka-logs-2 用多磁盘；default.replication.factor=3；min.insync.replicas=2；unclean.leader.election.enable=false；auto.leader.rebalance.enable=true；log.retention.hours=168；log.segment.bytes=1073741824；log.retention.check.interval.ms=300000。
71. ELR：用 Admin Client 或 DescribeTopicPartitions API 获取 Eligible Leader Replicas；ELR 非空时选择未 fenced 的 replica；若最后已知 leader 未 fenced 则选择它，模拟 4.0 前所有 replica 离线行为；更新主题 min.insync.replicas 会清空该主题 ELR；更新集群默认 min.insync.replicas 会清空所有主题 ELR。
72. SSL/TLS 配置：listeners=SSL://broker:9093；security.inter.broker.protocol=SSL；ssl.keystore.location=/var/private/ssl/kafka.server.keystore.jks；ssl.keystore.password=password；ssl.key.password=password；ssl.truststore.location=/var/private/ssl/kafka.server.truststore.jks；ssl.truststore.password=password；ssl.client.auth=required。
73. SASL/PLAIN 配置：listeners=SASL_SSL://broker:9093；security.inter.broker.protocol=SASL_SSL；sasl.mechanism.inter.broker.protocol=PLAIN；sasl.enabled.mechanisms=PLAIN；配置 listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=admin password=admin-secret user_admin=admin-secret user_alice=alice-secret;。
74. Broker 性能调优：num.network.threads=8；num.io.threads=16；socket.send.buffer.bytes=1048576；socket.receive.buffer.bytes=1048576；socket.request.max.bytes=104857600；num.replica.fetchers=4；replica.fetch.max.bytes=1048576；log.flush.interval.messages=9223372036854775807；log.flush.interval.ms=null，依赖 OS page cache。
75. Producer 吞吐调优：acks=1；linger.ms=100；batch.size=65536；compression.type=lz4；buffer.memory=67108864；max.in.flight.requests.per.connection=5。
76. Consumer 调优：fetch.min.bytes=1；fetch.max.wait.ms=500；max.partition.fetch.bytes=1048576；max.poll.records=500；session.timeout.ms=30000；heartbeat.interval.ms=3000。
77. Producer 最佳实践补充：正确配置 acks 平衡持久性和吞吐；实现合理重试逻辑；监控发送速率和错误率；close(timeout) 优雅关闭。
78. Consumer 最佳实践补充：用 seek() 支持重放；对瞬时错误重试，永久错误 DLQ；监控高 lag 并告警；一消费者一线程。
79. Kafka Streams 最佳实践：用 Named、Grouped、Materialized 显式命名；正确配置 changelog 主题；实现 ProductionExceptionHandler；应用实例数与输入分区匹配；用 TopologyTestDriver 做单元测试；监控 lag、处理速率、错误率；关键应用启用 exactly-once；优雅处理关闭信号。
80. Topic 设计最佳实践：按吞吐需求定分区数；生产主题 replication factor=3；按用例设置时间或大小保留；changelog 和 lookup 主题用 compaction；统一命名规范；文档化主题用途和 schema；实现 ACL；监控分区指标。
81. 运维最佳实践：全面采集指标；对关键指标告警；监控磁盘、网络、CPU 做容量规划；实现灾难恢复备份策略；滚动升级并测试；启用加密和认证；维护 runbook；生产前做负载测试。
82. Event Sourcing 模式：把所有状态变化存为不可变事件，例如 OrderCreated -> OrderPaid -> OrderShipped -> OrderDelivered；事件存储用 Kafka 主题 order-events；不压缩，保留完整历史，retention 无限或很长。
83. CQRS 模式：写侧命令经 producer 产生事件到 events-topic；读侧消费者用 streams 构建物化视图 KTable。
84. Saga 模式：用 Kafka 协调分布式事务，例如 order-requested -> payment-requested -> payment-completed -> inventory-reserved -> order-confirmed；失败时发补偿事务如 payment-failed -> order-cancelled。
85. Outbox 模式：在数据库事务中同时写业务表和 outbox 表，BEGIN TRANSACTION; INSERT INTO orders VALUES (...); INSERT INTO outbox VALUES (event_data); COMMIT; 然后用 CDC connector 读 outbox 并发布到 Kafka。
86. Fan-out 模式：单主题多 consumer group 广播事件，例如 user-events topic 分别给 email-service（group: email）、analytics-service（group: analytics）、notification-service（group: notifications）。
87. DLQ 模式：processRecord(record) 抛 RetriableException 时重试；抛 NonRetriableException 时 sendToDLQ(record, e)。
88. 窗口聚合：KStream<String, PageView> views = ...；views.groupByKey().windowedBy(TimeWindows.of(Duration.ofMinutes(5))).count() 做翻滚窗口；加 .advanceBy(Duration.ofMinutes(1)) 做跳跃窗口；用 SessionWindows.with(Duration.ofMinutes(30)) 做会话窗口。
89. 排障 consumer lag 增加：检查消费者处理时间；扩容 consumer group 实例；优化处理逻辑；适当增加 max.poll.records。
90. 排障消息不到达：检查 producer send() 错误回调；验证主题存在且可访问；检查网络连通性；查看 broker 日志。
91. 排障重复消息：启用幂等 producer；实现幂等消费者处理；检查 offset 提交策略；验证 exactly-once 配置。
92. 排障 rebalance 太久：减少 max.poll.interval.ms；增加 session.timeout.ms；优化 poll() 处理时间；检查 consumer 健康。
93. 排障分区 leader 不可用：检查 broker 健康和日志；验证复制状态；检查 broker 间网络；检查 ISR。
94. 排障 OOM：减小 batch.size 和 buffer.memory；调 JVM heap；监控内存；检查处理逻辑内存泄漏。
95. Kafka Streams 迁移：KStreamBuilder 到 StreamsBuilder：用 kstream.repartition(...) 或用户管理主题 kstream.to('user-topic'); streamsBuilder.stream('user-topic'); 替代 KStream.through() 管理重新分区。
96. 主题前缀配置：Properties props = new Properties(); props.put(StreamsConfig.topicPrefix('my-prefix.') + 'replication.factor', '3'); KafkaStreams streams = new KafkaStreams(topology, props);。
97. 滚动升级：先在 staging 测试新版本；逐个升级 broker；每升级一个检查集群健康；升级 producer、consumer、streams 客户端；全程监控指标。
98. 资源参考：Apache Kafka Documentation、Confluent Platform、Kafka Streams Documentation、Schema Registry、Kafka Connect 文档。

适用边界 · 注意事项

• 不适合简单请求-响应、无需事件流或低吞吐 CRUD 场景；引入 Kafka 会增加运维复杂度。
• 没有 Kafka 集群、无法管理 broker/ZooKeeper/KRaft、Schema Registry、Connect 等基础设施时，不要直接套用生产部署与高级组件方案。
• 要求跨分区全局严格有序，或需要极低延迟且无法接受批处理/异步处理延迟时，需要重新设计分区键、并行度和语义。