0
0

【手写一个RPC框架】simpleRPC-06

在这里插入图片描述

本项目所有代码可见：https://github.com/weiyu-zeng/SimpleRPC

前言

本次改进我们将引入zookeeper作为RPC框架的注册中心，服务端在zookeeper上注册自己的服务，而客户端调用服务，回去zookeeper上根据服务名寻找调用的服务器地址，使得我们RPC支持集群调度通信的能力。

实现

zookeeper安装与使用

zookeeper安装请见：

【zookeeper】windows版zookeeper安装与启动 可能遇到的各种问题

安装好之后，我们打开zookeeper的server：

在这里插入图片描述
server启动如下：
在这里插入图片描述

开启zookeeper的client：

在这里插入图片描述
如下，说明成功启动了
在这里插入图片描述

按回车：
在这里插入图片描述

输入

  1. ls /

我们查看目录：

在这里插入图片描述

到此为止，先放在这不要关，我们写代码去。

项目创建

创建一个名为simpleRPC-06的module：

在这里插入图片描述

创建com.rpc的package：

在这里插入图片描述

依赖配置

pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <parent>
  6.         <artifactId>SimpleRPC</artifactId>
  7.         <groupId>org.example</groupId>
  8.         <version>1.0-SNAPSHOT</version>
  9.     </parent>
  10.     <modelVersion>4.0.0</modelVersion>
  12.     <artifactId>simpleRPC-06</artifactId>
  14.     <properties>
  15.         <maven.compiler.source>8</maven.compiler.source>
  16.         <maven.compiler.target>8</maven.compiler.target>
  17.     </properties>
  19.     <dependencies>
  20.         <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
  21.         <dependency>
  22.             <groupId>org.projectlombok</groupId>
  23.             <artifactId>lombok</artifactId>
  24.             <version>1.18.12</version>
  25.             <scope>provided</scope>
  26.         </dependency>
  28.         <dependency>
  29.             <groupId>io.netty</groupId>
  30.             <artifactId>netty-all</artifactId>
  31.             <version>4.1.51.Final</version>
  32.         </dependency>
  34.         <!-- 阿里的fastjson序列化框架 -->
  35.         <dependency>
  36.             <groupId>com.alibaba</groupId>
  37.             <artifactId>fastjson</artifactId>
  38.             <version>1.2.67</version>
  39.         </dependency>
  41.         <!--这个jar包应该依赖log4j,不引入log4j会有控制台会有warn，但不影响正常使用-->
  42.         <dependency>
  43.             <groupId>org.apache.curator</groupId>
  44.             <artifactId>curator-recipes</artifactId>
  45.             <version>2.13.0</version>
  46.         </dependency>
  47.         <dependency>
  48.             <groupId>org.slf4j</groupId>
  49.             <artifactId>slf4j-nop</artifactId>
  50.             <version>1.7.30</version>
  51.         </dependency>
  52.     </dependencies>
  54. </project>

请注意一下，curator必须要和zookeeper版本适配，如果curator版本太高，项目将无法运行。

我们在resources目录下配置一下 log4j的配置，文件名为 log4j.properties：

在这里插入图片描述

log4j.properties

  1. log4j.rootLogger=INFO, stdout 
  2. log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
  3. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
  4. log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
  5. log4j.appender.logfile=org.apache.log4j.FileAppender 
  6. log4j.appender.logfile.File=target/spring.log
  7. log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
  8. log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

register

我们创建一个名为register的package：

在这里插入图片描述

创建注册中心的注册服务接口ServiceRegister.java：

  1. package com.rpc.register;
  3. import java.net.InetSocketAddress;
  5. /**
  6.  * @author weiyu_zeng
  7.  *
  8.  * 服务注册接口，两大基本功能，注册：保存服务与地址。 查询：根据服务名查找地址
  9.  */
  10. public interface ServiceRegister {
  12.     void register(String serviceName, InetSocketAddress serverAddress);
  14.     InetSocketAddress serviceDiscovery(String serviceName);
  15. }

