AI 摘要 文章介绍将Netty RPC客户端接收服务端返回结果的方式，从阻塞的AttributeMap改为CompletableFuture，实现异步非阻塞调用，代码更清晰，使用更简单，并建议限制未处理请求Map容量防OOM。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结12 （优化）使用CompletableFuture优化接受服务提供端返回结果使用 AttributeMap 接受服务端返回结果最开始的时候是通过 AttributeMap 绑定到Channel上实现的，相关代码如下：NettyClientTransport.java（用来发送 RpcRequest 请求）NettyClientHandler.java （自定义客户端 ChannelHandler 来处理服务端发过来的数据）这种是实现的缺点是不清晰，而且你每次都要调用 channel.closeFuture().sync(); 阻塞来手动等待请求返回。使用 CompletableFuture 进行优化我使用 CompletableFuture 包装返回结果，对代码进行了重构，重要部分的代码如下：NettyClientTransport.java（用来发送 RpcRequest 请求）NettyClientHandler.java （自定义客户端 ChannelHandler 来处理服务端发过来的数据）UnprocessedRequests.java 存放了未处理的请求（建议限制 map 容器大小，避免未处理请求过多 OOM现在，在你只需要通过下面的方式就能成功接收到客户端返回的结果：CompletableFuture<RpcResponse> completableFuture = (CompletableFuture<RpcResponse>) clientTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get();
AI 摘要 文章剖析RPC框架的“其他模块”：用动态代理RpcClientProxy屏蔽网络细节，统一封装请求；借Spring BeanPostProcessor，通过@RpcService发布服务、@RpcReference注入代理，实现注解式注册与消费。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结11 RPC 框架代码分析之其他模块动态代理屏蔽网络传输细节我们在前面的章节讲到过我们需要用到动态代理来屏蔽复杂的网络传输细节。对应的代码: RpcClientProxy.java@Slf4j public class RpcClientProxy implements InvocationHandler { public <T> T getProxy(Class<T> clazz) { return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this); } @Override public Object invoke(Object proxy, Method method, Object[] args) { } }当我们去调用一个远程的方法的时候，实际上是通过代理对象调用的。获取代理对象的方法如下： public <T> T getProxy(Class<T> clazz) { return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class<?>[]{clazz}, this); }网络传输细节都被封装在了 invoke() 方法中。 public Object invoke(Object proxy, Method method, Object[] args) { log.info("invoked method: [{}]", method.getName()); RpcRequest rpcRequest = RpcRequest.builder().methodName(method.getName()) .parameters(args) .interfaceName(method.getDeclaringClass().getName()) .paramTypes(method.getParameterTypes()) .requestId(UUID.randomUUID().toString()) .group(rpcServiceProperties.getGroup()) .version(rpcServiceProperties.getVersion()) .build(); RpcResponse<Object> rpcResponse = null; if (rpcRequestTransport instanceof NettyRpcClient) { CompletableFuture<RpcResponse<Object>> completableFuture = (CompletableFuture<RpcResponse<Object>>) rpcRequestTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get(); } if (rpcRequestTransport instanceof SocketRpcClient) { rpcResponse = (RpcResponse<Object>) rpcRequestTransport.sendRpcRequest(rpcRequest); } this.check(rpcResponse, rpcRequest); return rpcResponse.getData(); }通过注解注册/消费服务我们这里借用了 Spring 容器相关的功能。核心代码都放在了 ： src/main/java/github/javaguide/spring 包下面。我们定义两个注解：RcpService ：注册服务RpcReference ：消费服务**RcpService.java**@Documented @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.TYPE}) @Inherited public @interface RpcService { /** * Service version, default value is empty string */ String version() default ""; /** * Service group, default value is empty string */ String group() default ""; }**RpcReference.java**@Documented @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.FIELD}) @Inherited public @interface RpcReference { /** * Service version, default value is empty string */ String version() default ""; /** * Service group, default value is empty string */ String group() default ""; }简单说一下原理。我们实现需要 BeanPostProcessor 接口并重写 postProcessBeforeInitialization()方法和 postProcessAfterInitialization() 方法。Spring bean 在实例化之前会调用 postProcessBeforeInitialization()方法，在 Spring bean 实例化之后会调用 postProcessAfterInitialization() 方法。@Slf4j @Component public class SpringBeanPostProcessor implements BeanPostProcessor { @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { } @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { } }被我们使用 RpcService和RpcReference 注解的类都算是 Spring Bean。我们可以在postProcessBeforeInitialization()方法中去判断类上是否有RpcService 注解。如果有的话，就取出 group 和 version 的值。然后，再调用 ServiceProvider 的 publishService() 方法发布服务即可！我们可以在 postProcessAfterInitialization() 方法中遍历类的属性上是否有 RpcReference 注解。如果有的话，我们就通过反射将这个属性赋值即可！
AI 摘要 数据冷热分离按访问频率与业务价值将数据分为热、冷两类，分别存于高性能与低成本介质，以提升查询速度、节省成本；常用时间或访问频率划分，迁移可定时任务或监听binlog，冷数据可选MySQL、Hbase等存储。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: 数据冷热分离详解category: 高性能head:metaname: keywordscontent: 数据冷热分离,冷数据迁移,冷数据存储metaname: descriptioncontent: 数据冷热分离是指根据数据的访问频率和业务重要性，将数据分为冷数据和热数据，冷数据一般存储在存储在低成本、低性能的介质中，热数据高性能存储介质中。什么是数据冷热分离？数据冷热分离是指根据数据的访问频率和业务重要性，将数据分为冷数据和热数据，冷数据一般存储在存储在低成本、低性能的介质中，热数据高性能存储介质中。冷数据和热数据热数据是指经常被访问和修改且需要快速访问的数据，冷数据是指不经常访问，对当前项目价值较低，但需要长期保存的数据。冷热数据到底如何区分呢？有两个常见的区分方法：时间维度区分：按照数据的创建时间、更新时间、过期时间等，将一定时间段内的数据视为热数据，超过该时间段的数据视为冷数据。例如，订单系统可以将 1 年前的订单数据作为冷数据，1 年内的订单数据作为热数据。这种方法适用于数据的访问频率和时间有较强的相关性的场景。访问频率区分：将高频访问的数据视为热数据，低频访问的数据视为冷数据。例如，内容系统可以将浏览量非常低的文章作为冷数据，浏览量较高的文章作为热数据。这种方法需要记录数据的访问频率，成本较高，适合访问频率和数据本身有较强的相关性的场景。几年前的数据并不一定都是热数据，例如一些优质文章发表几年后依然有很多人访问，大部分普通用户新发表的文章却基本没什么人访问。这两种区分冷热数据的方法各有优劣，实际项目中，可以将两者结合使用。冷热分离的思想冷热分离的思想非常简单，就是对数据进行分类，然后分开存储。冷热分离的思想可以应用到很多领域和场景中，而不仅仅是数据存储，例如：邮件系统中，可以将近期的比较重要的邮件放在收件箱，将比较久远的不太重要的邮件存入归档。日常生活中，可以将常用的物品放在显眼的位置，不常用的物品放入储藏室或者阁楼。图书馆中，可以将最受欢迎和最常借阅的图书单独放在一个显眼的区域，将较少借阅的书籍放在不起眼的位置。……数据冷热分离的优缺点优点：热数据的查询性能得到优化（用户的绝大部分操作体验会更好）、节约成本（可以冷热数据的不同存储需求，选择对应的数据库类型和硬件配置，比如将热数据放在 SSD 上，将冷数据放在 HDD 上）缺点：系统复杂性和风险增加（需要分离冷热数据，数据错误的风险增加）、统计效率低（统计的时候可能需要用到冷库的数据）。冷数据如何迁移？冷数据迁移方案：业务层代码实现：当有对数据进行写操作时，触发冷热分离的逻辑，判断数据是冷数据还是热数据，冷数据就入冷库，热数据就入热库。这种方案会影响性能且冷热数据的判断逻辑不太好确定，还需要修改业务层代码，因此一般不会使用。任务调度：可以利用 xxl-job 或者其他分布式任务调度平台定时去扫描数据库，找出满足冷数据条件的数据，然后批量地将其复制到冷库中，并从热库中删除。这种方法修改的代码非常少，非常适合按照时间区分冷热数据的场景。监听数据库的变更日志 binlog ：将满足冷数据条件的数据从 binlog 中提取出来，然后复制到冷库中，并从热库中删除。这种方法可以不用修改代码，但不适合按照时间维度区分冷热数据的场景。如果你的公司有 DBA 的话，也可以让 DBA 进行冷数据的人工迁移，一次迁移完成冷数据到冷库。然后，再搭配上面介绍的方案实现后续冷数据的迁移工作。冷数据如何存储？冷数据的存储要求主要是容量大，成本低，可靠性高，访问速度可以适当牺牲。冷数据存储方案：中小厂：直接使用 MySQL/PostgreSQL 即可（不改变数据库选型和项目当前使用的数据库保持一致），比如新增一张表来存储某个业务的冷数据或者使用单独的冷库来存放冷数据（涉及跨库查询，增加了系统复杂性和维护难度）大厂：Hbase（常用）、RocksDB、Doris、Cassandra如果公司成本预算足的话，也可以直接上 TiDB 这种分布式关系型数据库，直接一步到位。TiDB 6.0 正式支持数据冷热存储分离，可以降低 SSD 使用成本。使用 TiDB 6.0 的数据放置功能，可以在同一个集群实现海量数据的冷热存储，将新的热数据存入 SSD，历史冷数据存入 HDD。案例分享如何快速优化几千万数据量的订单表 - 程序员济癫 - 2023海量数据冷热分离方案与实践 - 字节跳动技术团队 - 2022
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 摘要 文章剖析了自研 RPC 框架的网络传输模块：先定义 RpcRequest/Response 实体，再抽象 RpcRequestTransport 接口，分别用 Socket 与 Netty 实现客户端与服务端收发；Netty 版引入编解码器、自定义协议及心跳保活，并借助 CompletableFuture 实现异步回调，完成高性能远程调用。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结09 RPC 框架代码分析之网络传输模块以下提到的 服务端 指的是提供服务/方法的一端，客户端 指的是调用远程(服务端)服务/方法的一端。我们之前在“如何自己实现一个 RPC 框架？”这篇文章中介绍到说：既然我们要调用远程的方法，就要发送网络请求来传递目标类和方法的信息以及方法的参数等数据到服务端。 这就涉及到了网络传输！网络传输具体实现你可以使用 Socket （ Java 中最原始、最基础的网络通信方式。但是，Socket 是阻塞 IO、性能低并且功能单一）。你也可以使用同步非阻塞的 I/O 模型 NIO ，但是用它来进行网络编程真的太麻烦了。不过没关系，你可以使用基于 NIO 的网络编程框架 Netty ，它将是你最好的选择！guide-rpc-framework 使用了一种基于 Socket，一种基于 Netty 的方式（循序渐进）。网络传输模块整体结构如下：一共被分为了 4 个包constants : 存放一些网络传输模块共用的常量dto : 用于网络传输的类。handler : 里面只有一个用于处理 rpc 请求的类RpcRequestHandler（根据 rpc 请求调用目标类的目标方法）。transport : 用户网络传输相关类（真正传输网络请求的地方。提供了 Socket 和 Netty 两种网络传输方式）。网络传输实体类网络传输实体类在 dto 包下，主要有两个类。**RpcRequest.java**rpc 请求实体类。当你要调用远程方法的时候，你需要先传输一个 RpcRequest 给对方，RpcRequest 里面包含了要调用的目标方法和类的名称、参数等数据。另外，version 字段（服务版本）主要是为后续不兼容升级提供可能。group 字段主要用于处理一个接口有多个类实现的情况。@AllArgsConstructor @NoArgsConstructor @Getter @Builder @ToString public class RpcRequest implements Serializable { private static final long serialVersionUID = 1905122041950251207L; private String requestId; private String interfaceName; private String methodName; private Object[] parameters; private Class<?>[] paramTypes; private RpcMessageType rpcMessageType; private String version; private String group; public RpcServiceProperties toRpcProperties() { return RpcServiceProperties.builder().serviceName(this.getInterfaceName()) .version(this.getVersion()) .group(this.getGroup()).build(); } }**RpcResponse.java**既然有了 rpc 请求实体类，那肯定就要有 rpc 响应实体类了。当服务端通过 RpcRequest 中的相关数据调用到目标服务的目标方法之后，调用结果就通过 RpcResponse 返回给客户端。@AllArgsConstructor @NoArgsConstructor @Getter @Setter @Builder @ToString public class RpcResponse<T> implements Serializable { private static final long serialVersionUID = 715745410605631233L; private String requestId; /** * response code */ private Integer code; /** * response message */ private String message; /** * response body */ private T data; public static <T> RpcResponse<T> success(T data, String requestId) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(RpcResponseCode.SUCCESS.getCode()); response.setMessage(RpcResponseCode.SUCCESS.getMessage()); response.setRequestId(requestId); if (null != data) { response.setData(data); } return response; } public static <T> RpcResponse<T> fail(RpcResponseCode rpcResponseCode) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(rpcResponseCode.getCode()); response.setMessage(rpcResponseCode.getMessage()); return response; } }网络传输由于，这部分我提供了一种基于 Socket，一种基于 Netty 的网络传输方式（循序渐进）。因此，我先定义了一个发送 RPC 请求的顶层接口，然后我们分别使用 Socket 和 Netty 两种方式对这个接口进行实现即可！RpcRequestTransport.java 传输请求的接口/** * send RpcRequest。 * * @author shuang.kou * @createTime 2020年05月29日 13:26:00 */ @SPI public interface RpcRequestTransport { /** * send rpc request to server and get result * * @param rpcRequest message body * @return data from server */ Object sendRpcRequest(RpcRequest rpcRequest); }下面，我们先来看一下比较简单点的使用 Socket 进行网络传输的方式。Socket客户端这里的客户端实际就是发送 RPC 请求的一端，可以对照我们之间画的 RPC 调用的原理图来理解。客户端主要用于发送网络请求到服务端（目标方法所在的服务器）。当我们知道了服务端的地址之后，我们就可以通过 SocketRpcClient 发送 rpc 请求(RpcRequest) 到服务端了(如果我们要找到服务端的地址，涉及到了注册中心相关的知识。下一节会提到。)。我们直接实现上面定义的 RpcRequestTransport.java 即可。这样的话，通过 Socket 来传输消息的模块就写好了！/** * 基于 Socket 传输 RpcRequest * * @author shuang.kou * @createTime 2020年05月10日 18:40:00 */ @AllArgsConstructor @Slf4j public class SocketRpcClient implements RpcRequestTransport { private final ServiceDiscovery serviceDiscovery; public SocketRpcClient() { this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension("zk"); } @Override public Object sendRpcRequest(RpcRequest rpcRequest) { // build rpc service name by rpcRequest String rpcServiceName = RpcServiceProperties.builder().serviceName(rpcRequest.getInterfaceName()) .group(rpcRequest.getGroup()).version(rpcRequest.getVersion()).build().toRpcServiceName(); InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName); try (Socket socket = new Socket()) { socket.connect(inetSocketAddress); ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); // Send data to the server through the output stream objectOutputStream.writeObject(rpcRequest); ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); // Read RpcResponse from the input stream return objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { throw new RpcException("调用服务失败:", e); } } }上面的逻辑很简单，就是对 Socket 发送网络请求这个基础知识的运用。我这里就不再对上面的代码进行解析了，看不懂的小伙伴自行翻看之前关于 Socket 讲解的章节。服务端**SocketRpcServer.java**Socket 服务端。用于等待客户端连接。当客户端成功连接之后，就可以发送 rpc 请求(RpcRequest) 到服务端了。然后，服务端拿到 RpcRequest就会去执行对应的方法。执行完对应的方法之后，就把执行得到的结果放在 RpcResponse 中返回给客户端。/** * @author shuang.kou * @createTime 2020年05月10日 08:01:00 */ @Slf4j public class SocketRpcServer { private final ExecutorService threadPool; private final ServiceProvider serviceProvider; public SocketRpcServer() { threadPool = ThreadPoolFactoryUtils.createCustomThreadPoolIfAbsent("socket-server-rpc-pool"); serviceProvider = SingletonFactory.getInstance(ServiceProviderImpl.class); } public void registerService(Object service) { serviceProvider.publishService(service); } public void registerService(Object service, RpcServiceProperties rpcServiceProperties) { serviceProvider.publishService(service, rpcServiceProperties); } public void start() { try (ServerSocket server = new ServerSocket()) { String host = InetAddress.getLocalHost().getHostAddress(); server.bind(new InetSocketAddress(host, PORT)); CustomShutdownHook.getCustomShutdownHook().clearAll(); Socket socket; while ((socket = server.accept()) != null) { log.info("client connected [{}]", socket.getInetAddress()); threadPool.execute(new SocketRpcRequestHandlerRunnable(socket)); } threadPool.shutdown(); } catch (IOException e) { log.error("occur IOException:", e); } } }NettyNetty 这部分的原理也差不多，不过实现代码差别很大。客户端**NettyClient.java**Netty 客户端主要提供了:**doConnect()** :用于连接服务端（目标方法所在的服务器）并返回对应的 Channel。当我们知道了服务端的地址之后，我们就可以通过 NettyClient 成功连接服务端了。（有了 Channel 之后就能发送数据到服务端了）**sendRpcRequest()** : 用于传输 rpc 请求(RpcRequest) 到服务端。@Slf4j public final class NettyRpcClient implements RpcRequestTransport { private final ServiceDiscovery serviceDiscovery; private final UnprocessedRequests unprocessedRequests; private final ChannelProvider channelProvider; private final Bootstrap bootstrap; private final EventLoopGroup eventLoopGroup; @SneakyThrows public Channel doConnect(InetSocketAddress inetSocketAddress) { CompletableFuture<Channel> completableFuture = new CompletableFuture<>(); bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("The client has connected [{}] successful!", inetSocketAddress.toString()); completableFuture.complete(future.channel()); } else { throw new IllegalStateException(); } }); return completableFuture.get(); } @Override public Object sendRpcRequest(RpcRequest rpcRequest) { // build return value CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>(); // build rpc service name by rpcRequest String rpcServiceName = rpcRequest.toRpcProperties().toRpcServiceName(); // get server address InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcServiceName); // get server address related channel Channel channel = getChannel(inetSocketAddress); if (channel.isActive()) { // put unprocessed request unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture); RpcMessage rpcMessage = new RpcMessage(); rpcMessage.setData(rpcRequest); rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode()); rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode()); rpcMessage.setMessageType(RpcConstants.REQUEST_TYPE); channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("client send message: [{}]", rpcMessage); } else { future.channel().close(); resultFuture.completeExceptionally(future.cause()); log.error("Send failed:", future.cause()); } }); } else { throw new IllegalStateException(); } return resultFuture; } }**UnprocessedRequests.java**用于存放未被服务端处理的请求（建议限制 map 容器大小，避免未处理请求过多 OOM)。public class UnprocessedRequests { private static final Map<String, CompletableFuture<RpcResponse<Object>>> UNPROCESSED_RESPONSE_FUTURES = new ConcurrentHashMap<>(); public void put(String requestId, CompletableFuture<RpcResponse<Object>> future) { UNPROCESSED_RESPONSE_FUTURES.put(requestId, future); } public void complete(RpcResponse<Object> rpcResponse) { CompletableFuture<RpcResponse<Object>> future = UNPROCESSED_RESPONSE_FUTURES.remove(rpcResponse.getRequestId()); if (null != future) { future.complete(rpcResponse); } else { throw new IllegalStateException(); } } }**NettyClientHandler**自定义客户端 ChannelHandler 用于处理服务器发送的数据。@Slf4j public class NettyClientHandler extends ChannelInboundHandlerAdapter { private final UnprocessedRequests unprocessedRequests; private final ChannelProvider channelProvider; public NettyClientHandler() { this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class); this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class); } /** * 读取从服务端返回的消息 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { log.info("client receive msg: [{}]", msg); if (msg instanceof RpcResponse) { RpcResponse<Object> rpcResponse = (RpcResponse<Object>) msg; unprocessedRequests.complete(rpcResponse); } } finally { ReferenceCountUtil.release(msg); } } // Netty 心跳机制相关。保证客户端和服务端的连接不被断掉，避免重连。 @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { //省略部分代码 } }从代码中，可以看出当 rpc 请求被成功处理（客户端收到服务端的执行结果）之后，我们调用了 unprocessedRequests.complete(rpcResponse) 方法，这样的话，你只需要通过下面的方式就能成功接收到服务端返回的结果。CompletableFuture<RpcResponse> completableFuture = (CompletableFuture<RpcResponse>) clientTransport.sendRpcRequest(rpcRequest); rpcResponse = completableFuture.get();**ChannelProvider.java**用于存放 Channel（Channel用于在服务端和客户端之间传输数据）。@Slf4j public class ChannelProvider { private final Map<String, Channel> channelMap; public ChannelProvider() { channelMap = new ConcurrentHashMap<>(); } public Channel get(InetSocketAddress inetSocketAddress) { String key = inetSocketAddress.toString(); // determine if there is a connection for the corresponding address if (channelMap.containsKey(key)) { Channel channel = channelMap.get(key); // if so, determine if the connection is available, and if so, get it directly if (channel != null && channel.isActive()) { return channel; } else { channelMap.remove(key); } } return null; } public void set(InetSocketAddress inetSocketAddress, Channel channel) { String key = inetSocketAddress.toString(); channelMap.put(key, channel); } public void remove(InetSocketAddress inetSocketAddress) { String key = inetSocketAddress.toString(); channelMap.remove(key); log.info("Channel map size :[{}]", channelMap.size()); } }服务端相关**NettyRpcServer.java**Netty 服务端。并监听客户端的连接。另外，还提供了两个用户手动注册服务的方法（_还可以通过注解__RpcService__注册服务，这个后面也会介绍到_）。@Slf4j @Component public class NettyRpcServer { public static final int PORT = 9998; private final ServiceProvider serviceProvider = SingletonFactory.getInstance(ServiceProviderImpl.class); public void registerService(Object service, RpcServiceProperties rpcServiceProperties) { serviceProvider.publishService(service, rpcServiceProperties); } @SneakyThrows public void start() { CustomShutdownHook.getCustomShutdownHook().clearAll(); String host = InetAddress.getLocalHost().getHostAddress(); EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); DefaultEventExecutorGroup serviceHandlerGroup = new DefaultEventExecutorGroup( RuntimeUtil.cpus() * 2, ThreadPoolFactoryUtils.createThreadFactory("service-handler-group", false) ); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) // TCP默认开启了 Nagle 算法，该算法的作用是尽可能的发送大数据快，减少网络传输。TCP_NODELAY 参数的作用就是控制是否启用 Nagle 算法。 .childOption(ChannelOption.TCP_NODELAY, true) // 是否开启 TCP 底层心跳机制 .childOption(ChannelOption.SO_KEEPALIVE, true) //表示系统用于临时存放已完成三次握手的请求的队列的最大长度,如果连接建立频繁，服务器处理创建新连接较慢，可以适当调大这个参数 .option(ChannelOption.SO_BACKLOG, 128) .handler(new LoggingHandler(LogLevel.INFO)) // 当客户端第一次进行请求的时候才会进行初始化 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { // 30 秒之内没有收到客户端请求的话就关闭连接 ChannelPipeline p = ch.pipeline(); p.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS)); p.addLast(new RpcMessageEncoder()); p.addLast(new RpcMessageDecoder()); p.addLast(serviceHandlerGroup, new NettyRpcServerHandler()); } }); // 绑定端口，同步等待绑定成功 ChannelFuture f = b.bind(host, PORT).sync(); // 等待服务端监听端口关闭 f.channel().closeFuture().sync(); } catch (InterruptedException e) { log.error("occur exception when start server:", e); } finally { log.error("shutdown bossGroup and workerGroup"); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); serviceHandlerGroup.shutdownGracefully(); } } }**NettyServerHandler.java**自定义服务端 ChannelHandler 用于处理客户端发送的数据。当客户端发的 rpc 请求(RpcRequest) 来了之后，服务端就会处理 rpc 请求(RpcRequest) ，处理完之后就把得到 rpc 相应(RpcResponse)传输给客户端。@Slf4j public class NettyServerHandler extends ChannelInboundHandlerAdapter { private final RpcRequestHandler rpcRequestHandler; public NettyServerHandler() { this.rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class); } /** * 读取从客户端消息，然后调用目标服务的目标方法并返回给客户端。 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { // 省略部分代码 } // Netty 心跳机制相关。保证客户端和服务端的连接不被断掉，避免重连。 @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { // 省略部分代码 } }传输协议在《如何自己实现一个 RPC 框架》这一节，我们就提到了传输协议的作用。简单来说：通过设计协议，我们定义需要传输哪些类型的数据， 并且还会规定每一种类型的数据应该占多少字节。这样我们在接收到二级制数据之后，就可以正确的解析出我们需要的数据。这有一点像密文传输的感觉。以下便是我们设计的传输协议（编解码器这里会用到！！！）： * 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 * +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+ * | magic code |version | full length | messageType| codec|compress| RequestId | * +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+ * | | * | body | * | | * | ... ... | * +-------------------------------------------------------------------------------------------------------+ * 4B magic code（魔法数） 1B version（版本） 4B full length（消息长度） 1B messageType（消息类型） * 1B compress（压缩类型） 1B codec（序列化类型） 4B requestId（请求的Id）魔法数 ： 通常是 4 个字节。这个魔数主要是为了筛选来到服务端的数据包，有了这个魔数之后，服务端首先取出前面四个字节进行比对，能够在第一时间识别出这个数据包并非是遵循自定义协议的，也就是无效数据包，为了安全考虑可以直接关闭连接以节省资源。序列化器类型 ：标识序列化的方式，比如是使用 Java 自带的序列化，还是 json，kyro 等序列化方式。消息长度 ： 运行时计算出来。......编解码器编解码器这里主要用到了 Kryo 序列化和反序列化以及 Netty 网络传输字节容器 ByteBuf 相关的知识。编解码器的作用主要是让我们在 Netty 进行网络传输所用的对象类型 ByteBuf 与 我们代码层面需要的业务对象之间转换。这部分的代码还是比较多的，小伙伴们可以自己阅读以下，整体逻辑还是比较简单的。一定要先搞懂传输协议之后再去看这部分代码。**RpcMessageDecoder.java**自定义解码器。负责处理"入站"消息，将 ByteBuf 消息格式的对象转换为我们需要的业务对象。网络传输需要通过字节流来实现，ByteBuf 可以看作是 Netty 提供的字节数据的容器，使用它会让我们更加方便地处理字节数据。**RpcMessageEncoder.java**自定义编码器。负责处理"出站"消息，将消息格式转换字节数组然后写入到字节数据的容器 ByteBuf 对象中。
AI 摘要 文章系统梳理了JDK自带监控与故障处理工具：命令行侧jps、jstat、jinfo、jmap、jhat、jstack分别用于查进程、看运行数据、调参数、抓堆转储、离线分析堆、线程快照；可视化侧JConsole、VisualVM提供内存、线程、GC等图形化监控，支持远程连接与插件扩展，可快速定位死锁、OOM等问题。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: JDK监控和故障处理工具总结category: Javatag:JVMJDK 命令行工具这些命令在 JDK 安装目录下的 bin 目录下：jps (JVM Process Status）: 类似 UNIX 的 ps 命令。用于查看所有 Java 进程的启动类、传入参数和 Java 虚拟机参数等信息；jstat（JVM Statistics Monitoring Tool）: 用于收集 HotSpot 虚拟机各方面的运行数据;jinfo (Configuration Info for Java) : Configuration Info for Java,显示虚拟机配置信息;jmap (Memory Map for Java) : 生成堆转储快照;jhat (JVM Heap Dump Browser) : 用于分析 heapdump 文件，它会建立一个 HTTP/HTML 服务器，让用户可以在浏览器上查看分析结果;jstack (Stack Trace for Java) : 生成虚拟机当前时刻的线程快照，线程快照就是当前虚拟机内每一条线程正在执行的方法堆栈的集合。jps:查看所有 Java 进程jps(JVM Process Status) 命令类似 UNIX 的 ps 命令。jps：显示虚拟机执行主类名称以及这些进程的本地虚拟机唯一 ID（Local Virtual Machine Identifier,LVMID）。jps -q：只输出进程的本地虚拟机唯一 ID。C:\Users\SnailClimb>jps 7360 NettyClient2 17396 7972 Launcher 16504 Jps 17340 NettyServerjps -l:输出主类的全名，如果进程执行的是 Jar 包，输出 Jar 路径。C:\Users\SnailClimb>jps -l 7360 firstNettyDemo.NettyClient2 17396 7972 org.jetbrains.jps.cmdline.Launcher 16492 sun.tools.jps.Jps 17340 firstNettyDemo.NettyServerjps -v：输出虚拟机进程启动时 JVM 参数。jps -m：输出传递给 Java 进程 main() 函数的参数。jstat: 监视虚拟机各种运行状态信息jstat（JVM Statistics Monitoring Tool） 使用于监视虚拟机各种运行状态信息的命令行工具。 它可以显示本地或者远程（需要远程主机提供 RMI 支持）虚拟机进程中的类信息、内存、垃圾收集、JIT 编译等运行数据，在没有 GUI，只提供了纯文本控制台环境的服务器上，它将是运行期间定位虚拟机性能问题的首选工具。jstat 命令使用格式：jstat - [-t] [-h<lines>] <vmid> [<interval> [<count>]]比如 jstat -gc -h3 31736 1000 10表示分析进程 id 为 31736 的 gc 情况，每隔 1000ms 打印一次记录，打印 10 次停止，每 3 行后打印指标头部。常见的 option 如下：jstat -class vmid：显示 ClassLoader 的相关信息；jstat -compiler vmid：显示 JIT 编译的相关信息；jstat -gc vmid：显示与 GC 相关的堆信息；jstat -gccapacity vmid：显示各个代的容量及使用情况；jstat -gcnew vmid：显示新生代信息；jstat -gcnewcapcacity vmid：显示新生代大小与使用情况；jstat -gcold vmid：显示老年代和永久代的行为统计，从 jdk1.8 开始,该选项仅表示老年代，因为永久代被移除了；jstat -gcoldcapacity vmid：显示老年代的大小；jstat -gcpermcapacity vmid：显示永久代大小，从 jdk1.8 开始,该选项不存在了，因为永久代被移除了；jstat -gcutil vmid：显示垃圾收集信息；另外，加上 -t参数可以在输出信息上加一个 Timestamp 列，显示程序的运行时间。jinfo: 实时地查看和调整虚拟机各项参数jinfo vmid :输出当前 jvm 进程的全部参数和系统属性 (第一部分是系统的属性，第二部分是 JVM 的参数)。jinfo -flag name vmid :输出对应名称的参数的具体值。比如输出 MaxHeapSize、查看当前 jvm 进程是否开启打印 GC 日志 ( -XX:PrintGCDetails :详细 GC 日志模式，这两个都是默认关闭的)。C:\Users\SnailClimb>jinfo -flag MaxHeapSize 17340 -XX:MaxHeapSize=2124414976 C:\Users\SnailClimb>jinfo -flag PrintGC 17340 -XX:-PrintGC使用 jinfo 可以在不重启虚拟机的情况下，可以动态的修改 jvm 的参数。尤其在线上的环境特别有用,请看下面的例子：jinfo -flag [+|-]name vmid 开启或者关闭对应名称的参数。C:\Users\SnailClimb>jinfo -flag PrintGC 17340 -XX:-PrintGC C:\Users\SnailClimb>jinfo -flag +PrintGC 17340 C:\Users\SnailClimb>jinfo -flag PrintGC 17340 -XX:+PrintGCjmap:生成堆转储快照jmap（Memory Map for Java）命令用于生成堆转储快照。 如果不使用 jmap 命令，要想获取 Java 堆转储，可以使用 “-XX:+HeapDumpOnOutOfMemoryError” 参数，可以让虚拟机在 OOM 异常出现之后自动生成 dump 文件，Linux 命令下可以通过 kill -3 发送进程退出信号也能拿到 dump 文件。jmap 的作用并不仅仅是为了获取 dump 文件，它还可以查询 finalizer 执行队列、Java 堆和永久代的详细信息，如空间使用率、当前使用的是哪种收集器等。和jinfo一样，jmap有不少功能在 Windows 平台下也是受限制的。示例：将指定应用程序的堆快照输出到桌面。后面，可以通过 jhat、Visual VM 等工具分析该堆文件。C:\Users\SnailClimb>jmap -dump:format=b,file=C:\Users\SnailClimb\Desktop\heap.hprof 17340 Dumping heap to C:\Users\SnailClimb\Desktop\heap.hprof ... Heap dump file createdjhat: 分析 heapdump 文件jhat 用于分析 heapdump 文件，它会建立一个 HTTP/HTML 服务器，让用户可以在浏览器上查看分析结果。C:\Users\SnailClimb>jhat C:\Users\SnailClimb\Desktop\heap.hprof Reading from C:\Users\SnailClimb\Desktop\heap.hprof... Dump file created Sat May 04 12:30:31 CST 2019 Snapshot read, resolving... Resolving 131419 objects... Chasing references, expect 26 dots.......................... Eliminating duplicate references.......................... Snapshot resolved. Started HTTP server on port 7000 Server is ready.访问 http://localhost:7000/jstack :生成虚拟机当前时刻的线程快照jstack（Stack Trace for Java）命令用于生成虚拟机当前时刻的线程快照。线程快照就是当前虚拟机内每一条线程正在执行的方法堆栈的集合.生成线程快照的目的主要是定位线程长时间出现停顿的原因，如线程间死锁、死循环、请求外部资源导致的长时间等待等都是导致线程长时间停顿的原因。线程出现停顿的时候通过jstack来查看各个线程的调用堆栈，就可以知道没有响应的线程到底在后台做些什么事情，或者在等待些什么资源。下面是一个线程死锁的代码。我们下面会通过 jstack 命令进行死锁检查，输出死锁信息，找到发生死锁的线程。public class DeadLockDemo { private static Object resource1 = new Object();//资源 1 private static Object resource2 = new Object();//资源 2 public static void main(String[] args) { new Thread(() -> { synchronized (resource1) { System.out.println(Thread.currentThread() + "get resource1"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread() + "waiting get resource2"); synchronized (resource2) { System.out.println(Thread.currentThread() + "get resource2"); } } }, "线程 1").start(); new Thread(() -> { synchronized (resource2) { System.out.println(Thread.currentThread() + "get resource2"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread() + "waiting get resource1"); synchronized (resource1) { System.out.println(Thread.currentThread() + "get resource1"); } } }, "线程 2").start(); } }OutputThread[线程 1,5,main]get resource1 Thread[线程 2,5,main]get resource2 Thread[线程 1,5,main]waiting get resource2 Thread[线程 2,5,main]waiting get resource1线程 A 通过 synchronized (resource1) 获得 resource1 的监视器锁，然后通过Thread.sleep(1000);让线程 A 休眠 1s 为的是让线程 B 得到执行然后获取到 resource2 的监视器锁。线程 A 和线程 B 休眠结束了都开始企图请求获取对方的资源，然后这两个线程就会陷入互相等待的状态，这也就产生了死锁。通过 jstack 命令分析：C:\Users\SnailClimb>jps 13792 KotlinCompileDaemon 7360 NettyClient2 17396 7972 Launcher 8932 Launcher 9256 DeadLockDemo 10764 Jps 17340 NettyServer C:\Users\SnailClimb>jstack 9256输出的部分内容如下：Found one Java-level deadlock: ============================= "线程 2": waiting to lock monitor 0x000000000333e668 (object 0x00000000d5efe1c0, a java.lang.Object), which is held by "线程 1" "线程 1": waiting to lock monitor 0x000000000333be88 (object 0x00000000d5efe1d0, a java.lang.Object), which is held by "线程 2" Java stack information for the threads listed above: =================================================== "线程 2": at DeadLockDemo.lambda$main$1(DeadLockDemo.java:31) - waiting to lock <0x00000000d5efe1c0> (a java.lang.Object) - locked <0x00000000d5efe1d0> (a java.lang.Object) at DeadLockDemo$$Lambda$2/1078694789.run(Unknown Source) at java.lang.Thread.run(Thread.java:748) "线程 1": at DeadLockDemo.lambda$main$0(DeadLockDemo.java:16) - waiting to lock <0x00000000d5efe1d0> (a java.lang.Object) - locked <0x00000000d5efe1c0> (a java.lang.Object) at DeadLockDemo$$Lambda$1/1324119927.run(Unknown Source) at java.lang.Thread.run(Thread.java:748) Found 1 deadlock.可以看到 jstack 命令已经帮我们找到发生死锁的线程的具体信息。JDK 可视化分析工具JConsole:Java 监视与管理控制台JConsole 是基于 JMX 的可视化监视、管理工具。可以很方便的监视本地及远程服务器的 java 进程的内存使用情况。你可以在控制台输入jconsole命令启动或者在 JDK 目录下的 bin 目录找到jconsole.exe然后双击启动。连接 Jconsole如果需要使用 JConsole 连接远程进程，可以在远程 Java 程序启动时加上下面这些参数:-Djava.rmi.server.hostname=外网访问 ip 地址 -Dcom.sun.management.jmxremote.port=60001 //监控的端口号 -Dcom.sun.management.jmxremote.authenticate=false //关闭认证 -Dcom.sun.management.jmxremote.ssl=false在使用 JConsole 连接时，远程进程地址如下：外网访问 ip 地址:60001查看 Java 程序概况内存监控JConsole 可以显示当前内存的详细信息。不仅包括堆内存/非堆内存的整体信息，还可以细化到 eden 区、survivor 区等的使用情况，如下图所示。点击右边的“执行 GC(G)”按钮可以强制应用程序执行一个 Full GC。新生代 GC（Minor GC）:指发生新生代的的垃圾收集动作，Minor GC 非常频繁，回收速度一般也比较快。老年代 GC（Major GC/Full GC）:指发生在老年代的 GC，出现了 Major GC 经常会伴随至少一次的 Minor GC（并非绝对），Major GC 的速度一般会比 Minor GC 的慢 10 倍以上。线程监控类似我们前面讲的 jstack 命令，不过这个是可视化的。最下面有一个"检测死锁 (D)"按钮，点击这个按钮可以自动为你找到发生死锁的线程以及它们的详细信息 。Visual VM:多合一故障处理工具VisualVM 提供在 Java 虚拟机 (Java Virtual Machine, JVM) 上运行的 Java 应用程序的详细信息。在 VisualVM 的图形用户界面中，您可以方便、快捷地查看多个 Java 应用程序的相关信息。Visual VM 官网：https://visualvm.github.io/。VisualVM中文文档:https://visualvm.github.io/documentation.html。下面这段话摘自《深入理解 Java 虚拟机》。VisualVM（All-in-One Java Troubleshooting Tool）是到目前为止随 JDK 发布的功能最强大的运行监视和故障处理程序，官方在 VisualVM 的软件说明中写上了“All-in-One”的描述字样，预示着他除了运行监视、故障处理外，还提供了很多其他方面的功能，如性能分析（Profiling）。VisualVM 的性能分析功能甚至比起 JProfiler、YourKit 等专业且收费的 Profiling 工具都不会逊色多少，而且 VisualVM 还有一个很大的优点：不需要被监视的程序基于特殊 Agent 运行，因此他对应用程序的实际性能的影响很小，使得他可以直接应用在生产环境中。这个优点是 JProfiler、YourKit 等工具无法与之媲美的。VisualVM 基于 NetBeans 平台开发，因此他一开始就具备了插件扩展功能的特性，通过插件扩展支持，VisualVM 可以做到：显示虚拟机进程以及进程的配置、环境信息（jps、jinfo）。监视应用程序的 CPU、GC、堆、方法区以及线程的信息（jstat、jstack）。dump 以及分析堆转储快照（jmap、jhat）。方法级的程序运行性能分析，找到被调用最多、运行时间最长的方法。离线程序快照：收集程序的运行时配置、线程 dump、内存 dump 等信息建立一个快照，可以将快照发送开发者处进行 Bug 反馈。其他 plugins 的无限的可能性……这里就不具体介绍 VisualVM 的使用，如果想了解的话可以看:https://visualvm.github.io/documentation.htmlhttps://www.ibm.com/developerworks/cn/java/j-lo-visualvm/index.html
AI 摘要 文章演示了Docker安装ZooKeeper及常用命令（create/set/get/ls/stat/delete），并详解Curator连接、节点增删改查、监听器的代码实现，覆盖持久/临时/顺序节点与重试策略。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结08 ZooKeeper常用命令+ Curator使用详解guide-rpc-framework 使用了 Zookeeper 来存储服务的相关信息 ，并且使用的是 ZooKeeper Java客户端 Curator 来对 ZooKeeper 进行增删改查等操作。所以，本文就简单介绍一下 ZooKeeper常用命令 以及 Curator 的基本使用。前言这篇文章简单给演示一下 ZooKeeper 常见命令的使用以及 ZooKeeper Java客户端 Curator 的基本使用。介绍到的内容都是最基本的操作，能满足日常工作的基本需要。如果文章有任何需要改善和完善的地方，欢迎在评论区指出，共同进步！ZooKeeper 安装和使用使用Docker 安装 zookeepera.使用 Docker 下载 ZooKeeperdocker pull zookeeper:3.5.8b.运行 ZooKeeperdocker run -d --name zookeeper -p 2181:2181 zookeeper:3.5.8连接 ZooKeeper 服务a.进入ZooKeeper容器中先使用 docker ps 查看 ZooKeeper 的 ContainerID，然后使用 docker exec -it ContainerID /bin/bash 命令进入容器中。b.先进入 bin 目录,然后通过 **./zkCli.sh -server 127.0.0.1:2181**命令连接ZooKeeper 服务root@eaf70fc620cb:/apache-zookeeper-3.5.8-bin# cd bin如果你看到控制台成功打印出如下信息的话，说明你已经成功连接 ZooKeeper 服务。常用命令演示查看常用命令(help 命令)通过 help 命令查看 ZooKeeper 常用命令创建节点(create 命令)通过 create 命令在根目录创建了 node1 节点，与它关联的字符串是"node1"[zk: 127.0.0.1:2181(CONNECTED) 34] create /node1 “node1”通过 create 命令在根目录创建了 node1 节点，与它关联的内容是数字 123[zk: 127.0.0.1:2181(CONNECTED) 1] create /node1/node1.1 123 Created /node1/node1.1更新节点数据内容(set 命令)[zk: 127.0.0.1:2181(CONNECTED) 11] set /node1 "set node1"获取节点的数据(get 命令)get 命令可以获取指定节点的数据内容和节点的状态,可以看出我们通过 set 命令已经将节点数据内容改为 "set node1"。set node1 cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x4b mtime = Sun Jan 20 10:41:10 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 1 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 9 numChildren = 1查看某个目录下的子节点(ls 命令)通过 ls 命令查看根目录下的节点[zk: 127.0.0.1:2181(CONNECTED) 37] ls / [dubbo, ZooKeeper, node1]通过 ls 命令查看 node1 目录下的节点[zk: 127.0.0.1:2181(CONNECTED) 5] ls /node1 [node1.1]ZooKeeper 中的 ls 命令和 linux 命令中的 ls 类似， 这个命令将列出绝对路径 path 下的所有子节点信息（列出 1 级，并不递归）查看节点状态(stat 命令)通过 stat 命令查看节点状态[zk: 127.0.0.1:2181(CONNECTED) 10] stat /node1 cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x47 mtime = Sun Jan 20 10:22:59 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 11 numChildren = 1上面显示的一些信息比如 cversion、aclVersion、numChildren 等等，我在上面 “ZooKeeper 相关概念总结(入门)” 这篇文章中已经介绍到。查看节点信息和状态(ls2 命令)ls2 命令更像是 ls 命令和 stat 命令的结合。 ls2 命令返回的信息包括 2 部分：子节点列表当前节点的 stat 信息。[zk: 127.0.0.1:2181(CONNECTED) 7] ls2 /node1 [node1.1] cZxid = 0x47 ctime = Sun Jan 20 10:22:59 CST 2019 mZxid = 0x47 mtime = Sun Jan 20 10:22:59 CST 2019 pZxid = 0x4a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 11 numChildren = 1删除节点(delete 命令)这个命令很简单，但是需要注意的一点是如果你要删除某一个节点，那么这个节点必须无子节点才行。[zk: 127.0.0.1:2181(CONNECTED) 3] delete /node1/node1.1在后面我会介绍到 Java 客户端 API 的使用以及开源 ZooKeeper 客户端 ZkClient 和 Curator 的使用。ZooKeeper Java客户端 Curator简单使用Curator 是Netflix公司开源的一套 ZooKeeper Java客户端框架，相比于 Zookeeper 自带的客户端 zookeeper 来说，Curator 的封装更加完善，各种 API 都可以比较方便地使用。下面我们就来简单地演示一下 Curator 的使用吧！Curator4.0+版本对ZooKeeper 3.5.x支持比较好。开始之前，请先将下面的依赖添加进你的项目。<dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.2.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.2.0</version> </dependency>连接 ZooKeeper 客户端通过 CuratorFrameworkFactory 创建 CuratorFramework 对象，然后再调用 CuratorFramework 对象的 start() 方法即可！private static final int BASE_SLEEP_TIME = 1000; private static final int MAX_RETRIES = 3; // Retry strategy. Retry 3 times, and will increase the sleep time between retries. RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES); CuratorFramework zkClient = CuratorFrameworkFactory.builder() // the server to connect to (can be a server list) .connectString("127.0.0.1:2181") .retryPolicy(retryPolicy) .build(); zkClient.start();对于一些基本参数的说明：baseSleepTimeMs：重试之间等待的初始时间maxRetries ：最大重试次数connectString ：要连接的服务器列表retryPolicy ：重试策略数据节点的增删改查创建节点我们在 ZooKeeper常见概念解读 中介绍到，我们通常是将 znode 分为 4 大类：持久（PERSISTENT）节点 ：一旦创建就一直存在即使 ZooKeeper 集群宕机，直到将其删除。临时（EPHEMERAL）节点 ：临时节点的生命周期是与 客户端会话（session） 绑定的，会话消失则节点消失 。并且，临时节点 只能做叶子节点 ，不能创建子节点。持久顺序（PERSISTENT_SEQUENTIAL）节点 ：除了具有持久（PERSISTENT）节点的特性之外， 子节点的名称还具有顺序性。比如 /node1/app0000000001 、/node1/app0000000002 。临时顺序（EPHEMERAL_SEQUENTIAL）节点 ：除了具备临时（EPHEMERAL）节点的特性之外，子节点的名称还具有顺序性。你在使用的 ZooKeeper 的时候，会发现 CreateMode 类中实际有 7种 znode 类型 ，但是用的最多的还是上面介绍的 4 种。a.创建持久化节点你可以通过下面两种方式创建持久化的节点。//注意:下面的代码会报错，下文说了具体原因 zkClient.create().forPath("/node1/00001"); zkClient.create().withMode(CreateMode.PERSISTENT).forPath("/node1/00002");但是，你运行上面的代码会报错，这是因为的父节点node1还未创建。你可以先创建父节点 node1 ，然后再执行上面的代码就不会报错了。zkClient.create().forPath("/node1");更推荐的方式是通过下面这行代码， **creatingParentsIfNeeded()** 可以保证父节点不存在的时候自动创建父节点，这是非常有用的。zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/node1/00001");b.创建临时节点zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001");c.创建节点并指定数据内容zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001","java".getBytes()); zkClient.getData().forPath("/node1/00001");//获取节点的数据内容，获取到的是 byte数组d.检测节点是否创建成功zkClient.checkExists().forPath("/node1/00001");//不为null的话，说明节点创建成功删除节点a.删除一个子节点zkClient.delete().forPath("/node1/00001");b.删除一个节点以及其下的所有子节点zkClient.delete().deletingChildrenIfNeeded().forPath("/node1");获取/更新节点数据内容zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/node1/00001","java".getBytes()); zkClient.getData().forPath("/node1/00001");//获取节点的数据内容 zkClient.setData().forPath("/node1/00001","c++".getBytes());//更新节点数据内容获取某个节点的所有子节点路径List<String> childrenPaths = zkClient.getChildren().forPath("/node1");监听器下面简单演示一下如何给某个节点注册子节点监听器 。注册了监听器之后，这个节点的子节点发生变化比如增加、减少或者更新的时候，你可以自定义回调操作。String path = "/node1"; PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, path, true); PathChildrenCacheListener pathChildrenCacheListener = (curatorFramework, pathChildrenCacheEvent) -> { // do something }; pathChildrenCache.getListenable().addListener(pathChildrenCacheListener); pathChildrenCache.start();如果你要获取节点事件类型的话，可以通过：pathChildrenCacheEvent.getType()一共有下面几种类型： public static enum Type { CHILD_ADDED,//子节点增加 CHILD_UPDATED,//子节点更新 CHILD_REMOVED,//子节点被删除 CONNECTION_SUSPENDED, CONNECTION_RECONNECTED, CONNECTION_LOST, INITIALIZED; private Type() { } }
AI 摘要 文章系统梳理分布式消息队列基础：定义、生产者-消费者模型，三大核心价值（异步提速、削峰限流、解耦），并指出可用性下降、复杂度上升、一致性风险等代价；对比JMS/AMQP协议差异，RPC与MQ场景定位，最后盘点Kafka、RocketMQ、RabbitMQ、Pulsar等主流产品选型要点。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: 消息队列基础知识总结category: 高性能tag:消息队列::: tip这篇文章中的消息队列主要指的是分布式消息队列。:::“RabbitMQ？”“Kafka？”“RocketMQ？”...在日常学习与开发过程中，我们常常听到消息队列这个关键词。我也在我的多篇文章中提到了这个概念。可能你是熟练使用消息队列的老手，又或者你是不懂消息队列的新手，不论你了不了解消息队列，本文都将带你搞懂消息队列的一些基本理论。如果你是老手，你可能从本文学到你之前不曾注意的一些关于消息队列的重要概念，如果你是新手，相信本文将是你打开消息队列大门的一板砖。什么是消息队列？我们可以把消息队列看作是一个存放消息的容器，当我们需要使用消息的时候，直接从容器中取出消息供自己使用即可。由于队列 Queue 是一种先进先出的数据结构，所以消费消息时也是按照顺序来消费的。参与消息传递的双方称为 生产者 和 消费者 ，生产者负责发送消息，消费者负责处理消息。我们知道操作系统中的进程通信的一种很重要的方式就是消息队列。我们这里提到的消息队列稍微有点区别，更多指的是各个服务以及系统内部各个组件/模块之前的通信，属于一种 中间件 。维基百科是这样介绍中间件的：中间件（英语：Middleware），又译中间件、中介层，是一类提供系统软件和应用软件之间连接、便于软件各部件之间的沟通的软件，应用软件可以借助中间件在不同的技术架构之间共享信息与资源。中间件位于客户机服务器的操作系统之上，管理着计算资源和网络通信。简单来说：中间件就是一类为应用软件服务的软件，应用软件是为用户服务的，用户不会接触或者使用到中间件。除了消息队列之外，常见的中间件还有 RPC 框架、分布式组件、HTTP 服务器、任务调度框架、配置中心、数据库层的分库分表工具和数据迁移工具等等。关于中间件比较详细的介绍可以参考阿里巴巴淘系技术的一篇回答：https://www.zhihu.com/question/19730582/answer/1663627873 。随着分布式和微服务系统的发展，消息队列在系统设计中有了更大的发挥空间，使用消息队列可以降低系统耦合性、实现任务异步、有效地进行流量削峰，是分布式和微服务系统中重要的组件之一。消息队列有什么用？通常来说，使用消息队列能为我们的系统带来下面三点好处：通过异步处理提高系统性能（减少响应所需时间）削峰/限流降低系统耦合性。如果在面试的时候你被面试官问到这个问题的话，一般情况是你在你的简历上涉及到消息队列这方面的内容，这个时候推荐你结合你自己的项目来回答。通过异步处理提高系统性能（减少响应所需时间）将用户的请求数据存储到消息队列之后就立即返回结果。随后，系统再对消息进行消费。因为用户请求数据写入消息队列之后就立即返回给用户了，但是请求数据在后续的业务校验、写数据库等操作中可能失败。因此，使用消息队列进行异步处理之后，需要适当修改业务流程进行配合，比如用户在提交订单之后，订单数据写入消息队列，不能立即返回用户订单提交成功，需要在消息队列的订单消费者进程真正处理完该订单之后，甚至出库后，再通过电子邮件或短信通知用户订单成功，以免交易纠纷。这就类似我们平时手机订火车票和电影票。削峰/限流先将短时间高并发产生的事务消息存储在消息队列中，然后后端服务再慢慢根据自己的能力去消费这些消息，这样就避免直接把后端服务打垮掉。举例：在电子商务一些秒杀、促销活动中，合理使用消息队列可以有效抵御促销活动刚开始大量订单涌入对系统的冲击。如下图所示：降低系统耦合性使用消息队列还可以降低系统耦合性。我们知道如果模块之间不存在直接调用，那么新增模块或者修改模块就对其他模块影响较小，这样系统的可扩展性无疑更好一些。还是直接上图吧：生产者（客户端）发送消息到消息队列中去，接受者（服务端）处理消息，需要消费的系统直接去消息队列取消息进行消费即可而不需要和其他系统有耦合，这显然也提高了系统的扩展性。消息队列使用发布-订阅模式工作，消息发送者（生产者）发布消息，一个或多个消息接受者（消费者）订阅消息。 从上图可以看到消息发送者（生产者）和消息接受者（消费者）之间没有直接耦合，消息发送者将消息发送至分布式消息队列即结束对消息的处理，消息接受者从分布式消息队列获取该消息后进行后续处理，并不需要知道该消息从何而来。对新增业务，只要对该类消息感兴趣，即可订阅该消息，对原有系统和业务没有任何影响，从而实现网站业务的可扩展性设计。例如，我们商城系统分为用户、订单、财务、仓储、消息通知、物流、风控等多个服务。用户在完成下单后，需要调用财务（扣款）、仓储（库存管理）、物流（发货）、消息通知（通知用户发货）、风控（风险评估）等服务。使用消息队列后，下单操作和后续的扣款、发货、通知等操作就解耦了，下单完成发送一个消息到消息队列，需要用到的地方去订阅这个消息进行消息即可。另外，为了避免消息队列服务器宕机造成消息丢失，会将成功发送到消息队列的消息存储在消息生产者服务器上，等消息真正被消费者服务器处理后才删除消息。在消息队列服务器宕机后，生产者服务器会选择分布式消息队列服务器集群中的其他服务器发布消息。备注： 不要认为消息队列只能利用发布-订阅模式工作，只不过在解耦这个特定业务环境下是使用发布-订阅模式的。除了发布-订阅模式，还有点对点订阅模式（一个消息只有一个消费者），我们比较常用的是发布-订阅模式。另外，这两种消息模型是 JMS 提供的，AMQP 协议还提供了另外 5 种消息模型。实现分布式事务我们知道分布式事务的解决方案之一就是 MQ 事务。RocketMQ、 Kafka、Pulsar、QMQ 都提供了事务相关的功能。事务允许事件流应用将消费，处理，生产消息整个过程定义为一个原子操作。详细介绍可以查看 分布式事务详解(付费) 这篇文章。使用消息队列会带来哪些问题？系统可用性降低： 系统可用性在某种程度上降低，为什么这样说呢？在加入 MQ 之前，你不用考虑消息丢失或者说 MQ 挂掉等等的情况，但是，引入 MQ 之后你就需要去考虑了！系统复杂性提高： 加入 MQ 之后，你需要保证消息没有被重复消费、处理消息丢失的情况、保证消息传递的顺序性等等问题！一致性问题： 我上面讲了消息队列可以实现异步，消息队列带来的异步确实可以提高系统响应速度。但是，万一消息的真正消费者并没有正确消费消息怎么办？这样就会导致数据不一致的情况了!JMS 和 AMQPJMS 是什么？JMS（JAVA Message Service,java 消息服务）是 Java 的消息服务，JMS 的客户端之间可以通过 JMS 服务进行异步的消息传输。JMS（JAVA Message Service，Java 消息服务）API 是一个消息服务的标准或者说是规范，允许应用程序组件基于 JavaEE 平台创建、发送、接收和读取消息。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。JMS 定义了五种不同的消息正文格式以及调用的消息类型，允许你发送并接收以一些不同形式的数据：StreamMessage：Java 原始值的数据流MapMessage：一套名称-值对TextMessage：一个字符串对象ObjectMessage：一个序列化的 Java 对象BytesMessage：一个字节的数据流ActiveMQ（已被淘汰） 就是基于 JMS 规范实现的。JMS 两种消息模型点到点（P2P）模型使用队列（Queue）作为消息通信载体；满足生产者与消费者模式，一条消息只能被一个消费者使用，未被消费的消息在队列中保留直到被消费或超时。比如：我们生产者发送 100 条消息的话，两个消费者来消费一般情况下两个消费者会按照消息发送的顺序各自消费一半（也就是你一个我一个的消费。）发布/订阅（Pub/Sub）模型发布订阅模型（Pub/Sub） 使用主题（Topic）作为消息通信载体，类似于广播模式；发布者发布一条消息，该消息通过主题传递给所有的订阅者。AMQP 是什么？AMQP，即 Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准 高级消息队列协议（二进制应用层协议），是应用层协议的一个开放标准，为面向消息的中间件设计，兼容 JMS。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件同产品，不同的开发语言等条件的限制。RabbitMQ 就是基于 AMQP 协议实现的。JMS vs AMQP对比方向JMSAMQP定义Java API协议跨语言否是跨平台否是支持消息类型提供两种消息模型：①Peer-2-Peer;②Pub/sub提供了五种消息模型：①direct exchange；②fanout exchange；③topic change；④headers exchange；⑤system exchange。本质来讲，后四种和 JMS 的 pub/sub 模型没有太大差别，仅是在路由机制上做了更详细的划分；支持消息类型支持多种消息类型 ，我们在上面提到过byte[]（二进制）总结：AMQP 为消息定义了线路层（wire-level protocol）的协议，而 JMS 所定义的是 API 规范。在 Java 体系中，多个 client 均可以通过 JMS 进行交互，不需要应用修改代码，但是其对跨平台的支持较差。而 AMQP 天然具有跨平台、跨语言特性。JMS 支持 TextMessage、MapMessage 等复杂的消息类型；而 AMQP 仅支持 byte[] 消息类型（复杂的类型可序列化后发送）。由于 Exchange 提供的路由算法，AMQP 可以提供多样化的路由方式来传递消息到消息队列，而 JMS 仅支持 队列 和 主题/订阅 方式两种。RPC 和消息队列的区别RPC 和消息队列都是分布式微服务系统中重要的组件之一，下面我们来简单对比一下两者：从用途来看：RPC 主要用来解决两个服务的远程通信问题，不需要了解底层网络的通信机制。通过 RPC 可以帮助我们调用远程计算机上某个服务的方法，这个过程就像调用本地方法一样简单。消息队列主要用来降低系统耦合性、实现任务异步、有效地进行流量削峰。从通信方式来看：RPC 是双向直接网络通讯，消息队列是单向引入中间载体的网络通讯。从架构上来看：消息队列需要把消息存储起来，RPC 则没有这个要求，因为前面也说了 RPC 是双向直接网络通讯。从请求处理的时效性来看：通过 RPC 发出的调用一般会立即被处理，存放在消息队列中的消息并不一定会立即被处理。RPC 和消息队列本质上是网络通讯的两种不同的实现机制，两者的用途不同，万不可将两者混为一谈。分布式消息队列技术选型常见的消息队列有哪些？KafkaKafka 是 LinkedIn 开源的一个分布式流式处理平台，已经成为 Apache 顶级项目，早期被用来用于处理海量的日志，后面才慢慢发展成了一款功能全面的高性能消息队列。流式处理平台具有三个关键功能：消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是 Kafka 也被归类为消息队列的原因。容错的持久方式存储记录消息流：Kafka 会把消息持久化到磁盘，有效避免了消息丢失的风险。流式处理平台： 在消息发布的时候进行处理，Kafka 提供了一个完整的流式处理类库。Kafka 是一个分布式系统，由通过高性能 TCP 网络协议进行通信的服务器和客户端组成，可以部署在在本地和云环境中的裸机硬件、虚拟机和容器上。在 Kafka 2.8 之前，Kafka 最被大家诟病的就是其重度依赖于 Zookeeper 做元数据管理和集群的高可用。在 Kafka 2.8 之后，引入了基于 Raft 协议的 KRaft 模式，不再依赖 Zookeeper，大大简化了 Kafka 的架构，让你可以以一种轻量级的方式来使用 Kafka。不过，要提示一下：如果要使用 KRaft 模式的话，建议选择较高版本的 Kafka，因为这个功能还在持续完善优化中。Kafka 3.3.1 版本是第一个将 KRaft（Kafka Raft）共识协议标记为生产就绪的版本。Kafka 官网：http://kafka.apache.org/Kafka 更新记录（可以直观看到项目是否还在维护）：https://kafka.apache.org/downloadsRocketMQRocketMQ 是阿里开源的一款云原生“消息、事件、流”实时数据处理平台，借鉴了 Kafka，已经成为 Apache 顶级项目。RocketMQ 的核心特性（摘自 RocketMQ 官网）：云原生：生与云，长与云，无限弹性扩缩，K8s 友好高吞吐：万亿级吞吐保证，同时满足微服务与大数据场景。流处理：提供轻量、高扩展、高性能和丰富功能的流计算引擎。金融级：金融级的稳定性，广泛用于交易核心链路。架构极简：零外部依赖，Shared-nothing 架构。生态友好：无缝对接微服务、实时计算、数据湖等周边生态。根据官网介绍：Apache RocketMQ 自诞生以来，因其架构简单、业务功能丰富、具备极强可扩展性等特点被众多企业开发者以及云厂商广泛采用。历经十余年的大规模场景打磨，RocketMQ 已经成为业内共识的金融级可靠业务消息首选方案，被广泛应用于互联网、大数据、移动互联网、物联网等领域的业务场景。RocketMQ 官网：https://rocketmq.apache.org/ （文档很详细，推荐阅读）RocketMQ 更新记录（可以直观看到项目是否还在维护）：https://github.com/apache/rocketmq/releasesRabbitMQRabbitMQ 是采用 Erlang 语言实现 AMQP(Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，它最初起源于金融系统，用于在分布式系统中存储转发消息。RabbitMQ 发展到今天，被越来越多的人认可，这和它在易用性、扩展性、可靠性和高可用性等方面的卓著表现是分不开的。RabbitMQ 的具体特点可以概括为以下几点：可靠性： RabbitMQ 使用一些机制来保证消息的可靠性，如持久化、传输确认及发布确认等。灵活的路由： 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能，RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个交换器绑定在一起，也可以通过插件机制来实现自己的交换器。这个后面会在我们讲 RabbitMQ 核心概念的时候详细介绍到。扩展性： 多个 RabbitMQ 节点可以组成一个集群，也可以根据实际业务情况动态地扩展集群中节点。高可用性： 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队列仍然可用。支持多种协议： RabbitMQ 除了原生支持 AMQP 协议，还支持 STOMP、MQTT 等多种消息中间件协议。多语言客户端： RabbitMQ 几乎支持所有常用语言，比如 Java、Python、Ruby、PHP、C#、JavaScript 等。易用的管理界面： RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集群中的节点等。在安装 RabbitMQ 的时候会介绍到，安装好 RabbitMQ 就自带管理界面。插件机制： RabbitMQ 提供了许多插件，以实现从多方面进行扩展，当然也可以编写自己的插件。感觉这个有点类似 Dubbo 的 SPI 机制RabbitMQ 官网：https://www.rabbitmq.com/ 。RabbitMQ 更新记录（可以直观看到项目是否还在维护）：https://www.rabbitmq.com/news.htmlPulsarPulsar 是下一代云原生分布式消息流平台，最初由 Yahoo 开发 ，已经成为 Apache 顶级项目。Pulsar 集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性，被看作是云原生时代实时消息流传输、存储和计算最佳解决方案。Pulsar 的关键特性如下（摘自官网）：是下一代云原生分布式消息流平台。Pulsar 的单个实例原生支持多个集群，可跨机房在集群间无缝地完成消息复制。极低的发布延迟和端到端延迟。可无缝扩展到超过一百万个 topic。简单的客户端 API，支持 Java、Go、Python 和 C++。主题的多种订阅模式（独占、共享和故障转移）。通过 Apache BookKeeper 提供的持久化消息存储机制保证消息传递 。由轻量级的 serverless 计算框架 Pulsar Functions 实现流原生的数据处理。基于 Pulsar Functions 的 serverless connector 框架 Pulsar IO 使得数据更易移入、移出 Apache Pulsar。分层式存储可在数据陈旧时，将数据从热存储卸载到冷/长期存储（如 S3、GCS）中。Pulsar 官网：https://pulsar.apache.org/Pulsar 更新记录（可以直观看到项目是否还在维护）：https://github.com/apache/pulsar/releasesActiveMQ目前已经被淘汰，不推荐使用，不建议学习。如何选择？参考《Java 工程师面试突击第 1 季-中华石杉老师》对比方向概要吞吐量万级的 ActiveMQ 和 RabbitMQ 的吞吐量（ActiveMQ 的性能最差）要比十万级甚至是百万级的 RocketMQ 和 Kafka 低一个数量级。可用性都可以实现高可用。ActiveMQ 和 RabbitMQ 都是基于主从架构实现高可用性。RocketMQ 基于分布式架构。 Kafka 也是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用时效性RabbitMQ 基于 Erlang 开发，所以并发能力很强，性能极其好，延时很低，达到微秒级，其他几个都是 ms 级。功能支持Pulsar 的功能更全面，支持多租户、多种消费模式和持久性模式等功能，是下一代云原生分布式消息流平台。消息丢失ActiveMQ 和 RabbitMQ 丢失的可能性非常低， Kafka、RocketMQ 和 Pulsar 理论上可以做到 0 丢失。总结：ActiveMQ 的社区算是比较成熟，但是较目前来说，ActiveMQ 的性能比较差，而且版本迭代很慢，不推荐使用，已经被淘汰了。RabbitMQ 在吞吐量方面虽然稍逊于 Kafka、RocketMQ 和 Pulsar，但是由于它基于 Erlang 开发，所以并发能力很强，性能极其好，延时很低，达到微秒级。但是也因为 RabbitMQ 基于 Erlang 开发，所以国内很少有公司有实力做 Erlang 源码级别的研究和定制。如果业务场景对并发量要求不是太高（十万级、百万级），那这几种消息队列中，RabbitMQ 或许是你的首选。RocketMQ 和 Pulsar 支持强一致性，对消息一致性要求比较高的场景可以使用。RocketMQ 阿里出品，Java 系开源项目，源代码我们可以直接阅读，然后可以定制自己公司的 MQ，并且 RocketMQ 有阿里巴巴的实际业务场景的实战考验。Kafka 的特点其实很明显，就是仅仅提供较少的核心功能，但是提供超高的吞吐量，ms 级的延迟，极高的可用性以及可靠性，而且分布式可以任意扩展。同时 Kafka 最好是支撑较少的 topic 数量即可，保证其超高吞吐量。Kafka 唯一的一点劣势是有可能消息重复消费，那么对数据准确性会造成极其轻微的影响，在大数据领域中以及日志采集中，这点轻微影响可以忽略这个特性天然适合大数据实时计算以及日志收集。如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。参考《大型网站技术架构 》KRaft: Apache Kafka Without ZooKeeper：https://developer.confluent.io/learn/kraft/消息队列的使用场景是什么样的？：https://mp.weixin.qq.com/s/4V1jI6RylJr7Jr9JsQe73A
AI 摘要 文章通过短信发送示例，对比演示静态代理、JDK动态代理与CGLIB动态代理的编码步骤、优缺点及适用场景：静态代理编译期硬编码，维护困难；JDK动态代理基于接口生成运行时代理类；CGLIB用继承生成子类，可代理无接口类；总结三者灵活性、性能与框架应用差异。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结07 静态代理+JDK/CGLIB 动态代理实战代理模式代理模式是一种比较好的理解的设计模式。简单来说就是 我们使用代理对象来代替对真实对象(real object)的访问，这样就可以在不修改原目标对象的前提下，提供额外的功能操作，扩展目标对象的功能。代理模式的主要作用是扩展目标对象的功能，比如说在目标对象的某个方法执行前后你可以增加一些自定义的操作。举个例子：你找了小红来帮你问话，小红就可以看作是代理你的代理对象，代理的行为（方法）是问话。https://medium.com/@mithunsasidharan/understanding-the-proxy-design-pattern-5e63fe38052a代理模式有静态代理和动态代理两种实现方式，我们 先来看一下静态代理模式的实现。静态代理静态代理中，我们对目标对象的每个方法的增强都是手动完成的（_后面会具体演示代码_），非常不灵活（_比如接口一旦新增加方法，目标对象和代理对象都要进行修改_）且麻烦(_需要对每个目标类都单独写一个代理类_)。 实际应用场景非常非常少，日常开发几乎看不到使用静态代理的场景。上面我们是从实现和应用角度来说的静态代理，从 JVM 层面来说， 静态代理在编译时就将接口、实现类、代理类这些都变成了一个个实际的 class 文件。静态代理实现步骤:定义一个接口及其实现类；创建一个代理类同样实现这个接口将目标对象注入进代理类，然后在代理类的对应方法调用目标类中的对应方法。这样的话，我们就可以通过代理类屏蔽对目标对象的访问，并且可以在目标方法执行前后做一些自己想做的事情。下面通过代码展示！1.定义发送短信的接口public interface SmsService { String send(String message); }2.实现发送短信的接口public class SmsServiceImpl implements SmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }3.创建代理类并同样实现发送短信的接口public class SmsProxy implements SmsService { private final SmsService smsService; public SmsProxy(SmsService smsService) { this.smsService = smsService; } @Override public String send(String message) { //调用方法之前，我们可以添加自己的操作 System.out.println("before method send()"); smsService.send(message); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method send()"); return null; } }4.实际使用public class Main { public static void main(String[] args) { SmsService smsService = new SmsServiceImpl(); SmsProxy smsProxy = new SmsProxy(smsService); smsProxy.send("java"); } }运行上述代码之后，控制台打印出：before method send() send message:java after method send()可以输出结果看出，我们已经增强了 SmsServiceImpl 的send()方法。动态代理相比于静态代理来说，动态代理更加灵活。我们不需要针对每个目标类都单独创建一个代理类，并且也不需要我们必须实现接口，我们可以直接代理实现类( _CGLIB 动态代理机制_)。从 JVM 角度来说，动态代理是在运行时动态生成类字节码，并加载到 JVM 中的。说到动态代理，Spring AOP、RPC 框架应该是两个不得不的提的，它们的实现都依赖了动态代理。动态代理在我们日常开发中使用的相对较小，但是在框架中的几乎是必用的一门技术。学会了动态代理之后，对于我们理解和学习各种框架的原理也非常有帮助。就 Java 来说，动态代理的实现方式有很多种，比如 JDK 动态代理、CGLIB 动态代理等等。guide-rpc-framework 使用的是 JDK 动态代理，我们先来看看 JDK 动态代理的使用。另外，虽然 guide-rpc-framework 没有用到 CGLIB 动态代理 ，我们这里还是简单介绍一下其使用以及和JDK 动态代理的对比。JDK 动态代理机制介绍在 Java 动态代理机制中 **InvocationHandler** 接口和 **Proxy** 类是核心。Proxy 类中使用频率最高的方法是：newProxyInstance() ，这个方法主要用来生成一个代理对象。 public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, InvocationHandler h) throws IllegalArgumentException { ...... }这个方法一共有 3 个参数：loader :类加载器，用于加载代理对象。interfaces : 被代理类实现的一些接口；h : 实现了 InvocationHandler 接口的对象；要实现动态代理的话，还必须需要实现InvocationHandler 来自定义处理逻辑。 当我们的动态代理对象调用一个方法时候，这个方法的调用就会被转发到实现InvocationHandler 接口类的 invoke 方法来调用。public interface InvocationHandler { /** * 当你使用代理对象调用方法的时候实际会调用到这个方法 */ public Object invoke(Object proxy, Method method, Object[] args) throws Throwable; }invoke() 方法有下面三个参数：proxy :动态生成的代理类method : 与代理类对象调用的方法相对应args : 当前 method 方法的参数也就是说：你通过**Proxy** 类的 **newProxyInstance()** 创建的代理对象在调用方法的时候，实际会调用到实现**InvocationHandler** 接口的类的 **invoke()**方法。 你可以在 invoke() 方法中自定义处理逻辑，比如在方法执行前后做什么事情。JDK 动态代理类使用步骤定义一个接口及其实现类；自定义 InvocationHandler 并重写invoke方法，在 invoke 方法中我们会调用原生方法（被代理类的方法）并自定义一些处理逻辑；通过 Proxy.newProxyInstance(ClassLoader loader,Class<?>[] interfaces,InvocationHandler h) 方法创建代理对象；代码示例这样说可能会有点空洞和难以理解，我上个例子，大家感受一下吧！1.定义发送短信的接口public interface SmsService { String send(String message); }2.实现发送短信的接口public class SmsServiceImpl implements SmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }3.定义一个 JDK 动态代理类import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; /** * @author shuang.kou * @createTime 2020年05月11日 11:23:00 */ public class DebugInvocationHandler implements InvocationHandler { /** * 代理类中的真实对象 */ private final Object target; public DebugInvocationHandler(Object target) { this.target = target; } public Object invoke(Object proxy, Method method, Object[] args) throws InvocationTargetException, IllegalAccessException { //调用方法之前，我们可以添加自己的操作 System.out.println("before method " + method.getName()); Object result = method.invoke(target, args); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method " + method.getName()); return result; } }invoke() 方法: 当我们的动态代理对象调用原生方法的时候，最终实际上调用到的是 invoke() 方法，然后 invoke() 方法代替我们去调用了被代理对象的原生方法。4.获取代理对象的工厂类public class JdkProxyFactory { public static Object getProxy(Object target) { return Proxy.newProxyInstance( target.getClass().getClassLoader(), // 目标类的类加载 target.getClass().getInterfaces(), // 代理需要实现的接口，可指定多个 new DebugInvocationHandler(target) // 代理对象对应的自定义 InvocationHandler ); } }getProxy() ：主要通过Proxy.newProxyInstance（）方法获取某个类的代理对象5.实际使用SmsService smsService = (SmsService) JdkProxyFactory.getProxy(new SmsServiceImpl()); smsService.send("java");运行上述代码之后，控制台打印出：before method send send message:java after method sendCGLIB 动态代理机制介绍JDK 动态代理有一个最致命的问题是其只能代理实现了接口的类。为了解决这个问题，我们可以用 CGLIB 动态代理机制来避免。CGLIB(_Code Generation Library_)是一个基于ASM的字节码生成库，它允许我们在运行时对字节码进行修改和动态生成。CGLIB 通过继承方式实现代理。很多知名的开源框架都使用到了CGLIB， 例如 Spring 中的 AOP 模块中：如果目标对象实现了接口，则默认采用 JDK 动态代理，否则采用 CGLIB 动态代理。在 CGLIB 动态代理机制中 **MethodInterceptor** 接口和 **Enhancer** 类是核心。你需要自定义 MethodInterceptor 并重写 intercept 方法，intercept 用于拦截增强被代理类的方法。public interface MethodInterceptor extends Callback{ // 拦截被代理类中的方法 public Object intercept(Object obj, java.lang.reflect.Method method, Object[] args, MethodProxy proxy) throws Throwable; }obj :被代理的对象（需要增强的对象）method :被拦截的方法（需要增强的方法）args :方法入参methodProxy :用于调用原始方法你可以通过 Enhancer类来动态获取被代理类，当代理类调用方法的时候，实际调用的是 MethodInterceptor 中的 intercept 方法。CGLIB 动态代理类使用步骤定义一个类；自定义 MethodInterceptor 并重写 intercept 方法，intercept 用于拦截增强被代理类的方法，和 JDK 动态代理中的 invoke 方法类似；通过 Enhancer 类的 create()创建代理类；代码示例不同于 JDK 动态代理不需要额外的依赖。CGLIB(_Code Generation Library_) 实际是属于一个开源项目，如果你要使用它的话，需要手动添加相关依赖。<dependency> <groupId>cglib</groupId> <artifactId>cglib</artifactId> <version>3.3.0</version> </dependency>1.实现一个使用阿里云发送短信的类package github.javaguide.dynamicProxy.cglibDynamicProxy; public class AliSmsService { public String send(String message) { System.out.println("send message:" + message); return message; } }2.自定义 **MethodInterceptor**（方法拦截器）import net.sf.cglib.proxy.MethodInterceptor; import net.sf.cglib.proxy.MethodProxy; import java.lang.reflect.Method; /** * 自定义MethodInterceptor */ public class DebugMethodInterceptor implements MethodInterceptor { /** * @param o 代理对象（增强的对象） * @param method 被拦截的方法（需要增强的方法） * @param args 方法入参 * @param methodProxy 用于调用原始方法 */ @Override public Object intercept(Object o, Method method, Object[] args, MethodProxy methodProxy) throws Throwable { //调用方法之前，我们可以添加自己的操作 System.out.println("before method " + method.getName()); Object object = methodProxy.invokeSuper(o, args); //调用方法之后，我们同样可以添加自己的操作 System.out.println("after method " + method.getName()); return object; } }3.获取代理类import net.sf.cglib.proxy.Enhancer; public class CglibProxyFactory { public static Object getProxy(Class<?> clazz) { // 创建动态代理增强类 Enhancer enhancer = new Enhancer(); // 设置类加载器 enhancer.setClassLoader(clazz.getClassLoader()); // 设置被代理类 enhancer.setSuperclass(clazz); // 设置方法拦截器 enhancer.setCallback(new DebugMethodInterceptor()); // 创建代理类 return enhancer.create(); } }4.实际使用AliSmsService aliSmsService = (AliSmsService) CglibProxyFactory.getProxy(AliSmsService.class); aliSmsService.send("java");运行上述代码之后，控制台打印出：before method send send message:java after method sendJDK 动态代理和 CGLIB 动态代理对比JDK 动态代理只能只能代理实现了接口的类，而 CGLIB 可以代理未实现任何接口的类。 另外， CGLIB 动态代理是通过生成一个被代理类的子类来拦截被代理类的方法调用，因此不能代理声明为 final 类型的类和方法。就二者的效率来说，大部分情况都是 JDK 动态代理更优秀，随着 JDK 版本的升级，这个优势更加明显。静态代理和动态代理的对比灵活性 ：动态代理更加灵活，不需要必须实现接口，可以直接代理实现类，并且可以不需要针对每个目标类都创建一个代理类。另外，静态代理中，接口一旦新增加方法，目标对象和代理对象都要进行修改，这是非常麻烦的！JVM 层面 ：静态代理在编译时就将接口、实现类、代理类这些都变成了一个个实际的 class 文件。而动态代理是在运行时动态生成类字节码，并加载到 JVM 中的。总结这篇文章中主要介绍了代理模式的两种实现：静态代理以及动态代理。涵盖了静态代理和动态代理实战、静态代理和动态代理的区别、JDK 动态代理和 Cglib 动态代理区别等内容。文中涉及到的所有源码，你可以在这里找到：https://github.com/Snailclimb/guide-rpc-framework-learning/tree/master/src/main/java/github/javaguide/proxy 。