SakuraMotion 2021 - 2025年10月
type: channel
link: https://yingwu2021.xyz/index.php/2025/10/
description: 长期的应试教育让我花了太久才明白：学习其实是一件只与自己相关的事。 现在的我，更愿意主动探索那些真正能激发好奇心、让生活变得更有趣的事物。 如果你有能够启发我、让我产生兴趣的想法、机会或故事，欢迎写信给我： （ljr0115@outlook.com） 我会优先回复那些 与我的时间安排契合、或与我当下关注方向相吻合 的邮件。
items: 10 RPC 框架代码分析之注册中心模块
items: Kafka 基础
items: AQS 介绍

10 RPC 框架代码分析之注册中心模块
type: item
date: 2025-10-23T05:49:00+08:00
description: AI 摘要 文章剖析自研RPC框架的注册中心模块：定义ServiceRegistry与ServiceDiscovery接口，基于ZooKeeper实现服务注册与发现，支持多版本、多分组、多实例，配合负载均衡策略完成地址选取。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结10 RPC 框架代码分析之注册中心模块我们之前在“如何自己实现一个 RPC 框架？”这篇文章中介绍到说：注册中心负责服务地址的注册与查找，相当于目录服务。 服务端启动的时候将服务名称及其对应的地址(ip+port)注册到注册中心，服务消费端根据服务名称找到对应的服务地址。有了服务地址之后，服务消费端就可以通过网络请求服务端了。简单来说注册中心就像是一个中转站，提供的作用就是根据调用的服务名称找到远程服务的地址（数据保存服务）。注册中心模块整体结构如下：我们定义了两个接口 ServiceDiscovery.java 和 ServiceRegistry.java，这两个接口分别定义了服务发现和服务注册行为。**ServiceRegistry.java**/** * 服务注册 * * @author shuang.kou * @createTime 2020年05月13日 08:39:00 */ public interface ServiceRegistry { /** * 注册服务到注册中心 * * @param rpcServiceName 完整的服务名称（class name+group+version） * @param inetSocketAddress 远程服务地址 */ void registerService(String rpcServiceName, InetSocketAddress inetSocketAddress); }**ServiceDiscovery.java**/** * 服务发现 * * @author shuang.kou * @createTime 2020年06月01日 15:16:00 */ public interface ServiceDiscovery { /** * 根据 rpcServiceName 获取远程服务地址 * * @param rpcServiceName 完整的服务名称（class name+group+version） * @return 远程服务地址 */ InetSocketAddress lookupService(String rpcServiceName); }接下来，我们使用 zookeeper 作为注册中心的实现方式，并实现了这两个接口。**ZkServiceRegistry.java**/** * 服务注册（基于zookeeper实现） */ @Slf4j public class ZkServiceRegistry implements ServiceRegistry { @Override public void registerService(String rpcServiceName, InetSocketAddress inetSocketAddress) { String servicePath = CuratorUtils.ZK_REGISTER_ROOT_PATH + "/" + rpcServiceName + inetSocketAddress.toString(); CuratorFramework zkClient = CuratorUtils.getZkClient(); CuratorUtils.createPersistentNode(zkClient, servicePath); } }当我们的服务被注册进 zookeeper 的时候，我们将完整的服务名称 rpcServiceName （class name+group+version）作为根节点 ，子节点是对应的服务地址（ip+端口号）。class name : 服务接口名也就是类名比如：github.javaguide.HelloService。version : 服务版本。主要是为后续不兼容升级提供可能group :服务所在的组。主要用于处理一个接口有多个类实现的情况。一个根节点（rpcServiceName）可能会对应多个服务地址（相同服务被部署多份的情况）。如果我们要获得某个服务对应的地址的话，就直接根据完整的服务名称来获取到其下的所有子节点，然后通过具体的负载均衡策略取出一个就可以了。相关代码如下在 ZkServiceDiscovery.java中已经给出。**ZkServiceDiscovery.java**/** * 服务发现（基于zookeeper实现） */ @Sl…