然后创建服务注册实现类 ZkServiceRegister.java：

  1. package com.rpc.register;
  3. import org.apache.curator.RetryPolicy;
  4. import org.apache.curator.framework.CuratorFramework;
  5. import org.apache.curator.framework.CuratorFrameworkFactory;
  6. import org.apache.curator.retry.ExponentialBackoffRetry;
  7. import org.apache.zookeeper.CreateMode;
  8. import org.apache.zookeeper.data.Stat;
  10. import java.net.InetSocketAddress;
  11. import java.util.List;
  13. /**
  14.  * @author weiyu_zeng
  15.  *
  16.  * Curator：是Zookeeper开源的客户端框架，封装了很多API，使用起来非常的方便
  17.  * CuratorFramework：连接zookeeper服务的框架，客户端创建使用静态工厂方式CuratorFrameworkFactory进行创建
  18.  * tickTime：zk的心跳间隔(heartbeat interval）,也是session timeout基本单位.单位为毫秒.
  19.  * minSessionTimeout:最小超时时间,zk设置的默认值为2*tickTime.
  20.  * maxSessionTimeout:最大超时时间,zk设置的默认值为20*tickTime.
  21.  * retryPolicy()重连策略：
  22.  * Curator 四种重连策略：
  23.  *      1.RetryUntilElapsed(int maxElapsedTimeMs, int sleepMsBetweenRetries)
  24.  *      以sleepMsBetweenRetries的间隔重连,直到超过maxElapsedTimeMs的时间设置
  25.  *
  26.  *      2.RetryNTimes(int n, int sleepMsBetweenRetries)
  27.  *      指定重连次数
  28.  *
  29.  *      3.RetryOneTime(int sleepMsBetweenRetry)
  30.  *      重连一次,简单粗暴
  31.  *
  32.  *      4.ExponentialBackoffRetry
  33.  *      ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries)
  34.  *      ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)
  35.  *      时间间隔 = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)))
  36.  *
  37.  * namespace(): 为了避免各个应用的zk patch冲突, Curator Framework内部会给每一个Curator Framework实例分配一个namespace(可选).
  38.  *              这样你在create ZNode的时候都会自动加上这个namespace作为这个node path的root.
  39.  * CuratorFramework.create()：开始创建操作，可以调用额外的方法(比如方式mode 或者后台执行background) 并在最后调用forPath()
  40.  *                            指定要操作的ZNode
  41.  * CuratorFramework.checkExists(): 开始检查ZNode是否存在的操作. 可以调用额外的方法(监控或者后台处理)并在最后调用forPath()
  42.  *                                指定要操作的ZNode
  43.  * CuratorFramework.start() / close()：启动和关闭客户端
  44.  * CuratorFramework(client).create().withMode(CreateMode.EPHEMERAL)：这将使用给定的数据创建临时结点 EPHEMERAL ZNode
  45.  * CuratorFramework.getChildren()：开始获得ZNode的子节点列表。 以调用额外的方法(监控、后台处理或者获取状态watch,
  46.  *                                 background or get stat)并在最后调用forPath()指定要操作的ZNode
  47.  *
  48.  * InetSocketAddress：该类实现了可序列化接口，直接继承自java.net.SocketAddress类。实现 IP 套接字地址（IP 地址 + 端口号）。
  49.  *                    它还可以是一个对（主机名 + 端口号），在此情况下，将尝试解析主机名。如果解析失败，则该地址将被视为未解析
  50.  *                    地址，但是其在某些情形下仍然可以使用，比如通过代理连接。
  51.  *                    构造方法：InetSocketAddress(InetAddress addr,  int port)  根据 IP 地址和端口号创建套接字地址。
  52.  *                             InetSocketAddress(String hostname, int port) 根据主机名（IP地址指代）和端口号创建套接字地址。
  53.  * InetSocketAddress.getHostName()：获取 hostname。即地址的主机名部分。
  54.  * InetSocketAddress.getPort()  获取端口号。
  55.  */
  56. public class ZkServiceRegister implements ServiceRegister {
  58.     // curator 提供的zookeeper客户端
  59.     private CuratorFramework client;
  61.     // zookeeper根路径结点
  62.     private static final String ROOT_PATH = "MyRPC";
  64.     // 构造方法
  65.     // 这里负责zookeeper客户端的初始化，并与zookeeper服务端建立连接。
  66.     // 初始化包括指定重连策略，指定连接zookeeper的端口，指定超时时间，指定命名空间
  67.     // 初始化完成之后start()开启zookeeper客户端。
  68.     public ZkServiceRegister() {
  70.         // 重连策略：指数时间重试
  71.         RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);
  73.         // zookeeper的地址固定，不管是服务提供者还是消费者，都要与之建立连接
  74.         // sessionTimeoutMs 与 zoo.cfg中的tickTime 有关系，
  75.         // zk还会根据minSessionTimeout与maxSessionTimeout两个参数重新调整最后的超时值。默认分别为tickTime 的2倍和20倍
  76.         // 使用心跳监听状态
  77.         this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
  78.                                              .sessionTimeoutMs(40000)
  79.                                              .retryPolicy(policy)
  80.                                              .namespace(ROOT_PATH)
  81.                                              .build();
  82.         this.client.start();
  83.         System.out.println("zookeeper 连接成功");
  84.     }
  86.     // 注册：传入服务方法名(String)，传入主机名和端口号的套接字地址(InetSocketAddress)
  87.     @Override
  88.     public void register(String serviceName, InetSocketAddress serverAddress) {
  89.         try {
  90.             // serviceName创建成永久节点，服务提供者下线时，不删服务名，只删地址
  91.             Stat stat = client.checkExists().forPath("/" + serviceName);
  92.             if (stat == null) {
  93.                 client.create()
  94.                         .creatingParentsIfNeeded()
  95.                         .withMode(CreateMode.PERSISTENT)
  96.                         .forPath("/" + serviceName);
  97.             }
  98.             // 路径地址，一个/代表一个节点
  99.             String path = "/" + serviceName + "/" + getServiceAddress(serverAddress);
  100.             // 临时节点，服务器下线就删除节点
  101.             client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
  102.         } catch (Exception e) {
  103.             System.out.println("此服务已存在");
  104.         }
  105.     }
  107.     // 根据服务名返回地址
  108.     @Override
  109.     public InetSocketAddress serviceDiscovery(String serviceName) {
  110.         try {
  111.             List<String> strings = client.getChildren().forPath("/" + serviceName);
  112.             // 这里默认用的第一个，后面加负载均衡
  113.             String string = strings.get(0);
  114.             return parseAddress(string);
  115.         } catch (Exception e) {
  116.             e.printStackTrace();
  117.         }
  118.         return null;
  119.     }
  121.     // 地址 -> XXX.XXX.XXX.XXX:port 字符串
  122.     private String getServiceAddress(InetSocketAddress serverAddress) {
  123.         return serverAddress.getHostName() + ":" + serverAddress.getPort();
  124.     }
  126.     // 字符串解析为地址：按照":"切分开，前半是host(String)，后半是port(int)
  127.     private InetSocketAddress parseAddress(String address) {
  128.         String[] result = address.split(":");
  129.         return new InetSocketAddress(result[0], Integer.parseInt(result[1]));
  130.     }
  131. }

