长期的应试教育让我花了太久才明白：学习其实是一件只与自己相关的事。 现在的我，更愿意主动探索那些真正能激发好奇心、让生活变得更有趣的事物。 如果你有能够启发我、让我产生兴趣的想法、机会或故事，欢迎写信给我： （ljr0115@outlook.com） 我会优先回复那些 与我的时间安排契合、或与我当下关注方向相吻合 的邮件。
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实现） */ @Slf4j public class ZkServiceDiscovery implements ServiceDiscovery { private final LoadBalance loadBalance; public ZkServiceDiscovery() { this.loadBalance = new RandomLoadBalance(); } @Override public InetSocketAddress lookupService(String rpcServiceName) { CuratorFramework zkClient = CuratorUtils.getZkClient(); List<String> serviceUrlList = CuratorUtils.getChildrenNodes(zkClient, rpcServiceName); if (serviceUrlList.size() == 0) { throw new RpcException(RpcErrorMessage.SERVICE_CAN_NOT_BE_FOUND, rpcServiceName); } // load balancing String targetServiceUrl = loadBalance.selectServiceAddress(serviceUrlList); log.info("Successfully found the service address:[{}]", targetServiceUrl); String[] socketAddressArray = targetServiceUrl.split(":"); String host = socketAddressArray[0]; int port = Integer.parseInt(socketAddressArray[1]); return new InetSocketAddress(host, port); } }我们根据完整的服务名称便可以将对应的服务地址查出来， 查出来的服务地址可能并不止一个。所以，我们可以通过对应的负载均衡策略来选择出一个服务地址。**CuratorUtils.java**另外，我们还自定义了一个 ZooKeeper Java 客户端 Curtor 的工具类 CuratorUtils.java 。关于这个工具类，这里就不再提了。在《08 Zookeeper 常用命令+ Curtor 使用详解》中已经介绍的非常详细了。
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（代理） : 可以看作是一个独立的 Kafka 实例。多个 Kafka Broker 组成一个 Kafka Cluster。同时，你一定也注意到每个 Broker 中又包含了 Topic 以及 Partition 这两个重要的概念：Topic（主题） : Producer 将消息发送到特定的主题，Consumer 通过订阅特定的 Topic(主题) 来消费消息。Partition（分区） : Partition 属于 Topic 的一部分。一个 Topic 可以有多个 Partition ，并且同一 Topic 下的 Partition 可以分布在不同的 Broker 上，这也就表明一个 Topic 可以横跨多个 Broker 。这正如我上面所画的图一样。划重点：Kafka 中的 Partition（分区） 实际上可以对应成为消息队列中的队列。这样是不是更好理解一点？Kafka 的多副本机制了解吗？带来了什么好处？还有一点我觉得比较重要的是 Kafka 为分区（Partition）引入了多副本（Replica）机制。分区（Partition）中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。生产者和消费者只与 leader 副本交互。你可以理解为其他副本只是 leader 副本的拷贝，它们的存在只是为了保证消息存储的安全性。当 leader 副本发生故障时会从 follower 中选举出一个 leader,但是 follower 中如果有和 leader 同步程度达不到要求的参加不了 leader 的竞选。Kafka 的多分区（Partition）以及多副本（Replica）机制有什么好处呢？Kafka 通过给特定 Topic 指定多个 Partition, 而各个 Partition 可以分布在不同的 Broker 上, 这样便能提供比较好的并发能力（负载均衡）。Partition 可以指定对应的 Replica 数, 这也极大地提高了消息存储的安全性, 提高了容灾能力，不过也相应的增加了所需要的存储空间。Zookeeper 和 KafkaZookeeper 在 Kafka 中的作用是什么？要想搞懂 zookeeper 在 Kafka 中的作用 一定要自己搭建一个 Kafka 环境然后自己进 zookeeper 去看一下有哪些文件夹和 Kafka 有关，每个节点又保存了什么信息。 一定不要光看不实践，这样学来的也终会忘记！这部分内容参考和借鉴了这篇文章：https://www.jianshu.com/p/a036405f989c 。下图就是我的本地 Zookeeper ，它成功和我本地的 Kafka 关联上（以下文件夹结构借助 idea 插件 Zookeeper tool 实现）。ZooKeeper 主要为 Kafka 提供元数据的管理的功能。从图中我们可以看出，Zookeeper 主要为 Kafka 做了下面这些事情：Broker 注册：在 Zookeeper 上会有一个专门用来进行 Broker 服务器列表记录的节点。每个 Broker 在启动时，都会到 Zookeeper 上进行注册，即到 /brokers/ids 下创建属于自己的节点。每个 Broker 就会将自己的 IP 地址和端口等信息记录到该节点中去Topic 注册：在 Kafka 中，同一个Topic 的消息会被分成多个分区并将其分布在多个 Broker 上，这些分区信息及与 Broker 的对应关系也都是由 Zookeeper 在维护。比如我创建了一个名字为 my-topic 的主题并且它有两个分区，对应到 zookeeper 中会创建这些文件夹：/brokers/topics/my-topic/Partitions/0、/brokers/topics/my-topic/Partitions/1负载均衡：上面也说过了 Kafka 通过给特定 Topic 指定多个 Partition, 而各个 Partition 可以分布在不同的 Broker 上, 这样便能提供比较好的并发能力。 对于同一个 Topic 的不同 Partition，Kafka 会尽力将这些 Partition 分布到不同的 Broker 服务器上。当生产者产生消息后也会尽量投递到不同 Broker 的 Partition 里面。当 Consumer 消费的时候，Zookeeper 可以根据当前的 Partition 数量以及 Consumer 数量来实现动态负载均衡。……使用 Kafka 能否不引入 Zookeeper?在 Kafka 2.8 之前，Kafka 最被大家诟病的就是其重度依赖于 Zookeeper。在 Kafka 2.8 之后，引入了基于 Raft 协议的 KRaft 模式，不再依赖 Zookeeper，大大简化了 Kafka 的架构，让你可以以一种轻量级的方式来使用 Kafka。不过，要提示一下：如果要使用 KRaft 模式的话，建议选择较高版本的 Kafka，因为这个功能还在持续完善优化中。Kafka 3.3.1 版本是第一个将 KRaft（Kafka Raft）共识协议标记为生产就绪的版本。Kafka 消费顺序、消息丢失和重复消费Kafka 如何保证消息的消费顺序？我们在使用消息队列的过程中经常有业务场景需要严格保证消息的消费顺序，比如我们同时发了 2 个消息，这 2 个消息对应的操作分别对应的数据库操作是：更改用户会员等级。根据会员等级计算订单价格。假如这两条消息的消费顺序不一样造成的最终结果就会截然不同。我们知道 Kafka 中 Partition(分区)是真正保存消息的地方，我们发送的消息都被放在了这里。而我们的 Partition(分区) 又存在于 Topic(主题) 这个概念中，并且我们可以给特定 Topic 指定多个 Partition。每次添加消息到 Partition(分区) 的时候都会采用尾加法，如上图所示。 Kafka 只能为我们保证 Partition(分区) 中的消息有序。消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。Kafka 通过偏移量（offset）来保证消息在分区内的顺序性。所以，我们就有一种很简单的保证消息消费顺序的方法：1 个 Topic 只对应一个 Partition。这样当然可以解决问题，但是破坏了 Kafka 的设计初衷。Kafka 中发送 1 条消息的时候，可以指定 topic, partition, key,data（数据） 4 个参数。如果你发送消息的时候指定了 Partition 的话，所有消息都会被发送到指定的 Partition。并且，同一个 key 的消息可以保证只发送到同一个 partition，这个我们可以采用表/对象的 id 来作为 key 。总结一下，对于如何保证 Kafka 中消息消费的顺序，有了下面两种方法：1 个 Topic 只对应一个 Partition。（推荐）发送消息的时候指定 key/Partition。当然不仅仅只有上面两种方法，上面两种方法是我觉得比较好理解的，Kafka 如何保证消息不丢失？生产者丢失消息的情况生产者(Producer) 调用send方法发送消息之后，消息可能因为网络问题并没有发送过去。所以，我们不能默认在调用send方法发送消息之后消息发送成功了。为了确定消息是发送成功，我们要判断消息发送的结果。但是要注意的是 Kafka 生产者(Producer) 使用 send 方法发送消息实际上是异步的操作，我们可以通过 get()方法获取调用结果，但是这样也让它变为了同步操作，示例代码如下：详细代码见我的这篇文章：Kafka 系列第三篇！10 分钟学会如何在 Spring Boot 程序中使用 Kafka 作为消息队列?SendResult<String, Object> sendResult = kafkaTemplate.send(topic, o).get(); if (sendResult.getRecordMetadata() != null) { logger.info("生产者成功发送消息到" + sendResult.getProducerRecord().topic() + "-> " + sendRe sult.getProducerRecord().value().toString()); }但是一般不推荐这么做！可以采用为其添加回调函数的形式，示例代码如下： ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, o); future.addCallback(result -> logger.info("生产者成功发送消息到topic:{} partition:{}的消息", result.getRecordMetadata().topic(), result.getRecordMetadata().partition()), ex -> logger.error("生产者发送消失败，原因：{}", ex.getMessage()));如果消息发送失败的话，我们检查失败的原因之后重新发送即可！另外，这里推荐为 Producer 的retries（重试次数）设置一个比较合理的值，一般是 3 ，但是为了保证消息不丢失的话一般会设置比较大一点。设置完成之后，当出现网络问题之后能够自动重试消息发送，避免消息丢失。另外，建议还要设置重试间隔，因为间隔太小的话重试的效果就不明显了，网络波动一次你 3 次一下子就重试完了。消费者丢失消息的情况我们知道消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。偏移量（offset)表示 Consumer 当前消费到的 Partition(分区)的所在的位置。Kafka 通过偏移量（offset）可以保证消息在分区内的顺序性。当消费者拉取到了分区的某个消息之后，消费者会自动提交了 offset。自动提交的话会有一个问题，试想一下，当消费者刚拿到这个消息准备进行真正消费的时候，突然挂掉了，消息实际上并没有被消费，但是 offset 却被自动提交了。解决办法也比较粗暴，我们手动关闭自动提交 offset，每次在真正消费完消息之后再自己手动提交 offset 。 但是，细心的朋友一定会发现，这样会带来消息被重新消费的问题。比如你刚刚消费完消息之后，还没提交 offset，结果自己挂掉了，那么这个消息理论上就会被消费两次。Kafka 弄丢了消息我们知道 Kafka 为分区（Partition）引入了多副本（Replica）机制。分区（Partition）中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。生产者和消费者只与 leader 副本交互。你可以理解为其他副本只是 leader 副本的拷贝，它们的存在只是为了保证消息存储的安全性。试想一种情况：假如 leader 副本所在的 broker 突然挂掉，那么就要从 follower 副本重新选出一个 leader ，但是 leader 的数据还有一些没有被 follower 副本的同步的话，就会造成消息丢失。设置 acks = all解决办法就是我们设置 acks = all。acks 是 Kafka 生产者(Producer) 很重要的一个参数。acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后就算被成功发送。当我们配置 acks = all 表示只有所有 ISR 列表的副本全部收到消息时，生产者才会接收到来自服务器的响应. 这种模式是最高级别的，也是最安全的，可以确保不止一个 Broker 接收到了消息. 该模式的延迟会很高.设置 replication.factor >= 3为了保证 leader 副本能有 follower 副本能同步消息，我们一般会为 topic 设置 replication.factor >= 3。这样就可以保证每个 分区(partition) 至少有 3 个副本。虽然造成了数据冗余，但是带来了数据的安全性。设置 min.insync.replicas > 1一般情况下我们还需要设置 min.insync.replicas> 1 ，这样配置代表消息至少要被写入到 2 个副本才算是被成功发送。min.insync.replicas 的默认值为 1 ，在实际生产中应尽量避免默认值 1。但是，为了保证整个 Kafka 服务的高可用性，你需要确保 replication.factor > min.insync.replicas 。为什么呢？设想一下假如两者相等的话，只要是有一个副本挂掉，整个分区就无法正常工作了。这明显违反高可用性！一般推荐设置成 replication.factor = min.insync.replicas + 1。设置 unclean.leader.election.enable = falseKafka 0.11.0.0 版本开始 unclean.leader.election.enable 参数的默认值由原来的 true 改为 false我们最开始也说了我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。多个 follower 副本之间的消息同步情况不一样，当我们配置了 unclean.leader.election.enable = false 的话，当 leader 副本发生故障时就不会从 follower 副本中和 leader 同步程度达不到要求的副本中选择出 leader ，这样降低了消息丢失的可能性。Kafka 如何保证消息不重复消费？kafka 出现消息重复消费的原因：服务端侧已经消费的数据没有成功提交 offset（根本原因）。Kafka 侧 由于服务端处理业务时间长或者网络链接等等原因让 Kafka 认为服务假死，触发了分区 rebalance。解决方案：消费消息服务做幂等校验，比如 Redis 的 set、MySQL 的主键等天然的幂等功能。这种方法最有效。将 enable.auto.commit 参数设置为 false，关闭自动提交，开发者在代码中手动提交 offset。那么这里会有个问题：什么时候提交 offset 合适？处理完消息再提交：依旧有消息重复消费的风险，和自动提交一样拉取到消息即提交：会有消息丢失的风险。允许消息延时的场景，一般会采用这种方式。然后，通过定时任务在业务不繁忙（比如凌晨）的时候做数据兜底。Kafka 重试机制在 Kafka 如何保证消息不丢失这里，我们提到了 Kafka 的重试机制。由于这部分内容较为重要，我们这里再来详细介绍一下。网上关于 Spring Kafka 的默认重试机制文章很多，但大多都是过时的，和实际运行结果完全不一样。以下是根据 spring-kafka-2.9.3 源码重新梳理一下。消费失败会怎么样？在消费过程中，当其中一个消息消费异常时，会不会卡住后续队列消息的消费？这样业务岂不是卡住了？生产者代码： for (int i = 0; i < 10; i++) { kafkaTemplate.send(KafkaConst.TEST_TOPIC, String.valueOf(i)) }消费者消代码： @KafkaListener(topics = {KafkaConst.TEST_TOPIC},groupId = "apple") private void customer(String message) throws InterruptedException { log.info("kafka customer:{}",message); Integer n = Integer.parseInt(message); if (n%5==0){ throw new RuntimeException(); } }在默认配置下，当消费异常会进行重试，重试多次后会跳过当前消息，继续进行后续消息的消费，不会一直卡在当前消息。下面是一段消费的日志，可以看出当 test-0@95 重试多次后会被跳过。2023-08-10 12:03:32.918 DEBUG 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler : Skipping seek of: test-0@95 2023-08-10 12:03:32.918 TRACE 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler : Seeking: test-0 to: 96 2023-08-10 12:03:32.918 INFO 9700 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.KafkaConsumer : [Consumer clientId=consumer-apple-1, groupId=apple] Seeking to offset 96 for partition test-0 因此，即使某个消息消费异常，Kafka 消费者仍然能够继续消费后续的消息，不会一直卡在当前消息，保证了业务的正常进行。默认会重试多少次？默认配置下，消费异常会进行重试，重试次数是多少, 重试是否有时间间隔？看源码 FailedRecordTracker 类有个 recovered 函数，返回 Boolean 值判断是否要进行重试，下面是这个函数中判断是否重试的逻辑： @Override public boolean recovered(ConsumerRecord << ? , ? > record, Exception exception, @Nullable MessageListenerContainer container, @Nullable Consumer << ? , ? > consumer) throws InterruptedException { if (this.noRetries) { // 不支持重试 attemptRecovery(record, exception, null, consumer); return true; } // 取已经失败的消费记录集合 Map < TopicPartition, FailedRecord > map = this.failures.get(); if (map == null) { this.failures.set(new HashMap < > ()); map = this.failures.get(); } // 获取消费记录所在的Topic和Partition TopicPartition topicPartition = new TopicPartition(record.topic(), record.partition()); FailedRecord failedRecord = getFailedRecordInstance(record, exception, map, topicPartition); // 通知注册的重试监听器，消息投递失败 this.retryListeners.forEach(rl - > rl.failedDelivery(record, exception, failedRecord.getDeliveryAttempts().get())); // 获取下一次重试的时间间隔 long nextBackOff = failedRecord.getBackOffExecution().nextBackOff(); if (nextBackOff != BackOffExecution.STOP) { this.backOffHandler.onNextBackOff(container, exception, nextBackOff); return false; } else { attemptRecovery(record, exception, topicPartition, consumer); map.remove(topicPartition); if (map.isEmpty()) { this.failures.remove(); } return true; } }其中， BackOffExecution.STOP 的值为 -1。@FunctionalInterface public interface BackOffExecution { long STOP = -1; long nextBackOff(); }nextBackOff 的值调用 BackOff 类的 nextBackOff() 函数。如果当前执行次数大于最大执行次数则返回 STOP，既超过这个最大执行次数后才会停止重试。public long nextBackOff() { this.currentAttempts++; if (this.currentAttempts <= getMaxAttempts()) { return getInterval(); } else { return STOP; } }那么这个 getMaxAttempts 的值又是多少呢？回到最开始，当执行出错会进入 DefaultErrorHandler 。DefaultErrorHandler 默认的构造函数是：public DefaultErrorHandler() { this(null, SeekUtils.DEFAULT_BACK_OFF); }SeekUtils.DEFAULT_BACK_OFF 定义的是:public static final int DEFAULT_MAX_FAILURES = 10; public static final FixedBackOff DEFAULT_BACK_OFF = new FixedBackOff(0, DEFAULT_MAX_FAILURES - 1);DEFAULT_MAX_FAILURES 的值是 10，currentAttempts 从 0 到 9，所以总共会执行 10 次，每次重试的时间间隔为 0。最后，简单总结一下：Kafka 消费者在默认配置下会进行最多 10 次 的重试，每次重试的时间间隔为 0，即立即进行重试。如果在 10 次重试后仍然无法成功消费消息，则不再进行重试，消息将被视为消费失败。如何自定义重试次数以及时间间隔?从上面的代码可以知道，默认错误处理器的重试次数以及时间间隔是由 FixedBackOff 控制的，FixedBackOff 是 DefaultErrorHandler 初始化时默认的。所以自定义重试次数以及时间间隔，只需要在 DefaultErrorHandler 初始化的时候传入自定义的 FixedBackOff 即可。重新实现一个 KafkaListenerContainerFactory ，调用 setCommonErrorHandler 设置新的自定义的错误处理器就可以实现。@Bean public KafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); // 自定义重试时间间隔以及次数 FixedBackOff fixedBackOff = new FixedBackOff(1000, 5); factory.setCommonErrorHandler(new DefaultErrorHandler(fixedBackOff)); factory.setConsumerFactory(consumerFactory); return factory; }如何在重试失败后进行告警?自定义重试失败后逻辑，需要手动实现，以下是一个简单的例子，重写 DefaultErrorHandler 的 handleRemaining 函数，加上自定义的告警等操作。@Slf4j public class DelErrorHandler extends DefaultErrorHandler { public DelErrorHandler(FixedBackOff backOff) { super(null,backOff); } @Override public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) { super.handleRemaining(thrownException, records, consumer, container); log.info("重试多次失败"); // 自定义操作 } }DefaultErrorHandler 只是默认的一个错误处理器，Spring Kafka 还提供了 CommonErrorHandler 接口。手动实现 CommonErrorHandler 就可以实现更多的自定义操作，有很高的灵活性。例如根据不同的错误类型，实现不同的重试逻辑以及业务逻辑等。重试失败后的数据如何再次处理?当达到最大重试次数后，数据会直接被跳过，继续向后进行。当代码修复后，如何重新消费这些重试失败的数据呢？死信队列（Dead Letter Queue，简称 DLQ） 是消息中间件中的一种特殊队列。它主要用于处理无法被消费者正确处理的消息，通常是因为消息格式错误、处理失败、消费超时等情况导致的消息被"丢弃"或"死亡"的情况。当消息进入队列后，消费者会尝试处理它。如果处理失败，或者超过一定的重试次数仍无法被成功处理，消息可以发送到死信队列中，而不是被永久性地丢弃。在死信队列中，可以进一步分析、处理这些无法正常消费的消息，以便定位问题、修复错误，并采取适当的措施。@RetryableTopic 是 Spring Kafka 中的一个注解,它用于配置某个 Topic 支持消息重试，更推荐使用这个注解来完成重试。// 重试 5 次，重试间隔 100 毫秒,最大间隔 1 秒 @RetryableTopic( attempts = "5", backoff = @Backoff(delay = 100, maxDelay = 1000) ) @KafkaListener(topics = {KafkaConst.TEST_TOPIC}, groupId = "apple") private void customer(String message) { log.info("kafka customer:{}", message); Integer n = Integer.parseInt(message); if (n % 5 == 0) { throw new RuntimeException(); } System.out.println(n); }当达到最大重试次数后，如果仍然无法成功处理消息，消息会被发送到对应的死信队列中。对于死信队列的处理，既可以用 @DltHandler 处理，也可以使用 @KafkaListener 重新消费。参考Kafka 官方文档：https://kafka.apache.org/documentation/极客时间—《Kafka 核心技术与实战》第 11 节：无消息丢失配置怎么实现？
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 队列）中，直到其他线程释放该锁。假设线程 A 获取锁成功了，释放锁之前，A 线程自己是可以重复获取此锁的（state 会累加）。这就是可重入性的体现：一个线程可以多次获取同一个锁而不会被阻塞。但是，这也意味着，一个线程必须释放与获取的次数相同的锁，才能让 state 的值回到 0，也就是让锁恢复到未锁定状态。只有这样，其他等待的线程才能有机会获取该锁。线程 A 尝试获取锁的过程如下图所示（图源从 ReentrantLock 的实现看 AQS 的原理及应用 - 美团技术团队）：再以倒计时器 CountDownLatch 以例，任务分为 N 个子线程去执行，state 也初始化为 N（注意 N 要与线程个数一致）。这 N 个子线程开始执行任务，每执行完一个子线程，就调用一次 countDown() 方法。该方法会尝试使用 CAS(Compare and Swap) 操作，让 state 的值减少 1。当所有的子线程都执行完毕后（即 state 的值变为 0），CountDownLatch 会调用 unpark() 方法，唤醒主线程。这时，主线程就可以从 await() 方法（CountDownLatch 中的await() 方法而非 AQS 中的）返回，继续执行后续的操作。AQS 资源共享方式AQS 定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，如ReentrantLock）和Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。一般来说，自定义同步器的共享方式要么是独占，要么是共享，他们也只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但 AQS 也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。自定义同步器同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：使用者继承 AbstractQueuedSynchronizer 并重写指定的方法。将 AQS 组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用。AQS 使用了模板方法模式，自定义同步器时需要重写下面几个 AQS 提供的钩子方法：//独占方式。尝试获取资源，成功则返回true，失败则返回false。 protected boolean tryAcquire(int) //独占方式。尝试释放资源，成功则返回true，失败则返回false。 protected boolean tryRelease(int) //共享方式。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。 protected int tryAcquireShared(int) //共享方式。尝试释放资源，成功则返回true，失败则返回false。 protected boolean tryReleaseShared(int) //该线程是否正在独占资源。只有用到condition才需要去实现它。 protected boolean isHeldExclusively()什么是钩子方法呢？ 钩子方法是一种被声明在抽象类中的方法，一般使用 protected 关键字修饰，它可以是空方法（由子类实现），也可以是默认实现的方法。模板设计模式通过钩子方法控制固定步骤的实现。篇幅问题，这里就不详细介绍模板方法模式了，不太了解的小伙伴可以看看这篇文章：用 Java8 改造后的模板方法模式真的是 yyds!。除了上面提到的钩子方法之外，AQS 类中的其他方法都是 final ，所以无法被其他类重写。常见同步工具类下面介绍几个基于 AQS 的常见同步工具类。Semaphore(信号量)介绍synchronized 和 ReentrantLock 都是一次只允许一个线程访问某个资源，而Semaphore(信号量)可以用来控制同时访问特定资源的线程数量。Semaphore 的使用简单，我们这里假设有 N(N>5) 个线程来获取 Semaphore 中的共享资源，下面的代码表示同一时刻 N 个线程中只有 5 个线程能获取到共享资源，其他线程都会阻塞，只有获取到共享资源的线程才能执行。等到有线程释放了共享资源，其他阻塞的线程才能获取到。// 初始共享资源数量 final Semaphore semaphore = new Semaphore(5); // 获取1个许可 semaphore.acquire(); // 释放1个许可 semaphore.release();当初始的资源个数为 1 的时候，Semaphore 退化为排他锁。Semaphore 有两种模式：。公平模式： 调用 acquire() 方法的顺序就是获取许可证的顺序，遵循 FIFO；非公平模式： 抢占式的。Semaphore 对应的两个构造方法如下：public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }这两个构造方法，都必须提供许可的数量，第二个构造方法可以指定是公平模式还是非公平模式，默认非公平模式。Semaphore 通常用于那些资源有明确访问数量限制的场景比如限流（仅限于单机模式，实际项目中推荐使用 Redis +Lua 来做限流）。原理Semaphore 是共享锁的一种实现，它默认构造 AQS 的 state 值为 permits，你可以将 permits 的值理解为许可证的数量，只有拿到许可证的线程才能执行。以无参 acquire 方法为例，调用semaphore.acquire() ，线程尝试获取许可证，如果 state > 0 的话，则表示可以获取成功，如果 state <= 0 的话，则表示许可证数量不足，获取失败。如果可以获取成功的话(state > 0 )，会尝试使用 CAS 操作去修改 state 的值 state=state-1。如果获取失败则会创建一个 Node 节点加入等待队列，挂起当前线程。// 获取1个许可证 public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); } // 获取一个或者多个许可证 public void acquire(int permits) throws InterruptedException { if (permits < 0) throw new IllegalArgumentException(); sync.acquireSharedInterruptibly(permits); }acquireSharedInterruptibly方法是 AbstractQueuedSynchronizer 中的默认实现。// 共享模式下获取许可证，获取成功则返回，失败则加入等待队列，挂起线程 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获取许可证，arg为获取许可证个数，当获取失败时,则创建一个节点加入等待队列，挂起当前线程。 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }这里再以非公平模式（NonfairSync）的为例，看看 tryAcquireShared 方法的实现。// 共享模式下尝试获取资源(在Semaphore中的资源即许可证): protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } // 非公平的共享模式获取许可证 final int nonfairTryAcquireShared(int acquires) { for (;;) { // 当前可用许可证数量 int available = getState(); /* * 尝试获取许可证，当前可用许可证数量小于等于0时，返回负值，表示获取失败， * 当前可用许可证大于0时才可能获取成功，CAS失败了会循环重新获取最新的值尝试获取 */ int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } }以无参 release 方法为例，调用semaphore.release(); ，线程尝试释放许可证，并使用 CAS 操作去修改 state 的值 state=state+1。释放许可证成功之后，同时会唤醒等待队列中的一个线程。被唤醒的线程会重新尝试去修改 state 的值 state=state-1 ，如果 state > 0 则获取令牌成功，否则重新进入等待队列，挂起线程。// 释放一个许可证 public void release() { sync.releaseShared(1); } // 释放一个或者多个许可证 public void release(int permits) { if (permits < 0) throw new IllegalArgumentException(); sync.releaseShared(permits); }releaseShared方法是 AbstractQueuedSynchronizer 中的默认实现。// 释放共享锁 // 如果 tryReleaseShared 返回 true，就唤醒等待队列中的一个或多个线程。 public final boolean releaseShared(int arg) { //释放共享锁 if (tryReleaseShared(arg)) { //释放当前节点的后置等待节点 doReleaseShared(); return true; } return false; }tryReleaseShared 方法是Semaphore 的内部类 Sync 重写的一个方法， AbstractQueuedSynchronizer中的默认实现仅仅抛出 UnsupportedOperationException 异常。// 内部类 Sync 中重写的一个方法 // 尝试释放资源 protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); // 可用许可证+1 int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); // CAS修改state的值 if (compareAndSetState(current, next)) return true; } }可以看到，上面提到的几个方法底层基本都是通过同步器 sync 实现的。Sync 是 CountDownLatch 的内部类 , 继承了 AbstractQueuedSynchronizer ，重写了其中的某些方法。并且，Sync 对应的还有两个子类 NonfairSync（对应非公平模式） 和 FairSync（对应公平模式）。private static final class Sync extends AbstractQueuedSynchronizer { // ... } static final class NonfairSync extends Sync { // ... } static final class FairSync extends Sync { // ... }实战public class SemaphoreExample { // 请求的数量 private static final int threadCount = 550; public static void main(String[] args) throws InterruptedException { // 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很慢） ExecutorService threadPool = Executors.newFixedThreadPool(300); // 初始许可证数量 final Semaphore semaphore = new Semaphore(20); for (int i = 0; i < threadCount; i++) { final int threadnum = i; threadPool.execute(() -> {// Lambda 表达式的运用 try { semaphore.acquire();// 获取一个许可，所以可运行线程数量为20/1=20 test(threadnum); semaphore.release();// 释放一个许可 } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); System.out.println("finish"); } public static void test(int threadnum) throws InterruptedException { Thread.sleep(1000);// 模拟请求的耗时操作 System.out.println("threadnum:" + threadnum); Thread.sleep(1000);// 模拟请求的耗时操作 } }执行 acquire() 方法阻塞，直到有一个许可证可以获得然后拿走一个许可证；每个 release 方法增加一个许可证，这可能会释放一个阻塞的 acquire() 方法。然而，其实并没有实际的许可证这个对象，Semaphore 只是维持了一个可获得许可证的数量。 Semaphore 经常用于限制获取某种资源的线程数量。当然一次也可以一次拿取和释放多个许可，不过一般没有必要这样做：semaphore.acquire(5);// 获取5个许可，所以可运行线程数量为20/5=4 test(threadnum); semaphore.release(5);// 释放5个许可除了 acquire() 方法之外，另一个比较常用的与之对应的方法是 tryAcquire() 方法，该方法如果获取不到许可就立即返回 false。issue645 补充内容：Semaphore 与 CountDownLatch 一样，也是共享锁的一种实现。它默认构造 AQS 的 state 为 permits。当执行任务的线程数量超出 permits，那么多余的线程将会被放入等待队列 Park,并自旋判断 state 是否大于 0。只有当 state 大于 0 的时候，阻塞的线程才能继续执行,此时先前执行任务的线程继续执行 release() 方法，release() 方法使得 state 的变量会加 1，那么自旋的线程便会判断成功。如此，每次只有最多不超过 permits 数量的线程能自旋成功，便限制了执行任务线程的数量。CountDownLatch （倒计时器）介绍CountDownLatch 允许 count 个线程阻塞在一个地方，直至所有线程的任务都执行完毕。CountDownLatch 是一次性的，计数器的值只能在构造方法中初始化一次，之后没有任何机制再次对其设置值，当 CountDownLatch 使用完毕后，它不能再次被使用。原理CountDownLatch 是共享锁的一种实现，它默认构造 AQS 的 state 值为 count。这个我们通过 CountDownLatch 的构造方法即可看出。public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } private static final class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } //... }当线程调用 countDown() 时，其实使用了tryReleaseShared方法以 CAS 的操作来减少 state，直至 state 为 0 。当 state 为 0 时，表示所有的线程都调用了 countDown 方法，那么在 CountDownLatch 上等待的线程就会被唤醒并继续执行。public void countDown() { // Sync 是 CountDownLatch 的内部类 , 继承了 AbstractQueuedSynchronizer sync.releaseShared(1); }releaseShared方法是 AbstractQueuedSynchronizer 中的默认实现。// 释放共享锁 // 如果 tryReleaseShared 返回 true，就唤醒等待队列中的一个或多个线程。 public final boolean releaseShared(int arg) { //释放共享锁 if (tryReleaseShared(arg)) { //释放当前节点的后置等待节点 doReleaseShared(); return true; } return false; }tryReleaseShared 方法是CountDownLatch 的内部类 Sync 重写的一个方法， AbstractQueuedSynchronizer中的默认实现仅仅抛出 UnsupportedOperationException 异常。// 对 state 进行递减，直到 state 变成 0； // 只有 count 递减到 0 时，countDown 才会返回 true protected boolean tryReleaseShared(int releases) { // 自选检查 state 是否为 0 for (;;) { int c = getState(); // 如果 state 已经是 0 了，直接返回 false if (c == 0) return false; // 对 state 进行递减 int nextc = c-1; // CAS 操作更新 state 的值 if (compareAndSetState(c, nextc)) return nextc == 0; } }以无参 await方法为例，当调用 await() 的时候，如果 state 不为 0，那就证明任务还没有执行完毕，await() 就会一直阻塞，也就是说 await() 之后的语句不会被执行（main 线程被加入到等待队列也就是 CLH 队列中了）。然后，CountDownLatch 会自旋 CAS 判断 state == 0，如果 state == 0 的话，就会释放所有等待的线程，await() 方法之后的语句得到执行。// 等待（也可以叫做加锁） public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } // 带有超时时间的等待 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); }acquireSharedInterruptibly方法是 AbstractQueuedSynchronizer 中的默认实现。// 尝试获取锁，获取成功则返回，失败则加入等待队列，挂起线程 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); // 尝试获得锁，获取成功则返回 if (tryAcquireShared(arg) < 0) // 获取失败加入等待队列，挂起线程 doAcquireSharedInterruptibly(arg); }tryAcquireShared 方法是CountDownLatch 的内部类 Sync 重写的一个方法，其作用就是判断 state 的值是否为 0，是的话就返回 1，否则返回 -1。protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; }实战CountDownLatch 的两种典型用法：某一线程在开始运行前等待 n 个线程执行完毕 : 将 CountDownLatch 的计数器初始化为 n （new CountDownLatch(n)），每当一个任务线程执行完毕，就将计数器减 1 （countdownlatch.countDown()），当计数器的值变为 0 时，在 CountDownLatch 上 await() 的线程就会被唤醒。一个典型应用场景就是启动一个服务时，主线程需要等待多个组件加载完毕，之后再继续执行。实现多个线程开始执行任务的最大并行性：注意是并行性，不是并发，强调的是多个线程在某一时刻同时开始执行。类似于赛跑，将多个线程放到起点，等待发令枪响，然后同时开跑。做法是初始化一个共享的 CountDownLatch 对象，将其计数器初始化为 1 （new CountDownLatch(1)），多个线程在开始执行任务前首先 coundownlatch.await()，当主线程调用 countDown() 时，计数器变为 0，多个线程同时被唤醒。CountDownLatch 代码示例：public class CountDownLatchExample { // 请求的数量 private static final int THREAD_COUNT = 550; public static void main(String[] args) throws InterruptedException { // 创建一个具有固定线程数量的线程池对象（如果这里线程池的线程数量给太少的话你会发现执行的很慢） // 只是测试使用，实际场景请手动赋值线程池参数 ExecutorService threadPool = Executors.newFixedThreadPool(300); final CountDownLatch countDownLatch = new CountDownLatch(THREAD_COUNT); for (int i = 0; i < THREAD_COUNT; i++) { final int threadNum = i; threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { e.printStackTrace(); } finally { // 表示一个请求已经被完成 countDownLatch.countDown(); } }); } countDownLatch.await(); threadPool.shutdown(); System.out.println("finish"); } public static void test(int threadnum) throws InterruptedException { Thread.sleep(1000); System.out.println("threadNum:" + threadnum); Thread.sleep(1000); } }上面的代码中，我们定义了请求的数量为 550，当这 550 个请求被处理完成之后，才会执行System.out.println("finish");。与 CountDownLatch 的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用 CountDownLatch.await() 方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。其他 N 个线程必须引用闭锁对象，因为他们需要通知 CountDownLatch 对象，他们已经完成了各自的任务。这种通知机制是通过 CountDownLatch.countDown()方法来完成的；每调用一次这个方法，在构造函数中初始化的 count 值就减 1。所以当 N 个线程都调 用了这个方法，count 的值等于 0，然后主线程就能通过 await()方法，恢复执行自己的任务。再插一嘴：CountDownLatch 的 await() 方法使用不当很容易产生死锁，比如我们上面代码中的 for 循环改为：for (int i = 0; i < threadCount-1; i++) { ....... }这样就导致 count 的值没办法等于 0，然后就会导致一直等待。CyclicBarrier(循环栅栏)介绍CyclicBarrier 和 CountDownLatch 非常类似，它也可以实现线程间的技术等待，但是它的功能比 CountDownLatch 更加复杂和强大。主要应用场景和 CountDownLatch 类似。CountDownLatch 的实现是基于 AQS 的，而 CycliBarrier 是基于 ReentrantLock(ReentrantLock 也属于 AQS 同步器)和 Condition 的。CyclicBarrier 的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是：让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活。原理CyclicBarrier 内部通过一个 count 变量作为计数器，count 的初始值为 parties 属性的初始化值，每当一个线程到了栅栏这里了，那么就将计数器减 1。如果 count 值为 0 了，表示这是这一代最后一个线程到达栅栏，就尝试执行我们构造方法中输入的任务。//每次拦截的线程数 private final int parties; //计数器 private int count;下面我们结合源码来简单看看。1、CyclicBarrier 默认的构造方法是 CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用 await() 方法告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。public CyclicBarrier(int parties) { this(parties, null); } public CyclicBarrier(int parties, Runnable barrierAction) { if (parties <= 0) throw new IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }其中，parties 就代表了有拦截的线程的数量，当拦截的线程数量达到这个值的时候就打开栅栏，让所有线程通过。2、当调用 CyclicBarrier 对象调用 await() 方法时，实际上调用的是 dowait(false, 0L)方法。 await() 方法就像树立起一个栅栏的行为一样，将线程挡住了，当拦住的线程数量达到 parties 的值时，栅栏才会打开，线程才得以通过执行。public int await() throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { throw new Error(toe); // cannot happen } }dowait(false, 0L)方法源码分析如下： // 当线程数量或者请求数量达到 count 时 await 之后的方法才会被执行。上面的示例中 count 的值就为 5。 private int count; /** * Main barrier code, covering the various policies. */ private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; // 锁住 lock.lock(); try { final Generation g = generation; if (g.broken) throw new BrokenBarrierException(); // 如果线程中断了，抛出异常 if (Thread.interrupted()) { breakBarrier(); throw new InterruptedException(); } // count 减1 int index = --count; // 当 count 数量减为 0 之后说明最后一个线程已经到达栅栏了，也就是达到了可以执行await 方法之后的条件 if (index == 0) { // tripped boolean ranAction = false; try { final Runnable command = barrierCommand; if (command != null) command.run(); ranAction = true; // 将 count 重置为 parties 属性的初始化值 // 唤醒之前等待的线程 // 下一波执行开始 nextGeneration(); return 0; } finally { if (!ranAction) breakBarrier(); } } // loop until tripped, broken, interrupted, or timed out for (;;) { try { if (!timed) trip.await(); else if (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. Thread.currentThread().interrupt(); } } if (g.broken) throw new BrokenBarrierException(); if (g != generation) return index; if (timed && nanos <= 0L) { breakBarrier(); throw new TimeoutException(); } } } finally { lock.unlock(); } }实战示例 1：public class CyclicBarrierExample1 { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); try { /**等待60秒，保证子线程完全执行结束*/ cyclicBarrier.await(60, TimeUnit.SECONDS); } catch (Exception e) { System.out.println("-----CyclicBarrierException------"); } System.out.println("threadnum:" + threadnum + "is finish"); } }运行结果，如下：threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready threadnum:4is finish threadnum:0is finish threadnum:1is finish threadnum:2is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready threadnum:9is finish threadnum:5is finish threadnum:8is finish threadnum:7is finish threadnum:6is finish ......可以看到当线程数量也就是请求数量达到我们定义的 5 个的时候， await() 方法之后的方法才被执行。另外，CyclicBarrier 还提供一个更高级的构造函数 CyclicBarrier(int parties, Runnable barrierAction)，用于在线程到达屏障时，优先执行 barrierAction，方便处理更复杂的业务场景。示例 2：public class CyclicBarrierExample2 { // 请求的数量 private static final int threadCount = 550; // 需要同步的线程数量 private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> { System.out.println("------当线程数达到之后，优先执行------"); }); public static void main(String[] args) throws InterruptedException { // 创建线程池 ExecutorService threadPool = Executors.newFixedThreadPool(10); for (int i = 0; i < threadCount; i++) { final int threadNum = i; Thread.sleep(1000); threadPool.execute(() -> { try { test(threadNum); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (BrokenBarrierException e) { // TODO Auto-generated catch block e.printStackTrace(); } }); } threadPool.shutdown(); } public static void test(int threadnum) throws InterruptedException, BrokenBarrierException { System.out.println("threadnum:" + threadnum + "is ready"); cyclicBarrier.await(); System.out.println("threadnum:" + threadnum + "is finish"); } }运行结果，如下：threadnum:0is ready threadnum:1is ready threadnum:2is ready threadnum:3is ready threadnum:4is ready ------当线程数达到之后，优先执行------ threadnum:4is finish threadnum:0is finish threadnum:2is finish threadnum:1is finish threadnum:3is finish threadnum:5is ready threadnum:6is ready threadnum:7is ready threadnum:8is ready threadnum:9is ready ------当线程数达到之后，优先执行------ threadnum:9is finish threadnum:5is finish threadnum:6is finish threadnum:8is finish threadnum:7is finish ......参考Java 并发之 AQS 详解：https://www.cnblogs.com/waterystone/p/4920797.html从 ReentrantLock 的实现看 AQS 的原理及应用：https://tech.meituan.com/2019/12/05/aqs-theory-and-apply.html