Kafka 基础
type: item
date: 2025-10-08T15:47:00+08:00
description: AI 摘要 Kafka 是分布式流平台，兼具高吞吐消息队列与实时流处理能力。核心概念：Producer、Consumer、Broker、Topic、Partition；多副本+ISR 保障高可用。2.8 起支持无 Zookeeper 的 KRaft。顺序性靠单分区或 key 路由；防丢需 acks=all、手动提交、replica≥3；去重靠幂等或死信队列；默认重试 10 次可自定义，失败入死信队列。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: Kafka常见问题总结category: 高性能tag:消息队列Kafka 基础Kafka 是什么？主要应用场景有哪些？Kafka 是一个分布式流式处理平台。这到底是什么意思呢？流平台具有三个关键功能：消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是 Kafka 也被归类为消息队列的原因。容错的持久方式存储记录消息流：Kafka 会把消息持久化到磁盘，有效避免了消息丢失的风险。流式处理平台： 在消息发布的时候进行处理，Kafka 提供了一个完整的流式处理类库。Kafka 主要有两大应用场景：消息队列：建立实时流数据管道，以可靠地在系统或应用程序之间获取数据。数据处理： 构建实时的流数据处理程序来转换或处理数据流。和其他消息队列相比,Kafka 的优势在哪里？我们现在经常提到 Kafka 的时候就已经默认它是一个非常优秀的消息队列了，我们也会经常拿它跟 RocketMQ、RabbitMQ 对比。我觉得 Kafka 相比其他消息队列主要的优势如下：极致的性能：基于 Scala 和 Java 语言开发，设计中大量使用了批量处理和异步的思想，最高可以每秒处理千万级别的消息。生态系统兼容性无可匹敌：Kafka 与周边生态系统的兼容性是最好的没有之一，尤其在大数据和流计算领域。实际上在早期的时候 Kafka 并不是一个合格的消息队列，早期的 Kafka 在消息队列领域就像是一个衣衫褴褛的孩子一样，功能不完备并且有一些小问题比如丢失消息、不保证消息可靠性等等。当然，这也和 LinkedIn 最早开发 Kafka 用于处理海量的日志有很大关系，哈哈哈，人家本来最开始就不是为了作为消息队列滴，谁知道后面误打误撞在消息队列领域占据了一席之地。随着后续的发展，这些短板都被 Kafka 逐步修复完善。所以，Kafka 作为消息队列不可靠这个说法已经过时！队列模型了解吗？Kafka 的消息模型知道吗？题外话：早期的 JMS 和 AMQP 属于消息服务领域权威组织所做的相关的标准，我在 JavaGuide的 《消息队列其实很简单》这篇文章中介绍过。但是，这些标准的进化跟不上消息队列的演进速度，这些标准实际上已经属于废弃状态。所以，可能存在的情况是：不同的消息队列都有自己的一套消息模型。队列模型：早期的消息模型使用队列（Queue）作为消息通信载体，满足生产者与消费者模式，一条消息只能被一个消费者使用，未被消费的消息在队列中保留直到被消费或超时。 比如：我们生产者发送 100 条消息的话，两个消费者来消费一般情况下两个消费者会按照消息发送的顺序各自消费一半（也就是你一个我一个的消费。）队列模型存在的问题：假如我们存在这样一种情况：我们需要将生产者产生的消息分发给多个消费者，并且每个消费者都能接收到完整的消息内容。这种情况，队列模型就不好解决了。很多比较杠精的人就说：我们可以为每个消费者创建一个单独的队列，让生产者发送多份。这是一种非常愚蠢的做法，浪费资源不说，还违背了使用消息队列的目的。发布-订阅模型:Kafka 消息模型发布-订阅模型主要是为了解决队列模型存在的问题。发布订阅模型（Pub-Sub） 使用主题（Topic） 作为消息通信载体，类似于广播模式；发布者发布一条消息，该消息通过主题传递给所有的订阅者，在一条消息广播之后才订阅的用户则是收不到该条消息的。在发布 - 订阅模型中，如果只有一个订阅者，那它和队列模型就基本是一样的了。所以说，发布 - 订阅模型在功能层面上是可以兼容队列模型的。Kafka 采用的就是发布 - 订阅模型。RocketMQ 的消息模型和 Kafka 基本是完全一样的。唯一的区别是 Kafka 中没有队列这个概念，与之对应的是 Partition（分区）。Kafka 核心概念什么是 Producer、Consumer、Broker、Topic、Partition？Kafka 将生产者发布的消息发送到 Topic（主题） 中，需要这些消息的消费者可以订阅这些 Topic（主题），如下图所示：上面这张图也为我们引出了，Kafka 比较重要的几个概念：Producer（生产者） : 产生消息的一方。Consumer（消费者） : 消费消息的一方。Broker（…