接下来可以对service，client和server进行修改。

client

NettyRPCClient.java 做一点修改：

  1. package com.rpc.client;
  3. import com.rpc.common.RPCRequest;
  4. import com.rpc.common.RPCResponse;
  5. import com.rpc.register.ServiceRegister;
  6. import com.rpc.register.ZkServiceRegister;
  7. import io.netty.bootstrap.Bootstrap;
  8. import io.netty.channel.Channel;
  9. import io.netty.channel.ChannelFuture;
  10. import io.netty.channel.EventLoopGroup;
  11. import io.netty.channel.nio.NioEventLoopGroup;
  12. import io.netty.channel.socket.nio.NioSocketChannel;
  13. import io.netty.util.AttributeKey;
  15. import java.net.InetSocketAddress;
  17. /**
  18.  * @author zwy
  19.  *
  20.  * 实现RPCClient接口
  21.  */
  22. public class NettyRPCClient implements RPCClient {
  23.     private static final Bootstrap bootstrap;
  24.     private static final EventLoopGroup eventLoopGroup;
  25.     private String host;
  26.     private int port;
  27.     private ServiceRegister serviceRegister; // ServiceRegister接口类class
  29.     // 构造函数：初始化zookeeper
  30.     public NettyRPCClient() {
  31.         this.serviceRegister = new ZkServiceRegister();
  32.     }
  33.     // netty客户端初始化，重复使用
  34.     static {
  35.         eventLoopGroup = new NioEventLoopGroup();
  36.         bootstrap = new Bootstrap();
  37.         bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
  38.                 .handler(new NettyClientInitializer());
  39.     }
  41.     /**
  42.      * 这里需要操作一下，因为netty的传输都是异步的，你发送request，会立刻返回一个值， 而不是想要的相应的response
  43.      */
  44.     @Override
  45.     public RPCResponse sendRequest(RPCRequest request) {
  46.         InetSocketAddress address = serviceRegister.serviceDiscovery(request.getInterfaceName());
  47.         host = address.getHostName();
  48.         port = address.getPort();
  49.         try {
  50.             ChannelFuture channelFuture  = bootstrap.connect(host, port).sync();
  51.             Channel channel = channelFuture.channel();
  52.             // 发送数据
  53.             channel.writeAndFlush(request);
  54.             channel.closeFuture().sync();
  55.             // 阻塞的获得结果，通过给channel设计别名，获取特定名字下的channel中的内容（这个在hanlder中设置）
  56.             // AttributeKey是，线程隔离的，不会由线程安全问题。
  57.             // 实际上不应通过阻塞，可通过回调函数，后面可以再进行优化
  58.             AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
  59.             RPCResponse response = channel.attr(key).get();
  61.             System.out.println(response);
  62.             return response;
  63.         } catch (InterruptedException e) {
  64.             e.printStackTrace();
  65.         }
  66.         return null;
  67.     }
  68. }

