小菜鸟

java菜鸟号正在起航

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 5.1k 阅读时长 ≈ 5 分钟

Kafka 生产者深度解析：从命令行到源码实现

Kafka 生产者是消息流入 Kafka 集群的入口，负责将业务数据可靠、高效地发送到指定主题。无论是通过命令行工具快速发送消息，还是通过 KafkaProducer 客户端进行编程式集成，理解其工作机制、配置参数及底层流程都是优化性能和确保可靠性的关键。本文将从命令行工具、核心配置、性能测试到源码实现，全面解析 Kafka 生产者。

命令行生产者工具：kafka-console-producer

Kafka 提供 kafka-console-producer.sh（Linux/Mac）或 kafka-console-producer.bat（Windows）工具，用于快速通过命令行发送消息，适合测试和调试场景。

基本使用

（1）发送无 Key 消息
# Linux/Mac
./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

# Windows
kafka-console-producer.bat --bootstrap-server localhost:9092 --topic test-topic

输入消息并回车即可发送，每条输入会作为一条消息发送到 test-topic。

（2）发送带 Key 消息

通过 --property parse.key=true 启用 Key 解析，Key 与 Value 用Tab 键分隔：

# 发送带 Key 的消息（Key 为 "user1"，Value 为 "hello"）
./kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic test-topic \
  --property parse.key=true \
  --property key.separator=$'\t'  # 指定分隔符为 Tab（默认）

输入 user1<Tab>hello 并回车，消息 Key 为 user1，Value 为 hello。

核心参数说明

参数 作用 默认值
--bootstrap-server Kafka 集群地址（替代旧版 --broker-list） 无（必传）
--topic 目标主题名称 无（必传）
--property 自定义属性（如 parse.key=true 启用 Key 解析） 无
--producer-property 生产者配置（如 acks=all） 无
--sync 同步发送消息（默认异步） false
--compression-codec 消息压缩算法（none/gzip/snappy 等） none

生产者核心配置参数

Kafka 生产者的行为由一系列配置参数控制，可通过配置文件或代码指定。核心参数如下：

阅读全文 »

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 3.4k 阅读时长 ≈ 3 分钟

Kafka 主题（Topic）操作全解析：命令、源码与实践

Kafka 主题（Topic）是消息的逻辑容器，所有消息都需发送到指定主题。Kafka 提供了 kafka-topics.sh（或 .bat）脚本用于主题的创建、删除、查询和修改等操作。本文将详细介绍主题的核心操作命令、底层实现原理及最佳实践，帮助深入理解主题管理机制。

主题操作脚本与核心原理

脚本入口：kafka-topics

Kafka 主题操作的脚本是 kafka-topics.sh（Linux/Mac）或 kafka-topics.bat（Windows），其核心是调用 kafka.admin.TopicCommand 类执行操作：

# 脚本核心逻辑（kafka-topics.sh）
exec $(dirname $0)/kafka-run-class.sh kafka.admin.TopicCommand "$@"

TopicCommand 类的 main 方法解析命令参数，根据操作类型（创建、删除等）调用对应逻辑，底层通过 ZooKeeper 或 Kafka Admin API 交互。

主题创建的底层流程

主题创建分为客户端发起和服务端处理两个阶段：

  1. 客户端阶段：
    • 通过 kafka-topics 脚本发送创建请求，包含主题名、分区数、副本数等参数。
    • 客户端将主题元数据（如分区 - 副本分配方案）写入 ZooKeeper 的 /brokers/topics/[topic] 节点。
  2. 服务端阶段：
    • Kafka 控制器（Controller）监听 ZooKeeper 节点变化，发现新主题后触发创建流程。
    • 控制器在指定 Broker 上创建分区目录（如 topic-0、topic-1），并初始化 Leader 和 Follower 副本。

核心主题操作命令

创建主题（—create）

创建主题需指定分区数、副本数和 Kafka 集群地址，核心参数：

阅读全文 »

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 4k 阅读时长 ≈ 4 分钟

Kafka 启动与关闭全流程详解

Kafka 是一个依赖 ZooKeeper 的分布式消息系统，其启动过程涉及 ZooKeeper 初始化、Kafka 核心组件加载等步骤。本文将详细介绍 Kafka 及依赖的 ZooKeeper 的启动配置、操作步骤、源码级启动流程，以及安全关闭的方法，帮助理解其运行机制。

前置依赖：ZooKeeper 启动

Kafka 依赖 ZooKeeper 存储集群元数据（如 Broker 信息、主题配置、消费者偏移量等），需先启动 ZooKeeper。Kafka 自带 ZooKeeper 组件，推荐使用自带版本以避免版本兼容问题。

ZooKeeper 配置文件（zookeeper.properties）

Kafka 的 ZooKeeper 配置文件位于 config/zookeeper.properties，核心配置如下：