AQS 介绍
type: item
date: 2025-10-05T01:24:00+08:00
description: AI 摘要 AQS是Java并发包核心抽象队列同步器，用volatile int state与CLH队列管理锁状态，支持独占/共享两种模式。ReentrantLock、Semaphore、CountDownLatch等工具均基于其模板方法模式实现。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: AQS 详解category: Javatag:Java并发AQS 介绍AQS 的全称为 AbstractQueuedSynchronizer ，翻译过来的意思就是抽象队列同步器。这个类在 java.util.concurrent.locks 包下面。AQS 就是一个抽象类，主要用来构建锁和同步器。public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { }AQS 为构建锁和同步器提供了一些通用功能的实现，因此，使用 AQS 能简单且高效地构造出应用广泛的大量的同步器，比如我们提到的 ReentrantLock，Semaphore，其他的诸如 ReentrantReadWriteLock，SynchronousQueue等等皆是基于 AQS 的。AQS 原理在面试中被问到并发知识的时候，大多都会被问到“请你说一下自己对于 AQS 原理的理解”。下面给大家一个示例供大家参考，面试不是背题，大家一定要加入自己的思想，即使加入不了自己的思想也要保证自己能够通俗的讲出来而不是背出来。AQS 核心思想AQS 核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态。如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制，这个机制 AQS 是基于 CLH 锁 （Craig, Landin, and Hagersten locks） 实现的。CLH 锁是对自旋锁的一种改进，是一个虚拟的双向队列（虚拟的双向队列即不存在队列实例，仅存在结点之间的关联关系），暂时获取不到锁的线程将被加入到该队列中。AQS 将每条请求共享资源的线程封装成一个 CLH 队列锁的一个结点（Node）来实现锁的分配。在 CLH 队列锁中，一个节点表示一个线程，它保存着线程的引用（thread）、 当前节点在队列中的状态（waitStatus）、前驱节点（prev）、后继节点（next）。CLH 队列结构如下图所示：关于 AQS 核心数据结构-CLH 锁的详细解读，强烈推荐阅读 Java AQS 核心数据结构-CLH 锁 - Qunar 技术沙龙 这篇文章。AQS(AbstractQueuedSynchronizer)的核心原理图：AQS 使用 int 成员变量 state 表示同步状态，通过内置的 FIFO 线程等待/等待队列 来完成获取资源线程的排队工作。state 变量由 volatile 修饰，用于展示当前临界资源的获锁情况。// 共享变量，使用volatile修饰保证线程可见性 private volatile int state;另外，状态信息 state 可以通过 protected 类型的getState()、setState()和compareAndSetState() 进行操作。并且，这几个方法都是 final 修饰的，在子类中无法被重写。//返回同步状态的当前值 protected final int getState() { return state; } // 设置同步状态的值 protected final void setState(int newState) { state = newState; } //原子地（CAS操作）将同步状态值设置为给定值update如果当前同步状态的值等于expect（期望值） protected final boolean compareAndSetState(int expect, int update) { return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }以可重入的互斥锁 ReentrantLock 为例，它的内部维护了一个 state 变量，用来表示锁的占用状态。state 的初始值为 0，表示锁处于未锁定状态。当线程 A 调用 lock() 方法时，会尝试通过 tryAcquire() 方法独占该锁，并让 state 的值加 1。如果成功了，那么线程 A 就获取到了锁。如果失败了，那么线程 A 就会被加入到一个等待队列（CLH 队列）中，直到其他线…