TestClient.java 也做相应的修改：

  1. package com.rpc.client;
  3. import com.rpc.common.Blog;
  4. import com.rpc.common.User;
  5. import com.rpc.service.BlogService;
  6. import com.rpc.service.UserService;
  8. /**
  9.  * @author zwy
  10.  */
  11. public class TestClient {
  12.     public static void main(String[] args) {
  13.         // 不需传host，port
  14.         RPCClient rpcClient = new NettyRPCClient();
  15.         // 把这个客户端传入代理客户端
  16.         RPCClientProxy rpcClientProxy = new RPCClientProxy(rpcClient);
  17.         // 代理客户端根据不同的服务，获得一个代理类， 并且这个代理类的方法以或者增强（封装数据，发送请求）
  18.         UserService userService = rpcClientProxy.getProxy(UserService.class);
  20.         // 服务的方法1
  21.         User userByUserId = userService.getUserByUserId(10);
  22.         System.out.println("从服务器端得到的user为：" + userByUserId);
  24.         // 服务的方法2
  25.         User user = User.builder().userName("张三").id(100).sex(true).build();
  26.         Integer integer = userService.insertUserId(user);
  27.         System.out.println("向服务器端插入数据" + integer);
  29.         // 服务的方法3
  30.         BlogService blogService = rpcClientProxy.getProxy(BlogService.class);
  31.         Blog blogById = blogService.getBlogById(10000);
  32.         System.out.println("从服务端得到的blog为：" + blogById);
  33.     }
  34. }

client中的其他代码和simpleRPC-05一样，可以直接从simpleRPC-05复制粘贴过来，为了完整，我还是把代码放下面：

RPCClient.java

  1. package com.rpc.client;
  3. import com.rpc.common.RPCRequest;
  4. import com.rpc.common.RPCResponse;
  6. /**
  7.  * @author zwy
  8.  *
  9.  * RPC客户端：发送请求，获得response
  10.  */
  11. public interface RPCClient {
  12.     RPCResponse sendRequest(RPCRequest request);
  13. }

