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 摘要 文章剖析自研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 摘要 文章剖析了自研 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 摘要 文章演示了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 摘要 文章通过短信发送示例，对比演示静态代理、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 。
AI 摘要 文章介绍Netty高性能NIO框架特点，演示用Kryo序列化实现RPC通信：定义RpcRequest/RpcResponse实体，编写Netty客户端、服务端及编解码器，完成一次请求-响应测试。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结06 Netty从入门到网络通信实战Netty 介绍简单用 3 点概括一下 Netty 吧！Netty 是一个基于 NIO 的 client-server(客户端服务器)框架，使用它可以快速简单地开发网络应用程序。它极大地简化并简化了 TCP 和 UDP 套接字服务器等网络编程,并且性能以及安全性等很多方面甚至都要更好。支持多种协议如 FTP，SMTP，HTTP 以及各种二进制和基于文本的传统协议。用官方的总结就是：Netty 成功地找到了一种在不妥协可维护性和性能的情况下实现易于开发，性能，稳定性和灵活性的方法。Netty 特点根据官网的描述，我们可以总结出下面一些特点：统一的 API，支持多种传输类型，阻塞和非阻塞的。简单而强大的线程模型。自带编解码器解决 TCP 粘包/拆包问题。自带各种协议栈。真正的无连接数据包套接字支持。比直接使用 Java 核心 API 有更高的吞吐量、更低的延迟、更低的资源消耗和更少的内存复制。安全性不错，有完整的 SSL/TLS 以及 StartTLS 支持。社区活跃成熟稳定，经历了大型项目的使用和考验，而且很多开源项目都使用到了 Netty 比如我们经常接触的 Dubbo、RocketMQ 等等。......使用 Netty 能做什么？这个应该是老铁们最关心的一个问题了，凭借自己的了解，简单说一下，理论上 NIO 可以做的事情 ，使用 Netty 都可以做并且更好。Netty 主要用来做网络通信 :作为 RPC 框架的网络通信工具 ： 我们在分布式系统中，不同服务节点之间经常需要相互调用，这个时候就需要 RPC 框架了。不同服务节点的通信是如何做的呢？可以使用 Netty 来做。比如我调用另外一个节点的方法的话，至少是要让对方知道我调用的是哪个类中的哪个方法以及相关参数吧！实现一个自己的 HTTP 服务器 ：通过 Netty 我们可以自己实现一个简单的 HTTP 服务器，这个大家应该不陌生。说到 HTTP 服务器的话，作为 Java 后端开发，我们一般使用 Tomcat 比较多。一个最基本的 HTTP 服务器可要以处理常见的 HTTP Method 的请求，比如 POST 请求、GET 请求等等。实现一个即时通讯系统 ： 使用 Netty 我们可以实现一个可以聊天类似微信的即时通讯系统，这方面的开源项目还蛮多的，可以自行去 Github 找一找。消息推送系统 ：市面上有很多消息推送系统都是基于 Netty 来做的。......哪些开源项目用到了 Netty？我们平常经常接触的 Dubbo、RocketMQ、Elasticsearch、gRPC 等等都用到了 Netty。可以说大量的开源项目都用到了 Netty，所以掌握 Netty 有助于你更好的使用这些开源项目并且让你有能力对其进行二次开发。实际上还有很多很多优秀的项目用到了 Netty,Netty 官方也做了统计，统计结果在这里：https://netty.io/wiki/related-projects.html 。Netty 使用 kryo 序列化传输对象实战注意 ：Kryo不支持没有无参构造函数的对象进行反序列化，因此如果某个对象希望使用Kryo来进行序列化操作的话，需要有相应的无参构造函数才可以。传输实体类我们首先定义两个对象，这两个对象是客户端与服务端进行交互的实体类。 客户端将 RpcRequest 类型的对象发送到服务端，服务端进行相应的处理之后将得到结果 RpcResponse 对象返回给客户端。客户端请求RpcRequest.java :客户端请求实体类@AllArgsConstructor @Getter @NoArgsConstructor @Builder @ToString public class RpcRequest { private String interfaceName; private String methodName; }服务端响应RpcResponse.java ：服务端响应实体类@AllArgsConstructor @Getter @NoArgsConstructor @Builder @ToString public class RpcResponse { private String message; }客户端初始化客户端客户端中主要有一个用于向服务端发送消息的 **sendMessage()**方法，通过这个方法你可以将消息也就是**RpcRequest** 对象发送到服务端，并且你可以同步获取到服务端返回的结果也就是**RpcResponse** 对象。 /** * @author shuang.kou * @createTime 2020年05月13日 20:48:00 */ public class NettyClient { private static final Logger logger = LoggerFactory.getLogger(NettyClient.class); private final String host; private final int port; private static final Bootstrap b; public NettyClient(String host, int port) { this.host = host; this.port = port; } // 初始化相关资源比如 EventLoopGroup, Bootstrap static { EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); b = new Bootstrap(); KryoSerializer kryoSerializer = new KryoSerializer(); b.group(eventLoopGroup) .channel(NioSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) // 连接的超时时间，超过这个时间还是建立不上的话则代表连接失败 // 如果 15 秒之内没有发送数据给服务端的话，就发送一次心跳请求 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { /* 自定义序列化编解码器 */ // RpcResponse -> ByteBuf ch.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcResponse.class)); // ByteBuf -> RpcRequest ch.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcRequest.class)); ch.pipeline().addLast(new NettyClientHandler()); } }); } /** * 发送消息到服务端 * * @param rpcRequest 消息体 * @return 服务端返回的数据 */ public RpcResponse sendMessage(RpcRequest rpcRequest) { try { ChannelFuture f = b.connect(host, port).sync(); logger.info("client connect {}", host + ":" + port); Channel futureChannel = f.channel(); logger.info("send message"); if (futureChannel != null) { futureChannel.writeAndFlush(rpcRequest).addListener(future -> { if (future.isSuccess()) { logger.info("client send message: [{}]", rpcRequest.toString()); } else { logger.error("Send failed:", future.cause()); } }); //阻塞等待 ，直到Channel关闭 futureChannel.closeFuture().sync(); // 将服务端返回的数据也就是RpcResponse对象取出 AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse"); return futureChannel.attr(key).get(); } } catch (InterruptedException e) { logger.error("occur exception when connect server:", e); } return null; } }sendMessage()方法分析：首先初始化了一个 Bootstrap通过 Bootstrap 对象连接服务端通过 Channel 向服务端发送消息RpcRequest发送成功后，阻塞等待 ，直到Channel关闭拿到服务端返回的结果 RpcResponse自定义 ChannelHandler 处理服务端消息public class NettyClientHandler extends ChannelInboundHandlerAdapter { private static final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class); @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { RpcResponse rpcResponse = (RpcResponse) msg; logger.info("client receive msg: [{}]", rpcResponse.toString()); // 声明一个 AttributeKey 对象 AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse"); // 将服务端的返回结果保存到 AttributeMap 上，AttributeMap 可以看作是一个Channel的共享数据源 // AttributeMap的key是AttributeKey，value是Attribute ctx.channel().attr(key).set(rpcResponse); ctx.channel().close(); } finally { ReferenceCountUtil.release(msg); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { logger.error("client caught exception", cause); ctx.close(); } }NettyClientHandler用于读取服务端发送过来的 RpcResponse 消息对象，并将 RpcResponse 消息对象保存到 AttributeMap 上，AttributeMap 可以看作是一个Channel的共享数据源。这样的话，我们就能通过 channel 和 key 将数据读取出来。AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse"); return futureChannel.attr(key).get();这个额外提一下 AttributeMap ,AttributeMap 是一个接口，但是类似于 Map 数据结构 。public interface AttributeMap { <T> Attribute<T> attr(AttributeKey<T> key); <T> boolean hasAttr(AttributeKey<T> key); }Channel 实现了 AttributeMap 接口，这样也就表明它存在了AttributeMap 相关的属性。 每个 Channel上的AttributeMap属于共享数据。AttributeMap 的结构，和Map很像，我们可以把 key 看作是AttributeKey，value 看作是Attribute，我们可以根据 AttributeKey找到对应的Attribute。public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparable<Channel> { ...... }服务端初始化服务端**NettyServer** 主要作用就是开启了一个服务端用于接受客户端的请求并处理。public class NettyServer { private static final Logger logger = LoggerFactory.getLogger(NettyServer.class); private final int port; private NettyServer(int port) { this.port = port; } private void run() { EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); KryoSerializer kryoSerializer = new KryoSerializer(); 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) { ch.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcRequest.class)); ch.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcResponse.class)); ch.pipeline().addLast(new NettyServerHandler()); } }); // 绑定端口，同步等待绑定成功 ChannelFuture f = b.bind(port).sync(); // 等待服务端监听端口关闭 f.channel().closeFuture().sync(); } catch (InterruptedException e) { logger.error("occur exception when start server:", e); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }自定义 ChannelHandler 处理客户端消息NettyServerHandler 用于接收客户端发送过来的消息并返回结果给客户端。public class NettyServerHandler extends ChannelInboundHandlerAdapter { private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class); private static final AtomicInteger atomicInteger = new AtomicInteger(1); @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { try { RpcRequest rpcRequest = (RpcRequest) msg; logger.info("server receive msg: [{}] ,times:[{}]", rpcRequest, atomicInteger.getAndIncrement()); RpcResponse messageFromServer = RpcResponse.builder().message("message from server").build(); ChannelFuture f = ctx.writeAndFlush(messageFromServer); f.addListener(ChannelFutureListener.CLOSE); } finally { ReferenceCountUtil.release(msg); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { logger.error("server catch exception",cause); ctx.close(); } }编码器自定义编码器**NettyKryoEncoder** 是我们自定义的编码器。它负责处理"出站"消息，将消息格式转换为字节数组然后写入到字节数据的容器 **ByteBuf** 对象中。/** * 自定义编码器。 * * 网络传输需要通过字节流来实现，ByteBuf 可以看作是 Netty 提供的字节数据的容器，使用它会让我们更加方便地处理字节数据。 * * @author shuang.kou * @createTime 2020年05月25日 19:43:00 */ @AllArgsConstructor public class NettyKryoEncoder extends MessageToByteEncoder<Object> { private final Serializer serializer; private final Class<?> genericClass; /** * 将对象转换为字节码然后写入到 ByteBuf 对象中 */ @Override protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) { if (genericClass.isInstance(o)) { // 1. 将对象转换为byte byte[] body = serializer.serialize(o); // 2. 读取消息的长度 int dataLength = body.length; // 3.写入消息对应的字节数组长度,writerIndex 加 4 byteBuf.writeInt(dataLength); //4.将字节数组写入 ByteBuf 对象中 byteBuf.writeBytes(body); } } }自定义解码器**NettyKryoDecoder**是我们自定义的解码器。它负责处理"入站"消息，它会从 **ByteBuf** 中读取到业务对象对应的字节序列，然后再将字节序列转换为我们的业务对象。/** * 自定义解码器。 * * @author shuang.kou * @createTime 2020年05月25日 19:42:00 */ @AllArgsConstructor @Slf4j public class NettyKryoDecoder extends ByteToMessageDecoder { private final Serializer serializer; private final Class<?> genericClass; /** * Netty传输的消息长度也就是对象序列化后对应的字节数组的大小，存储在 ByteBuf 头部 */ private static final int BODY_LENGTH = 4; /** * 解码 ByteBuf 对象 * * @param ctx 解码器关联的 ChannelHandlerContext 对象 * @param in "入站"数据，也就是 ByteBuf 对象 * @param out 解码之后的数据对象需要添加到 out 对象里面 */ @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { //1.byteBuf中写入的消息长度所占的字节数已经是4了，所以 byteBuf 的可读字节必须大于 4， if (in.readableBytes() >= BODY_LENGTH) { //2.标记当前readIndex的位置，以便后面重置readIndex 的时候使用 in.markReaderIndex(); //3.读取消息的长度 //注意： 消息长度是encode的时候我们自己写入的，参见 NettyKryoEncoder 的encode方法 int dataLength = in.readInt(); //4.遇到不合理的情况直接 return if (dataLength < 0 || in.readableBytes() < 0) { log.error("data length or byteBuf readableBytes is not valid"); return; } //5.如果可读字节数小于消息长度的话，说明是不完整的消息，重置readIndex if (in.readableBytes() < dataLength) { in.resetReaderIndex(); return; } // 6.走到这里说明没什么问题了，可以序列化了 byte[] body = new byte[dataLength]; in.readBytes(body); // 将bytes数组转换为我们需要的对象 Object obj = serializer.deserialize(body, genericClass); out.add(obj); log.info("successful decode ByteBuf to Object"); } } }自定义序列化接口Serializer 接口主要有两个方法一个用于序列化，一个用户反序列化。public interface Serializer { /** * 序列化 * * @param obj 要序列化的对象 * @return 字节数组 */ byte[] serialize(Object obj); /** * 反序列化 * * @param bytes 序列化后的字节数组 * @param clazz 类 * @param <T> * @return 反序列化的对象 */ <T> T deserialize(byte[] bytes, Class<T> clazz); }实现序列化接口下面是我自定义 kryo 序列化实现类。public class KryoSerializer implements Serializer { /** * 由于 Kryo 不是线程安全的。每个线程都应该有自己的 Kryo，Input 和 Output 实例。 * 所以，使用 ThreadLocal 存放 Kryo 对象 */ private static final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> { Kryo kryo = new Kryo(); kryo.register(RpcResponse.class); kryo.register(RpcRequest.class); kryo.setReferences(true);//默认值为true,是否关闭注册行为,关闭之后可能存在序列化问题，一般推荐设置为 true kryo.setRegistrationRequired(false);//默认值为false,是否关闭循环引用，可以提高性能，但是一般不推荐设置为 true return kryo; }); @Override public byte[] serialize(Object obj) { try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); Output output = new Output(byteArrayOutputStream)) { Kryo kryo = kryoThreadLocal.get(); // Object->byte:将对象序列化为byte数组 kryo.writeObject(output, obj); kryoThreadLocal.remove(); return output.toBytes(); } catch (Exception e) { throw new SerializeException("序列化失败"); } } @Override public <T> T deserialize(byte[] bytes, Class<T> clazz) { try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes); Input input = new Input(byteArrayInputStream)) { Kryo kryo = kryoThreadLocal.get(); // byte->Object:从byte数组中反序列化出对对象 Object o = kryo.readObject(input, clazz); kryoThreadLocal.remove(); return clazz.cast(o); } catch (Exception e) { throw new SerializeException("反序列化失败"); } } }自定义序列化异常类 SerializeException 如下：public class SerializeException extends RuntimeException { public SerializeException(String message) { super(message); } }测试效果启动服务端：new NettyServer(8889).run();启动客户端并发送 4 次消息给服务端：RpcRequest rpcRequest = RpcRequest.builder() .interfaceName("interface") .methodName("hello").build(); NettyClient nettyClient = new NettyClient("127.0.0.1", 8889); for (int i = 0; i < 3; i++) { nettyClient.sendMessage(rpcRequest); } RpcResponse rpcResponse = nettyClient.sendMessage(rpcRequest); System.out.println(rpcResponse.toString());客户端控制台输出：服务端控制台输出：大功告成！源码地址：https://github.com/Snailclimb/guide-rpc-framework-learning/tree/master/src/main/java/github/javaguide/netty/kyro 。
AI 摘要 文章通过实战演示 Java Socket 通信：服务端用 ServerSocket 监听，客户端用 Socket 连接，双方以对象流收发 Message。指出单线程只能处理一个连接的缺陷，提出线程池改进，并说明 BIO 模型局限，引出后续用 Netty 实现 NIO 的方案。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结05 Socket 网络通信实战05 Socket 网络通信实战前言guide-rpc-framework 的第一版使用的是 JDK 提供了 socket 进行网络编程。为了搞懂具体原理，我们首先要学会使用 Socket 进行网络通信。什么是 Socket(套接字)Socket 是一个抽象概念，应用程序可以通过它发送或接收数据。在使用 Socket 进行网络通信的时候，通过 Socket 就可以让我们的数据在网络中传输。操作套接字的时候，和我们读写文件很像。套接字是 IP 地址与端口的组合，套接字 Socket=（IP 地址：端口号）。要通过互联网进行通信，至少需要一对套接字：运行于服务器端的 Server Socket。运行于客户机端的 Client Socket在 Java 开发中使用 Socket 时会常用到两个类，都在 java.net 包中：Socket: 一般用于客户端ServerSocket :用于服务端Socket 网络通信过程Socket 网络通信过程如下图所示：https://www.javatpoint.com/socket-programmingSocket 网络通信过程简单来说分为下面 4 步：建立服务端并且监听客户端请求客户端请求，服务端和客户端建立连接两端之间可以传递数据关闭资源对应到服务端和客户端的话，是下面这样的。服务器端：创建 ServerSocket 对象并且绑定地址（ip）和端口号(port)：server.bind(new InetSocketAddress(host, port))通过 accept()方法监听客户端请求连接建立后，通过输入流读取客户端发送的请求信息通过输出流向客户端发送响应信息关闭相关资源客户端：创建Socket 对象并且连接指定的服务器的地址（ip）和端口号(port)：socket.connect(inetSocketAddress)连接建立后，通过输出流向服务器端发送请求信息通过输入流获取服务器响应的信息关闭相关资源Socket 网络通信实战服务端public class HelloServer { private static final Logger logger = LoggerFactory.getLogger(HelloServer.class); public void start(int port) { //1.创建 ServerSocket 对象并且绑定一个端口 try (ServerSocket server = new ServerSocket(port);) { Socket socket; //2.通过 accept()方法监听客户端请求 while ((socket = server.accept()) != null) { logger.info("client connected"); try (ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) { //3.通过输入流读取客户端发送的请求信息 Message message = (Message) objectInputStream.readObject(); logger.info("server receive message:" + message.getContent()); message.setContent("new content"); //4.通过输出流向客户端发送响应信息 objectOutputStream.writeObject(message); objectOutputStream.flush(); } catch (IOException | ClassNotFoundException e) { logger.error("occur exception:", e); } } } catch (IOException e) { logger.error("occur IOException:", e); } } public static void main(String[] args) { HelloServer helloServer = new HelloServer(); helloServer.start(6666); } }ServerSocket 的 accept（） 方法是阻塞方法，也就是说 ServerSocket 在调用 accept（)等待客户端的连接请求时会阻塞，直到收到客户端发送的连接请求才会继续往下执行代码。很明显，我上面演示的代码片段有一个很严重的问题：只能同时处理一个客户端的连接，如果需要管理多个客户端的话，就需要为我们请求的客户端单独创建一个线程。 如下图所示：对应的 Java 代码可能是下面这样的：new Thread(() -> { // 创建 socket 连接 }).start();但是，这样会导致一个很严重的问题：资源浪费。我们知道线程是很宝贵的资源，如果我们为每一次连接都用一个线程处理的话，就会导致线程越来越多，最后达到了极限之后，就无法再创建线程处理请求了。处理的不好的话，甚至可能直接就宕机掉了。很多人就会问了：那有没有改进的方法呢？当然有！ 比较简单并且实际的改进方法就是使用线程池。线程池还可以让线程的创建和回收成本相对较低，并且我们可以指定线程池的可创建线程的最大数量，这样就不会导致线程创建过多，机器资源被不合理消耗。ThreadFactory threadFactory = Executors.defaultThreadFactory(); ExecutorService threadPool = new ThreadPoolExecutor(10, 100, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(100), threadFactory); threadPool.execute(() -> { // 创建 socket 连接 });但是，即使你再怎么优化和改变。也改变不了它的底层仍然是同步阻塞的 BIO 模型的事实，因此无法从根本上解决问题。为了解决上述的问题，Java 1.4 中引入了 NIO ，一种同步非阻塞的 I/O 模型。 由于使用同步非阻塞的 I/O 模型 NIO 来进行网络编程真的太麻烦了。你可以使用基于 NIO 的网络编程框架 Netty ，它将是你最好的选择（前面的章节提到过，后面的章节会详细讲解如何使用 Netty 进行网络编程）！客户端public class HelloClient { private static final Logger logger = LoggerFactory.getLogger(HelloClient.class); public Object send(Message message, String host, int port) { //1. 创建Socket对象并且指定服务器的地址和端口号 try (Socket socket = new Socket(host, port)) { ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream()); //2.通过输出流向服务器端发送请求信息 objectOutputStream.writeObject(message); //3.通过输入流获取服务器响应的信息 ObjectInputStream objectInputStream = new ObjectInputStream(socket.getInputStream()); return objectInputStream.readObject(); } catch (IOException | ClassNotFoundException e) { logger.error("occur exception:", e); } return null; } public static void main(String[] args) { HelloClient helloClient = new HelloClient(); Message message = (Message) helloClient.send(new Message("content from client"), "127.0.0.1", 6666); System.out.println("client receive message:" + message.getContent()); } }发送的消息实体类：@Data @AllArgsConstructor public class Message implements Serializable { private String content; }首先运行服务端，然后再运行客户端，控制台输出如下：服务端:[main] INFO github.javaguide.socket.HelloServer - client connected [main] INFO github.javaguide.socket.HelloServer - server receive message:content from client客户端：client receive message:new content好的！我们的第一个使用 Socket 进行网络编程的案例已经完成了。下一篇我们来看看如何使用 Netty 进行网络编程。
AI 摘要 文章介绍序列化与反序列化概念、TCP/IP模型归属，对比JDK、Kryo、Protobuf、ProtoStuff、Hessian等协议优缺点，指出Kryo适合纯Java高性能场景，Protobuf/ProtoStuff/Hessian支持跨语言，建议按需求选型。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结04 序列化介绍以及序列化协议选择什么是序列化和反序列化?如果我们需要持久化 Java 对象比如将 Java 对象保存在文件中，或者在网络传输 Java 对象，这些场景都需要用到序列化。简单来说：序列化： 将数据结构或对象转换成二进制字节流的过程反序列化：将在序列化过程中所生成的二进制字节流转换成数据结构或者对象的过程对于 Java 这种面向对象编程语言来说，我们序列化的都是对象（Object）也就是实例化后的类(Class)，但是在 C++这种半面向对象的语言中，struct(结构体)定义的是数据结构类型，而 class 对应的是对象类型。下面是序列化和反序列化常见应用场景：对象在进行网络传输（比如远程方法调用 RPC 的时候）之前需要先被序列化，接收到序列化的对象之后需要再进行反序列化；将对象存储到文件之前需要进行序列化，将对象从文件中读取出来需要进行反序列化；将对象存储到数据库（如 Redis）之前需要用到序列化，将对象从缓存数据库中读取出来需要反序列化；将对象存储到内存之前需要进行序列化，从内存中读取出来之后需要进行反序列化。维基百科是如是介绍序列化的：序列化（serialization）在计算机科学的数据处理中，是指将数据结构或对象状态转换成可取用格式（例如存成文件，存于缓冲，或经由网络中发送），以留待后续在相同或另一台计算机环境中，能恢复原先状态的过程。依照序列化格式重新获取字节的结果时，可以利用它来产生与原始对象相同语义的副本。对于许多对象，像是使用大量引用的复杂对象，这种序列化重建的过程并不容易。面向对象中的对象序列化，并不概括之前原始对象所关系的函数。这种过程也称为对象编组（marshalling）。从一系列字节提取数据结构的反向操作，是反序列化（也称为解编组、deserialization、unmarshalling）。综上：序列化的主要目的是通过网络传输对象或者说是将对象存储到文件系统、数据库、内存中。https://www.corejavaguru.com/java/serialization/interview-questions-1序列化协议对应于 TCP/IP 4 层模型的哪一层？我们知道网络通信的双方必须要采用和遵守相同的协议。TCP/IP 四层模型是下面这样的，序列化协议属于哪一层呢？应用层传输层网络层网络接口层如上图所示，OSI 七层协议模型中，表示层做的事情主要就是对应用层的用户数据进行处理转换为二进制流。反过来的话，就是将二进制流转换成应用层的用户数据。这不就对应的是序列化和反序列化么？因为，OSI 七层协议模型中的应用层、表示层和会话层对应的都是 TCP/IP 四层模型中的应用层，所以序列化协议属于 TCP/IP 协议应用层的一部分。常见序列化协议有哪些？JDK 自带的序列化方式一般不会用 ，因为序列化效率低并且存在安全问题。比较常用的序列化协议有 Hessian、Kryo、Protobuf、ProtoStuff，这些都是基于二进制的序列化协议。像 JSON 和 XML 这种属于文本类序列化方式。虽然可读性比较好，但是性能较差，一般不会选择。JDK 自带的序列化方式JDK 自带的序列化，只需实现 java.io.Serializable接口即可。@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 RpcMessageTypeEnum rpcMessageTypeEnum; }serialVersionUID 有什么作用？序列化号 serialVersionUID 属于版本控制的作用。反序列化时，会检查 serialVersionUID 是否和当前类的 serialVersionUID 一致。如果 serialVersionUID 不一致则会抛出 InvalidClassException 异常。强烈推荐每个序列化类都手动指定其 serialVersionUID，如果不手动指定，那么编译器会动态生成默认的 serialVersionUID。serialVersionUID 不是被 static 变量修饰了吗？为什么还会被“序列化”？static 修饰的变量是静态变量，位于方法区，本身是不会被序列化的。但是，serialVersionUID 的序列化做了特殊处理，在序列化时，会将 serialVersionUID 序列化到二进制字节流中；在反序列化时，也会解析它并做一致性判断。官方说明如下：A serializable class can declare its own serialVersionUID explicitly by declaring a field named "serialVersionUID" that must be static, final, and of type long;如果想显式指定 serialVersionUID ，则需要在类中使用 static 和 final 关键字来修饰一个 long 类型的变量，变量名字必须为 "serialVersionUID" 。也就是说，serialVersionUID 只是用来被 JVM 识别，实际并没有被序列化。如果有些字段不想进行序列化怎么办？对于不想进行序列化的变量，可以使用 transient 关键字修饰。transient 关键字的作用是：阻止实例中那些用此关键字修饰的的变量序列化；当对象被反序列化时，被 transient 修饰的变量值不会被持久化和恢复。关于 transient 还有几点注意：transient 只能修饰变量，不能修饰类和方法。transient 修饰的变量，在反序列化后变量值将会被置成类型的默认值。例如，如果是修饰 int 类型，那么反序列后结果就是 0。static 变量因为不属于任何对象(Object)，所以无论有没有 transient 关键字修饰，均不会被序列化。为什么不推荐使用 JDK 自带的序列化？我们很少或者说几乎不会直接使用 JDK 自带的序列化方式，主要原因有下面这些原因：不支持跨语言调用 : 如果调用的是其他语言开发的服务的时候就不支持了。性能差 ：相比于其他序列化框架性能更低，主要原因是序列化之后的字节数组体积较大，导致传输成本加大。存在安全问题 ：序列化和反序列化本身并不存在问题。但当输入的反序列化的数据可被用户控制，那么攻击者即可通过构造恶意输入，让反序列化产生非预期的对象，在此过程中执行构造的任意代码。相关阅读：应用安全:JAVA反序列化漏洞之殇 - Cryin 、Java反序列化安全漏洞怎么回事? - Monica。KryoKryo 是一个高性能的序列化/反序列化工具，由于其变长存储特性并使用了字节码生成机制，拥有较高的运行速度和较小的字节码体积。另外，Kryo 已经是一种非常成熟的序列化实现了，已经在 Twitter、Groupon、Yahoo 以及多个著名开源项目（如 Hive、Storm）中广泛的使用。guide-rpc-framework 就是使用的 kryo 进行序列化，序列化和反序列化相关的代码如下：/** * Kryo serialization class, Kryo serialization efficiency is very high, but only compatible with Java language * * @author shuang.kou * @createTime 2020年05月13日 19:29:00 */ @Slf4j public class KryoSerializer implements Serializer { /** * Because Kryo is not thread safe. So, use ThreadLocal to store Kryo objects */ private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> { Kryo kryo = new Kryo(); kryo.register(RpcResponse.class); kryo.register(RpcRequest.class); return kryo; }); @Override public byte[] serialize(Object obj) { try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); Output output = new Output(byteArrayOutputStream)) { Kryo kryo = kryoThreadLocal.get(); // Object->byte:将对象序列化为byte数组 kryo.writeObject(output, obj); kryoThreadLocal.remove(); return output.toBytes(); } catch (Exception e) { throw new SerializeException("Serialization failed"); } } @Override public <T> T deserialize(byte[] bytes, Class<T> clazz) { try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes); Input input = new Input(byteArrayInputStream)) { Kryo kryo = kryoThreadLocal.get(); // byte->Object:从byte数组中反序列化出对象 Object o = kryo.readObject(input, clazz); kryoThreadLocal.remove(); return clazz.cast(o); } catch (Exception e) { throw new SerializeException("Deserialization failed"); } } }Github 地址：https://github.com/EsotericSoftware/kryo 。ProtobufProtobuf 出自于 Google，性能还比较优秀，也支持多种语言，同时还是跨平台的。就是在使用中过于繁琐，因为你需要自己定义 IDL 文件和生成对应的序列化代码。这样虽然不灵活，但是，另一方面导致 protobuf 没有序列化漏洞的风险。Protobuf 包含序列化格式的定义、各种语言的库以及一个 IDL 编译器。正常情况下你需要定义 proto 文件，然后使用 IDL 编译器编译成你需要的语言一个简单的 proto 文件如下：// protobuf的版本 syntax = "proto3"; // SearchRequest会被编译成不同的编程语言的相应对象，比如Java中的class、Go中的struct message Person { //string类型字段 string name = 1; // int 类型字段 int32 age = 2; }Github 地址：https://github.com/protocolbuffers/protobuf。ProtoStuff由于 Protobuf 的易用性，它的哥哥 Protostuff 诞生了。protostuff 基于 Google protobuf，但是提供了更多的功能和更简易的用法。虽然更加易用，但是不代表 ProtoStuff 性能更差。Github 地址：https://github.com/protostuff/protostuff。HessianHessian 是一个轻量级的，自定义描述的二进制 RPC 协议。Hessian 是一个比较老的序列化实现了，并且同样也是跨语言的。Dubbo2.x 默认启用的序列化方式是 Hessian2 ,但是，Dubbo 对 Hessian2 进行了修改，不过大体结构还是差不多。总结Kryo 是专门针对 Java 语言序列化方式并且性能非常好，如果你的应用是专门针对 Java 语言的话可以考虑使用，并且 Dubbo 官网的一篇文章中提到说推荐使用 Kryo 作为生产环境的序列化方式。(文章地址：https://dubbo.apache.org/zh/docs/v2.7/user/references/protocol/rest/)像 Protobuf、 ProtoStuff、hessian 这类都是跨语言的序列化方式，如果有跨语言需求的话可以考虑使用。除了我上面介绍到的序列化方式的话，还有像 Thrift，Avro 这些。推荐阅读美团技术团队-序列化和反序列化：https://tech.meituan.com/2015/02/26/serialization-vs-deserialization.html在 Dubbo 中使用高效的 Java 序列化（Kryo 和 FST）: https://dubbo.apache.org/zh/docs/v3.0/references/serializations/serialization/
AI 摘要 HTTP 与 RPC 同属 TCP 之上的应用层协议，但定位不同：HTTP 面向浏览器通用访问，文本冗余大；RPC 面向内部服务，可定制二进制序列化、连接池，性能更高，像调本地函数。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结从 TCP 聊起作为一个程序员，假设我们需要在 A 电脑的进程发一段数据到 B 电脑的进程，我们一般会在代码里使用 Socket 进行编程。这时候，我们可选项一般也就 TCP 和 UDP 二选一。TCP 可靠，UDP 不可靠。除非是马总这种神级程序员（早期 QQ 大量使用 UDP），否则，只要稍微对可靠性有些要求，普通人一般无脑选 TCP 就对了。类似下面这样。fd = socket(AF_INET,SOCK_STREAM,0);其中 SOCK_STREAM，是指使用字节流传输数据，说白了就是 TCP 协议。在定义了 Socket 之后，我们就可以愉快的对这个 Socket 进行操作，比如用 bind() 绑定 IP 端口，用 connect() 发起建连。在连接建立之后，我们就可以使用 send() 发送数据，recv() 接收数据。光这样一个纯裸的 TCP 连接，就可以做到收发数据了，那是不是就够了？不行，这么用会有问题。使用纯裸 TCP 会有什么问题八股文常背，TCP 是有三个特点，面向连接、可靠、基于字节流。这三个特点真的概括的非常精辟，这个八股文我们没白背。每个特点展开都能聊一篇文章，而今天我们需要关注的是基于字节流这一点。字节流可以理解为一个双向的通道里流淌的数据，这个数据其实就是我们常说的二进制数据，简单来说就是一大堆 01 串。纯裸 TCP 收发的这些 01 串之间是没有任何边界的，你根本不知道到哪个地方才算一条完整消息。正因为这个没有任何边界的特点，所以当我们选择使用 TCP 发送"夏洛"和"特烦恼"的时候，接收端收到的就是"夏洛特烦恼"，这时候接收端没发区分你是想要表达"夏洛"+"特烦恼"还是"夏洛特"+"烦恼"。这就是所谓的粘包问题，之前也写过一篇专门的文章聊过这个问题。说这个的目的是为了告诉大家，纯裸 TCP 是不能直接拿来用的，你需要在这个基础上加入一些自定义的规则，用于区分消息边界。于是我们会把每条要发送的数据都包装一下，比如加入消息头，消息头里写清楚一个完整的包长度是多少，根据这个长度可以继续接收数据，截取出来后它们就是我们真正要传输的消息体。而这里头提到的消息头，还可以放各种东西，比如消息体是否被压缩过和消息体格式之类的，只要上下游都约定好了，互相都认就可以了，这就是所谓的协议。每个使用 TCP 的项目都可能会定义一套类似这样的协议解析标准，他们可能有区别，但原理都类似。于是基于 TCP，就衍生了非常多的协议，比如 HTTP 和 RPC。HTTP 和 RPC我们回过头来看网络的分层图。TCP 是传输层的协议，而基于 TCP 造出来的 HTTP 和各类 RPC 协议，它们都只是定义了不同消息格式的应用层协议而已。HTTP 协议（Hyper Text Transfer Protocol），又叫做超文本传输协议。我们用的比较多，平时上网在浏览器上敲个网址就能访问网页，这里用到的就是 HTTP 协议。而 RPC（Remote Procedure Call），又叫做远程过程调用。它本身并不是一个具体的协议，而是一种调用方式。举个例子，我们平时调用一个本地方法就像下面这样。 res = localFunc(req)如果现在这不是个本地方法，而是个远端服务器暴露出来的一个方法 remoteFunc，如果我们还能像调用本地方法那样去调用它，这样就可以屏蔽掉一些网络细节，用起来更方便，岂不美哉？ res = remoteFunc(req)基于这个思路，大佬们造出了非常多款式的 RPC 协议，比如比较有名的gRPC，thrift。值得注意的是，虽然大部分 RPC 协议底层使用 TCP，但实际上它们不一定非得使用 TCP，改用 UDP 或者 HTTP，其实也可以做到类似的功能。到这里，我们回到文章标题的问题。既然有 HTTP 协议，为什么还要有 RPC？其实，TCP 是70 年代出来的协议，而 HTTP 是 90 年代才开始流行的。而直接使用裸 TCP 会有问题，可想而知，这中间这么多年有多少自定义的协议，而这里面就有80 年代出来的 RPC。所以我们该问的不是既然有 HTTP 协议为什么要有 RPC，而是为什么有 RPC 还要有 HTTP 协议。那既然有 RPC 了，为什么还要有 HTTP 呢？现在电脑上装的各种联网软件，比如 xx 管家，xx 卫士，它们都作为客户端（Client）需要跟服务端（Server）建立连接收发消息，此时都会用到应用层协议，在这种 Client/Server (C/S) 架构下，它们可以使用自家造的 RPC 协议，因为它只管连自己公司的服务器就 ok 了。但有个软件不同，浏览器（Browser），不管是 Chrome 还是 IE，它们不仅要能访问自家公司的服务器（Server），还需要访问其他公司的网站服务器，因此它们需要有个统一的标准，不然大家没法交流。于是，HTTP 就是那个时代用于统一 Browser/Server (B/S) 的协议。也就是说在多年以前，HTTP 主要用于 B/S 架构，而 RPC 更多用于 C/S 架构。但现在其实已经没分那么清了，B/S 和 C/S 在慢慢融合。很多软件同时支持多端，比如某度云盘，既要支持网页版，还要支持手机端和 PC 端，如果通信协议都用 HTTP 的话，那服务器只用同一套就够了。而 RPC 就开始退居幕后，一般用于公司内部集群里，各个微服务之间的通讯。那这么说的话，都用 HTTP 得了，还用什么 RPC？仿佛又回到了文章开头的样子，那这就要从它们之间的区别开始说起。HTTP 和 RPC 有什么区别我们来看看 RPC 和 HTTP 区别比较明显的几个点。服务发现首先要向某个服务器发起请求，你得先建立连接，而建立连接的前提是，你得知道 IP 地址和端口。这个找到服务对应的 IP 端口的过程，其实就是服务发现。在 HTTP 中，你知道服务的域名，就可以通过 DNS 服务去解析得到它背后的 IP 地址，默认 80 端口。而 RPC 的话，就有些区别，一般会有专门的中间服务去保存服务名和 IP 信息，比如 Consul 或者 Etcd，甚至是 Redis。想要访问某个服务，就去这些中间服务去获得 IP 和端口信息。由于 DNS 也是服务发现的一种，所以也有基于 DNS 去做服务发现的组件，比如CoreDNS。可以看出服务发现这一块，两者是有些区别，但不太能分高低。底层连接形式以主流的 HTTP/1.1 协议为例，其默认在建立底层 TCP 连接之后会一直保持这个连接（Keep Alive），之后的请求和响应都会复用这条连接。而 RPC 协议，也跟 HTTP 类似，也是通过建立 TCP 长链接进行数据交互，但不同的地方在于，RPC 协议一般还会再建个连接池，在请求量大的时候，建立多条连接放在池内，要发数据的时候就从池里取一条连接出来，用完放回去，下次再复用，可以说非常环保。由于连接池有利于提升网络请求性能，所以不少编程语言的网络库里都会给 HTTP 加个连接池，比如 Go 就是这么干的。可以看出这一块两者也没太大区别，所以也不是关键。传输的内容基于 TCP 传输的消息，说到底，无非都是消息头 Header 和消息体 Body。Header 是用于标记一些特殊信息，其中最重要的是消息体长度。Body 则是放我们真正需要传输的内容，而这些内容只能是二进制 01 串，毕竟计算机只认识这玩意。所以 TCP 传字符串和数字都问题不大，因为字符串可以转成编码再变成 01 串，而数字本身也能直接转为二进制。但结构体呢，我们得想个办法将它也转为二进制 01 串，这样的方案现在也有很多现成的，比如 Json，Protobuf。这个将结构体转为二进制数组的过程就叫序列化，反过来将二进制数组复原成结构体的过程叫反序列化。对于主流的 HTTP/1.1，虽然它现在叫超文本协议，支持音频视频，但 HTTP 设计初是用于做网页文本展示的，所以它传的内容以字符串为主。Header 和 Body 都是如此。在 Body 这块，它使用 Json 来序列化结构体数据。我们可以随便截个图直观看下。可以看到这里面的内容非常多的冗余，显得非常啰嗦。最明显的，像 Header 里的那些信息，其实如果我们约定好头部的第几位是 Content-Type，就不需要每次都真的把"Content-Type"这个字段都传过来，类似的情况其实在 body 的 Json 结构里也特别明显。而 RPC，因为它定制化程度更高，可以采用体积更小的 Protobuf 或其他序列化协议去保存结构体数据，同时也不需要像 HTTP 那样考虑各种浏览器行为，比如 302 重定向跳转啥的。因此性能也会更好一些，这也是在公司内部微服务中抛弃 HTTP，选择使用 RPC 的最主要原因。当然上面说的 HTTP，其实特指的是现在主流使用的 HTTP/1.1，HTTP/2 在前者的基础上做了很多改进，所以性能可能比很多 RPC 协议还要好，甚至连 gRPC 底层都直接用的 HTTP/2。那么问题又来了，为什么既然有了 HTTP/2，还要有 RPC 协议？这个是由于 HTTP/2 是 2015 年出来的。那时候很多公司内部的 RPC 协议都已经跑了好些年了，基于历史原因，一般也没必要去换了。总结纯裸 TCP 是能收发数据，但它是个无边界的数据流，上层需要定义消息格式用于定义消息边界。于是就有了各种协议，HTTP 和各类 RPC 协议就是在 TCP 之上定义的应用层协议。RPC 本质上不算是协议，而是一种调用方式，而像 gRPC 和 Thrift 这样的具体实现，才是协议，它们是实现了 RPC 调用的协议。目的是希望程序员能像调用本地方法那样去调用远端的服务方法。同时 RPC 有很多种实现方式，不一定非得基于 TCP 协议。从发展历史来说，HTTP 主要用于 B/S 架构，而 RPC 更多用于 C/S 架构。但现在其实已经没分那么清了，B/S 和 C/S 在慢慢融合。很多软件同时支持多端，所以对外一般用 HTTP 协议，而内部集群的微服务之间则采用 RPC 协议进行通讯。RPC 其实比 HTTP 出现的要早，且比目前主流的 HTTP/1.1 性能要更好，所以大部分公司内部都还在使用 RPC。HTTP/2.0 在 HTTP/1.1 的基础上做了优化，性能可能比很多 RPC 协议都要好，但由于是这几年才出来的，所以也不太可能取代掉 RPC。