# 快照文件存储目录（默认/tmp/zookeeper，建议修改为持久化路径）
dataDir=E:\\zookeeper\\data  
# 事务日志存储目录（独立设置可提升性能）
dataLogDir=E:\\zookeeper\\logs  
# 客户端连接端口
clientPort=2181  
# 允许客户端最大连接数（0表示无限制）
maxClientCnxns=0
  • 注意：路径需避免包含空格，否则可能导致启动失败（报错 “系统找不到指定的路径”）。

启动 ZooKeeper

（1）Windows 环境

进入 Kafka 安装目录的 bin/windows 文件夹，执行启动命令：

# 启动 ZooKeeper（指定配置文件）
zookeeper-server-start ../../config/zookeeper.properties
（2）Linux/Mac 环境

进入 bin 目录：

# 启动 ZooKeeper
./zookeeper-server-start.sh ../config/zookeeper.properties
（3）验证启动

ZooKeeper 启动成功后，日志会显示 binding to port 0.0.0.0/0.0.0.0:2181，表示已监听 2181 端口。

Kafka 启动流程

Kafka 核心配置（server.properties）

启动前需确保 config/server.properties 配置正确，关键配置包括：

阅读全文 »

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 2.3k 阅读时长 ≈ 2 分钟

CyclicBarrier和CountDownLatch

这两个类都在jdk的并发包中，都可以用来表示代码运行到某个点上

两者的区别

  • CyclicBarrier表示达到一定数量的线程才会运行；CountDownLatch每来一个线程进行减一操作，直到0为止
  • CyclicBarrier只能唤起一个任务；CountDownLatch可以唤起多个任务
  • CyclicBarrier可重用；CountDownLatch不可重用，只能触发一次事件，值为0后就不可再用了
  • CyclicBarrier允许N个线程相互等待；CountDownLatch是允许1或N个线程等待其他线程完成执行

核心区别对比表

特性 CyclicBarrier CountDownLatch
计数器机制 初始值为 parties，每次 await() 减 1，归 0 后自动重置 初始值为 count，每次 countDown() 减 1，归 0 后不可用
线程协作模式 N 个线程互相等待，全部到达屏障后继续执行 1 个（或多个）线程等待其他线程完成操作
重用性 支持循环使用（通过 reset() 或自动重置） 一次性使用，计数器归 0 后无法重置
屏障动作 支持 Runnable 任务，在所有线程到达后执行 无特殊动作，仅唤醒等待线程
典型场景 并行计算的多阶段同步（如 MapReduce） 主线程等待多个子线程完成初始化
阅读全文 »

发表于 更新于 分类于 Waline： 阅读次数： 本文字数： 15k 阅读时长 ≈ 13 分钟

MyBatis 拦截器（插件）深度解析：从原理到实战扩展

MyBatis 拦截器（又称插件）是 MyBatis 最强大的扩展机制，允许开发者在不修改 MyBatis 核心源码的前提下，通过动态代理对 SQL 执行流程中的关键节点进行拦截，插入自定义逻辑（如日志记录、性能监控、SQL 改写、参数加密等）。从 “核心定位→接口设计→拦截原理→实战开发” 四个维度，彻底拆解 MyBatis 拦截器的工作机制。

拦截器核心定位与拦截对象

MyBatis 拦截器的核心是 “拦截四大核心对象的方法调用”，这四大对象是 MyBatis 执行 SQL 的关键组件，覆盖 “SQL 执行→参数处理→结果映射” 全流程。

四大可拦截对象与拦截点

MyBatis 仅允许拦截以下四类核心对象的特定方法，其他对象无法通过拦截器扩展：

可拦截对象 核心作用 允许拦截的方法（示例） 典型应用场景
Executor SQL 执行器（调度 StatementHandler） query()、update()、commit()、rollback() 全局 SQL 日志、缓存扩展、性能监控
ParameterHandler 参数处理器（绑定 SQL 参数） getParameterObject()、setParameters() 参数加密 / 解密、参数校验
ResultSetHandler 结果集处理器（映射结果） handleResultSets()、handleOutputParameters() 结果加密 / 解密、结果过滤、数据脱敏
StatementHandler SQL 语句处理器（创建 Statement） prepare()、parameterize()、update()、query() SQL 改写（如动态加租户条件）、执行日志

注意：拦截器只能拦截上述对象的特定方法（需通过 @Signature 精确匹配方法名和参数列表），无法拦截对象的所有方法。

拦截器的核心价值

  • 无侵入扩展：无需修改 MyBatis 核心代码，通过配置即可接入自定义逻辑；
  • 粒度可控：可精确指定拦截的对象、方法和参数，避免全局影响；
  • 责任链模式：支持多个拦截器叠加，形成拦截链，按配置顺序执行。

拦截器核心接口与注解

MyBatis 为拦截器定义了统一的接口规范（Interceptor）和注解（@Intercepts/@Signature），开发者需遵循该规范实现自定义拦截器。

1. Interceptor 接口：拦截器的标准定义

Interceptor 是所有自定义拦截器的顶层接口，定义了三个核心方法，分别对应 “拦截逻辑”“代理生成” 和 “参数初始化”：

阅读全文 »
0%