RPCClientProxy.java

  1. package com.rpc.client;
  3. import com.rpc.common.RPCRequest;
  4. import com.rpc.common.RPCResponse;
  5. import lombok.AllArgsConstructor;
  7. import java.lang.reflect.InvocationHandler;
  8. import java.lang.reflect.Method;
  9. import java.lang.reflect.Proxy;
  11. /**
  12.  * @author zwy
  13.  *
  14.  * 客户端代理：把动态代理封装request对象（这里和simpleRPC-02的ClientProxy函数一样，保留了动态代理的设计）
  15.  */
  16. @AllArgsConstructor
  17. public class RPCClientProxy implements InvocationHandler {
  18.     private RPCClient client;
  20.     // jdk动态代理，每一次代理对象调用方法，会经过此方法增强（反射获取request对象，socket发送至客户端）
  21.     @Override
  22.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  23.         // request的构建，使用了lombok中的builder，更加简洁
  24.         RPCRequest request = RPCRequest.builder().interfaceName(method.getDeclaringClass().getName())
  25.                                        .methodName(method.getName())
  26.                                        .params(args)
  27.                                        .paramsTypes(method.getParameterTypes())
  28.                                        .build();
  29.         // 数据传输
  30.         RPCResponse response = client.sendRequest(request);
  31. //        System.out.println(response);
  32.         return response.getData();
  33.     }
  35.     <T> T getProxy(Class<T> clazz) {
  36.         Object o = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, this);
  37.         return (T)o;
  38.     }
  39. }

NettyClientInitializer.java

  1. package com.rpc.client;
  3. import com.rpc.codec.JsonSerializer;
  4. import com.rpc.codec.MyDecode;
  5. import com.rpc.codec.MyEncode;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.ChannelPipeline;
  8. import io.netty.channel.socket.SocketChannel;
  10. /**
  11.  * @author zwy
  12.  *
  13.  * 同样的与服务端解码和编码格式
  14.  */
  15. public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {
  17.     @Override
  18.     protected void initChannel(SocketChannel ch) throws Exception {
  19.         ChannelPipeline pipeline = ch.pipeline();
  20.         // 使用自定义的编解码器
  21.         pipeline.addLast(new MyDecode());
  22.         // 编码需要传入序列化器，这里是json，还支持ObjectSerializer，也可以自己实现其他的
  23.         pipeline.addLast(new MyEncode(new JsonSerializer()));
  24.         pipeline.addLast(new NettyClientHandler());
  25.     }
  26. }

NettyClientHandler.java

  1. package com.rpc.client;
  3. import com.rpc.common.RPCResponse;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  6. import io.netty.util.AttributeKey;
  8. /**
  9.  * @author zwy
  10.  */
  11. public class NettyClientHandler extends SimpleChannelInboundHandler<RPCResponse> {
  12.     @Override
  13.     protected void channelRead0(ChannelHandlerContext ctx, RPCResponse msg) throws Exception {
  14.         // 接收到response, 给channel设计别名，让sendRequest里读取response
  15.         AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
  16.         ctx.channel().attr(key).set(msg);
  17.         ctx.channel().close();
  18.     }
  20.     // 跟NettyRPCServerHandler一样
  21.     @Override
  22.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  23.         cause.printStackTrace();
  24.         ctx.close();
  25.     }
  26. }

service

服务暴露类加入注册的功能，ServiceProvider.java 做相应的修改：

  1. package com.rpc.service;
  3. import com.rpc.register.ServiceRegister;
  4. import com.rpc.register.ZkServiceRegister;
  6. import java.net.InetSocketAddress;
  7. import java.util.HashMap;
  8. import java.util.Map;
  10. /**
  11.  * @author zwy
  12.  */
  13. public class ServiceProvider {
  14.     /**
  15.      * 一个实现类可能实现多个服务接口，
  16.      */
  17.     private Map<String, Object> interfaceProvider;
  19.     private ServiceRegister serviceRegister;
  20.     private String host;
  21.     private int port;
  23.     public ServiceProvider(String host, int port){
  24.         // 需要传入服务端自身的服务的网络地址
  25.         this.host = host;
  26.         this.port = port;
  27.         this.interfaceProvider = new HashMap<>();
  28.         this.serviceRegister = new ZkServiceRegister();
  29.     }
  31.     public void provideServiceInterface(Object service) throws Exception {
  32.         Class<?>[] interfaces = service.getClass().getInterfaces();
  34.         for(Class clazz : interfaces){
  35.             // 本机的映射表
  36.             interfaceProvider.put(clazz.getName(),service);
  37.             // 在注册中心注册服务
  38.             serviceRegister.register(clazz.getName(), new InetSocketAddress(host, port));
  39.         }
  40.     }
  42.     public Object getService(String interfaceName){
  43.         return interfaceProvider.get(interfaceName);
  44.     }
  45. }

service中的其他代码和simpleRPC-05一样，可以直接从simpleRPC-05复制粘贴过来，为了完整，我还是把代码放下面：

BlogService.java

  1. package com.rpc.service;
  3. import com.rpc.common.Blog;
  5. public interface BlogService {
  7.     Blog getBlogById(Integer id);
  8. }

BlogServiceImpl.java

  1. package com.rpc.service;
  3. import com.rpc.common.Blog;
  5. public class BlogServiceImpl implements BlogService {
  7.     @Override
  8.     public Blog getBlogById(Integer id) {
  9.         Blog blog = Blog.builder()
  10.                         .id(id)
  11.                         .title("我的博客")
  12.                         .useId(22).build();
  13.         System.out.println("客户端查询了" + id + "博客");
  14.         return blog;
  15.     }
  16. }

UserService.java

  1. package com.rpc.service;
  3. import com.rpc.common.User;
  5. /**
  6.  * @author zwy
  7.  */
  8. public interface UserService {
  10.     // 客户端通过这个接口调用服务端的实现类
  11.     User getUserByUserId(Integer id);
  13.     // 给这个服务增加一个功能
  14.     Integer insertUserId(User user);
  15. }

UserServiceImpl.java

  1. package com.rpc.service;
  3. import com.rpc.common.User;
  5. /**
  6.  * @author zwy
  7.  */
  8. public class UserServiceImpl implements UserService {
  10.     @Override
  11.     public User getUserByUserId(Integer id) {
  12.         // 模拟从数据库中取用户的行为
  13.         User user = User.builder()
  14.                         .id(id)
  15.                         .userName("he2121")
  16.                         .sex(true).build();
  17.         System.out.println("客户端查询了" + id + "的用户");
  18.         return user;
  19.     }
  21.     @Override
  22.     public Integer insertUserId(User user) {
  23.         System.out.println("插入数据成功: " + user);
  24.         return 1;
  25.     }
  26. }

server

TestServer.java 做相应的修改：

  1. package com.rpc.server;
  3. import com.rpc.service.*;
  5. public class TestServer {
  6.     public static void main(String[] args) throws Exception {
  7.         UserService userService = new UserServiceImpl();
  8.         BlogService blogService = new BlogServiceImpl();
  10.         // 这里重用了服务暴露类，顺便在注册中心注册，实际上应分开，每个类做各自独立的事
  11.         ServiceProvider serviceProvider = new ServiceProvider("127.0.0.1", 8899); // 8899
  12.         serviceProvider.provideServiceInterface(userService);
  13.         serviceProvider.provideServiceInterface(blogService);
  15.         RPCServer RPCServer = new NettyRPCServer(serviceProvider);
  16.         RPCServer.start(8899);
  17.     }
  18. }

server中的其他代码和simpleRPC-05一样，可以直接从simpleRPC-05复制粘贴过来，为了完整，我还是把代码放下面：

RPCServer.java

  1. package com.rpc.server;
  3. /**
  4.  * @author zwy
  5.  */
  6. public interface RPCServer {
  7.     void start(int port);
  8.     void stop();
  9. }

NettyRPCServer.java

  1. package com.rpc.server;
  3. import com.rpc.service.ServiceProvider;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import lombok.AllArgsConstructor;
  10. /**
  11.  * @author zwy
  12.  */
  13. @AllArgsConstructor
  14. public class NettyRPCServer implements RPCServer {
  15.     private ServiceProvider serviceProvider;
  17.     @Override
  18.     public void start(int port) {
  19.         // netty服务线程组负责建立连接(TCP/IP连接)，work负责具体的请求
  20.         NioEventLoopGroup bossGroup = new NioEventLoopGroup();
  21.         NioEventLoopGroup workGroup = new NioEventLoopGroup();
  22.         System.out.println("Netty服务端启动了");
  24.         try {
  25.             // 启动Netty服务器
  26.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  27.             // 初始化
  28.             serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
  29.                            .childHandler(new NettyServerInitializer(serviceProvider));
  30.             // 同步阻塞
  31.             ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
  32.             // 死循环监听
  33.             channelFuture.channel().closeFuture().sync();
  34.         } catch (InterruptedException e) {
  35.             e.printStackTrace();
  36.         } finally {
  37.             bossGroup.shutdownGracefully();
  38.             workGroup.shutdownGracefully();
  39.         }
  40.     }
  42.     @Override
  43.     public void stop() {
  45.     }
  46. }

NettyServerInitializer.java

  1. package com.rpc.server;
  3. import com.rpc.codec.JsonSerializer;
  4. import com.rpc.codec.MyDecode;
  5. import com.rpc.codec.MyEncode;
  6. import com.rpc.service.ServiceProvider;
  7. import io.netty.channel.ChannelInitializer;
  8. import io.netty.channel.ChannelPipeline;
  9. import io.netty.channel.socket.SocketChannel;
  10. import lombok.AllArgsConstructor;
  12. /**
  13.  * @author zwy
  14.  */
  15. @AllArgsConstructor
  16. public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {
  17.     private ServiceProvider serviceProvider;
  19.     @Override
  20.     protected void initChannel(SocketChannel ch) throws Exception {
  21.         ChannelPipeline pipeline = ch.pipeline();
  22.         // 使用自定义的解码器
  23.         pipeline.addLast(new MyDecode());
  24.         // 使用自定义的编码器，而且解码器需要传入序列化器，这里是json，还支持ObjectSerializer，也可以自己实现其他的
  25.         pipeline.addLast(new MyEncode(new JsonSerializer()));
  26.         pipeline.addLast(new NettyRPCServerHandler(serviceProvider));
  27.     }
  28. }

NettyRPCServerHandler.java

  1. package com.rpc.server;
  3. import com.rpc.common.RPCRequest;
  4. import com.rpc.common.RPCResponse;
  5. import com.rpc.service.ServiceProvider;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.SimpleChannelInboundHandler;
  8. import lombok.AllArgsConstructor;
  10. import java.lang.reflect.InvocationTargetException;
  11. import java.lang.reflect.Method;
  13. /**
  14.  * @author zwy
  15.  */
  16. @AllArgsConstructor
  17. public class NettyRPCServerHandler extends SimpleChannelInboundHandler<RPCRequest> {
  18.     private ServiceProvider serviceProvider;
  20.     @Override
  21.     protected void channelRead0(ChannelHandlerContext ctx, RPCRequest msg) throws Exception {
  22.         // System.out.println(msg);
  23.         RPCResponse response = getResponse(msg);
  24.         ctx.writeAndFlush(response);
  25.         ctx.close();
  26.     }
  28.     @Override
  29.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  30.         cause.printStackTrace();
  31.         ctx.close();
  32.     }
  34.     // 这里和WorkThread里的getResponse差不多
  35.     RPCResponse getResponse(RPCRequest request) {
  36.         // 得到服务名
  37.         String interfaceName = request.getInterfaceName();
  38.         // 得到服务器相应类
  39.         Object service = serviceProvider.getService(interfaceName);
  40.         // 反射调用方法
  41.         Method method = null;
  42.         try {
  43.             method = service.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
  44.             Object invoke = method.invoke(service, request.getParams());
  45.             return RPCResponse.success(invoke);
  46.         } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
  47.             e.printStackTrace();
  48.             System.out.println("方法执行错误");
  49.             return RPCResponse.fail();
  50.         }
  51.     }
  52. }

common

和simpleRPC-05一样，可以直接复制过来。

在这里插入图片描述

codec

和simpleRPC-05一样，可以直接复制过来。

在这里插入图片描述

文件结构

simpleRPC-06的文件结构如下：

在这里插入图片描述

在这里插入图片描述

运行

启动TestServer.java ：
在这里插入图片描述
然后启动TestClient.java：

在这里插入图片描述
在这里插入图片描述

我们来看看我们最开始开的zookeeper客户端：

现在输入

  1. ls /

在这里插入图片描述

发现我们多了一个结点 MyRPC：

输入

  1. ls /MyRPC

：

在这里插入图片描述
可以看到我们注册的服务都在这里，成功！

标签： rpc 网络 网络协议
本文转载自: https://blog.csdn.net/fisherish/article/details/122277235
版权归原作者 锥栗 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【手写一个RPC框架】simpleRPC-06”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？