[大数据]基于Zookeeper与Netty实现的分布式RPC服务

前言

大部分的互联网公司在应用的迭代演进过程中，随着系统访问量提高，业务复杂度提高，代码复杂度提高，应用逐渐从单体式架构向面向服务的分布式架构转变，但具体实现微服务架构的方式有所不同，主流上分为两种，一种是基于Http协议的远程调用，另外一种是基于RPC方式的调用。两种方式都有自己的代表框架，前者是著名的Spring Cloud，后者则是有阿里巴巴开源的Dubbo，二者都被广泛的采用。这篇文章，我们就一起来了解一下RPC，并且动手实现一个简单的RPC框架的Demo。

架构

在这个demo中主要有以下角色：

• Provider: 暴露服务的服务提供方。

• Consumer: 调用远程服务的服务消费方。

• Registry: 服务注册与发现的注册中心。

该服务功能主要是通过使用服务消费方获取到服务提供方查询到的用户信息，分为三个部分：

• 公共的服务接口API模块
• 服务提供者模块
• 服务消费者模块

其中服务注册与发现主要提供以下功能：

• 服务注册
• 服务实例的获取
• 服务变化的通知机制

前置知识

实现分布式RPC服务主要解决两个问题，一个是服务注册与发现，一个是服务调用者之间的通信问题，前者通过zookeeper实现，后者则基于netty实现。

在看这篇实战案例之前你需要了解以下知识，而这些在我之前的博客已经讲过了，这里就不详细介绍：

• zookeeper安装与使用：https://blog.csdn.net/c15158032319/article/details/118864745
• netty原理与使用：https://blog.csdn.net/c15158032319/article/details/103089874
• RPC原理：https://blog.csdn.net/c15158032319/article/details/116158702

服务接口API

用户服务接口

/**
 * 用户服务
 */
public interface IUserService {

    /**
     * 根据ID查询用户
     *
     * @param id
     * @return
     */
    User getById(int id);
}

用户对象

@Data
@ToString
public class User {
    private int id;
    private String name;
}

RPC请求对象

@Data
public class RpcRequest {

    /**
     * 请求对象的ID
     */
    private String requestId;
    /**
     * 类名
     */
    private String className;
    /**
     * 方法名
     */
    private String methodName;
    /**
     * 参数类型
     */
    private Class<?>[] parameterTypes;
    /**
     * 入参
     */
    private Object[] parameters;

}

RPC响应对象

@Data
public class RpcResponse {

    /**
     * 响应ID
     */
    private String requestId;

    /**
     * 错误信息
     */
    private String error;

    /**
     * 返回的结果
     */
    private Object result;

}

服务提供者

服务提供者启动的时候，我们RPC框架需要做以下几件事情：

1. 扫描服务提供者中所有提供服务的类（被@RpcService修饰的类），并将其交由BeanFactory管理。
2. 启动Netty服务端，用来收到消费者的调用消息，并且返回调用结果。
3. 向注册中心注册，本例中使用的注册中心是Zookeeper。

RPC服务端

@Service
public class RpcServer implements DisposableBean {

    @Autowired
    RpcServerHandler rpcServerHandler;
    private NioEventLoopGroup bossGroup;
    private NioEventLoopGroup workerGroup;

    public void startServer(String ip, int port) {
        try {
            //1. 创建线程组
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();
            //2. 创建服务端启动助手
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //3. 设置参数
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            //添加String的编解码器
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            //业务处理类
                            pipeline.addLast(rpcServerHandler);
                        }
                    });
            //4.绑定端口
            ChannelFuture sync = serverBootstrap.bind(ip, port).sync();
            System.out.println("==========服务端启动成功==========");
            sync.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
            }

            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
        }
    }


    @Override
    public void destroy() throws Exception {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }

        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
    }
}

RPC业务处理

处理逻辑如下：

1. 将标有@RpcService注解的bean缓存

2. 接收客户端请求

3. 根据传递过来的beanName从缓存中查找到对应的bean

4. 解析请求中的方法名称. 参数类型 参数信息

5. 反射调用bean的方法

6. 给客户端进行响应

@Component
@ChannelHandler.Sharable
public class RpcServerHandler extends SimpleChannelInboundHandler<String> implements ApplicationContextAware {

    private static final Map SERVICE_INSTANCE_MAP = new ConcurrentHashMap();

    /**
     * 1.将标有@RpcService注解的bean缓存
     *
     * @param applicationContext
     * @throws BeansException
     */
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String, Object> serviceMap = applicationContext.getBeansWithAnnotation(RpcService.class);
        if (serviceMap != null && serviceMap.size() > 0) {
            Set<Map.Entry<String, Object>> entries = serviceMap.entrySet();
            for (Map.Entry<String, Object> item : entries) {
                Object serviceBean = item.getValue();
                if (serviceBean.getClass().getInterfaces().length == 0) {
                    throw new RuntimeException("服务必须实现接口");
                }
                //默认取第一个接口作为缓存bean的名称
                String name = serviceBean.getClass().getInterfaces()[0].getName();
                SERVICE_INSTANCE_MAP.put(name, serviceBean);
            }
        }
    }

    /**
     * 通道读取就绪事件
     *
     * @param channelHandlerContext
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        //1.接收客户端请求- 将msg转化RpcRequest对象
        RpcRequest rpcRequest = JSON.parseObject(msg, RpcRequest.class);
        RpcResponse rpcResponse = new RpcResponse();
        rpcResponse.setRequestId(rpcRequest.getRequestId());
        try {
            //业务处理
            rpcResponse.setResult(handler(rpcRequest));
        } catch (Exception exception) {
            exception.printStackTrace();
            rpcResponse.setError(exception.getMessage());
        }
        //6.给客户端进行响应
        channelHandlerContext.writeAndFlush(JSON.toJSONString(rpcResponse));

    }

    /**
     * 业务处理逻辑
     *
     * @return
     */
    public Object handler(RpcRequest rpcRequest) throws InvocationTargetException {
        // 3.根据传递过来的beanName从缓存中查找到对应的bean
        Object serviceBean = SERVICE_INSTANCE_MAP.get(rpcRequest.getClassName());
        if (serviceBean == null) {
            throw new RuntimeException("根据beanName找不到服务,beanName:" + rpcRequest.getClassName());
        }
        //4.解析请求中的方法名称. 参数类型 参数信息
        Class<?> serviceBeanClass = serviceBean.getClass();
        String methodName = rpcRequest.getMethodName();
        Class<?>[] parameterTypes = rpcRequest.getParameterTypes();
        Object[] parameters = rpcRequest.getParameters();
        //5.反射调用bean的方法- CGLIB反射调用
        FastClass fastClass = FastClass.create(serviceBeanClass);
        FastMethod method = fastClass.getMethod(methodName, parameterTypes);
        return method.invoke(serviceBean, parameters);
    }
}

暴露接口

使用@RpcService注解的服务即可被注册到zookeeper服务器上去了

@RpcService
@Service
public class UserServiceImpl implements IUserService {
    Map<Object, User> userMap = new HashMap();

    @Override
    public User getById(int id) {
        if (userMap.size() == 0) {
            User user1 = new User();
            user1.setId(1);
            user1.setName("张三");
            User user2 = new User();
            user2.setId(2);
            user2.setName("李四");
            userMap.put(user1.getId(), user1);
            userMap.put(user2.getId(), user2);
        }
        return userMap.get(id);
    }
}

服务注册

@SpringBootApplication
public class RpcProviderApplication implements CommandLineRunner {

    @Autowired
    RpcServer rpcServer;

    public static void main(String[] args) {
        SpringApplication.run(RpcProviderApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("192.168.24.124:2181")
                .sessionTimeoutMs(100)
                .connectionTimeoutMs(15000)
                .retryPolicy(retryPolicy)
                .build();
        client.start();

        int port = 8901;
        new Thread(() -> rpcServer.startServer("127.0.0.1", port)).start();

        //创建服务器节点
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)
                .forPath("/rpc-base/127.0.0.1:" + port, "".getBytes());

    }
}

服务消费者

服务消费者是一个Web服务，所以它还添加了spring-boot-starter-web的依赖。对于服务消费者，我们框架需要对它的处理就是，为所有的RPC服务（被@RpcConsumer修饰的属性）设置上动态代理。具体的设置代码如下所示：

RPC客户端

客户端逻辑如下：

1. 连接Netty服务端

2. 提供给调用者主动关闭资源的方法

3. 提供消息发送的方法

public class RpcClient implements DisposableBean {

    private EventLoopGroup group;

    private Channel channel;

    private String ip;

    private int port;

    private RpcClientHandler rpcClientHandler = new RpcClientHandler();

    private ExecutorService executorService = Executors.newCachedThreadPool();

    public RpcClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
        initClient();
    }

    /**
     * 初始化方法-连接Netty服务端
     */
    public void initClient() {
        try {
            //1.创建线程组
            group = new NioEventLoopGroup();
            //2.创建启动助手
            Bootstrap bootstrap = new Bootstrap();
            //3.设置参数
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            ChannelPipeline pipeline = channel.pipeline();
                            //String类型编解码器
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            //添加客户端处理类
                            pipeline.addLast(rpcClientHandler);
                        }
                    });
            //4.连接Netty服务端
            channel = bootstrap.connect(ip, port).sync().channel();
        } catch (Exception exception) {
            exception.printStackTrace();
            if (channel != null) {
                channel.close();
            }
            if (group != null) {
                group.shutdownGracefully();
            }
        }
    }

    /**
     * 提供给调用者主动关闭资源的方法
     */
    public void close() {
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
    }

    /**
     * 提供消息发送的方法
     */
    public Object send(String msg) throws ExecutionException, InterruptedException {
        rpcClientHandler.setRequestMsg(msg);
        Future submit = executorService.submit(rpcClientHandler);
        return submit.get();
    }

    @Override
    public void destroy() throws Exception {
        if (channel != null) {
            channel.close();
        }
        if (group != null) {
            group.shutdownGracefully();
        }
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }
}

RPC客户端处理类

主要功能就是发送和接收消息：

@ChannelHandler.Sharable
public class RpcClientHandler extends SimpleChannelInboundHandler<String> implements Callable {

    ChannelHandlerContext context;
    //发送的消息
    String requestMsg;

    //服务端的消息
    String responseMsg;

    public void setRequestMsg(String requestMsg) {
        this.requestMsg = requestMsg;
    }

    /**
     * 通道连接就绪事件
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        context = ctx;
    }

    /**
     * 通道读取就绪事件
     *
     * @param channelHandlerContext
     * @param msg
     * @throws Exception
     */
    @Override
    protected synchronized void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        responseMsg = msg;
        //唤醒等待的线程
        notify();
    }

    /**
     * 发送消息到服务端
     *
     * @return
     * @throws Exception
     */
    @Override
    public synchronized Object call() throws Exception {
        //消息发送
        context.writeAndFlush(requestMsg);
        //线程等待
        wait();
        return responseMsg;
    }
}

RPC客户端代理类

客户端代理类-创建代理对象:

1. 封装request请求对象

2. 创建RpcClient对象

3. 发送消息

4. 返回结果

@Component
public class RpcClientProxy {

    private static int index = 0;

    @Autowired
    private MyZkClient myZkClient;

    private RpcClient getClient() {
        if (myZkClient.getServerMap().size() == 0) {
            throw new RuntimeException("无可用服务器");
        }
        String bestServer = myZkClient.getBestServer();
//        Object[] objects = myZkClient.getNodeSet().toArray();
//        String node = objects[index % myZkClient.getServerMap().size()].toString();
        Host host = myZkClient.getServerMap().get(bestServer);
        RpcClient rpcClient = new RpcClient(host.getIp(), host.getPort());
        index++;
        return rpcClient;
    }

    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        //2.创建RpcClient对象
                        RpcClient rpcClient = getClient();

                        try {
                            long startTime = System.currentTimeMillis();
                            //3.发送消息
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            //4.返回结果
                            Object result = rpcResponse.getResult();

                            long endTime = System.currentTimeMillis();
                            //上报请求时长
                            myZkClient.setResponseTimeToNodeData(rpcClient.getIp() + ":" + rpcClient.getPort(),
                                    endTime - startTime);

                            System.out.println(String.format("当前RPC-Client-Connect：%s:%d，结果：%s, 耗时：%d",
                                    rpcClient.getIp(), rpcClient.getPort(), result.toString(), (endTime - startTime)));
                            return JSON.parseObject(result.toString(), method.getReturnType());
                        } catch (Exception e) {
                            throw e;
                        } finally {
                            rpcClient.close();
                        }

                    }
                });
    }

}

服务发现

对注册到zookeeper服务器的服务信息进行拉取

@Component
@EnableScheduling
public class MyZkClient implements Watcher {

    private final static String basePath = "/rpc-base";
    private final static Map<String, Host> serverMap = new HashMap<>();
    private static CuratorFramework client;

    public CuratorFramework getClient() {
        if (client == null) {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
            client = CuratorFrameworkFactory.builder()
                    .connectString("47.106.168.17:2181")
                    .sessionTimeoutMs(60000)
                    .connectionTimeoutMs(15000)
                    .retryPolicy(retryPolicy)
                    .build();
            client.start();
            System.out.println("zookeeper session established.");

        }
        return client;
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        //监听字节的变动事件，并处理
        if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
            getChildren();
        }
        if (watchedEvent.getType() == Event.EventType.NodeDataChanged) {
            String path = watchedEvent.getPath();
            getChildrenData(path);
        }
    }

    /**
     * 获取子节点列表
     */
    public void getChildren() {
        try {
            List<String> nodes = client.getChildren()
                    .usingWatcher(new MyZkClient()).forPath(basePath);
            System.out.println("服务器节点列表：" + nodes);
            //把上线的服务放入map
            for (String node : nodes) {
                if (!serverMap.containsKey(node)) {
                    Host host = new Host().getHost(node);

                    byte[] bytes = client.getData().usingWatcher(new MyZkClient())
                            .forPath(basePath + "/" + node);
                    if (bytes != null && bytes.length > 0) {
                        String[] data = new String(bytes).split("#");
                        if (data.length == 2) {
                            host.setResponseTime(Long.parseLong(data[0]));
                            host.setLastTime(Long.parseLong(data[1]));
                        } else {
                            host.setResponseTime(0);
                            host.setLastTime(0);
                        }
                    }
                    serverMap.put(node, host);
                }
            }

            //删除下线的服务
            Iterator<String> iterator = serverMap.keySet().iterator();
            while (iterator.hasNext()) {
                if (!nodes.contains(iterator.next())) {
                    iterator.remove();
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取服务器节点内容
     *
     * @param path
     */
    public void getChildrenData(String path) {
        try {
//            /rpc-base/127.0.0.1:8901
            byte[] bytes = client.getData().usingWatcher(new MyZkClient()).forPath(path);
            String node = path.substring(path.lastIndexOf("/") + 1);
            Host host = serverMap.get(node);
            if (host != null) {
                String[] data = new String(bytes).split("#");
                if (data.length == 2) {
                    host.setResponseTime(Long.parseLong(data[0]));
                    host.setLastTime(Long.parseLong(data[1]));
                } else {
                    host.setResponseTime(0);
                    host.setLastTime(0);
                }
            }
            System.out.println(String.format("节点：%s,内容变更：%s", node, new String(bytes)));

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 将响应信息上报
     *
     * @param childNode
     * @param time
     */
    public void setResponseTimeToNodeData(String childNode, long time) {
        try {
            //上报【请求耗时|系统时间】rpc-base/127.0.0.1:8901
            client.setData().forPath(basePath + "/" + childNode,
                    (time + "#" + System.currentTimeMillis()).getBytes(CharsetUtil.UTF_8));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取当前最优的服务器
     *
     * @return
     */
    public String getBestServer() {
        long bestTime = -1;
        String key = null;
        for (Map.Entry<String, Host> entry : serverMap.entrySet()) {
            Host host = entry.getValue();
            if (host.getResponseTime() == 0) {
                key = entry.getKey();
                break;
            }
            long responseTime = host.getResponseTime();
            if (bestTime == -1 || bestTime > responseTime) {
                key = entry.getKey();
                bestTime = responseTime;
            }
        }
        return key;
    }

    public Map<String, Host> getServerMap() {
        if (serverMap.size() == 0) {
            getClient();
            getChildren();
        }
        return serverMap;
    }

    /**
     * 定时任务每5秒执行一次，把最后一次请求超过5秒的服务清0
     *
     * @throws Exception
     */
    @Scheduled(cron = "0/5 * * * * ?")
    public void scheduled() throws Exception {
        List<String> nodes = getClient().getChildren().forPath(basePath);
        for (String node : nodes) {
            byte[] bytes = getClient().getData().forPath(basePath + "/" + node);
            //responsetime#lasttime
            String[] data = new String(bytes).split("#");
            if (data.length == 2) {
                if (System.currentTimeMillis() - Long.parseLong(data[1]) > 5000) {
                    getClient().setData().forPath(basePath + "/" + node, "0".getBytes());
                    System.out.println(String.format("---定时任务执行---修改节点：%s, 内容为：%s",
                            node, 0));
                }
            }
        }
    }

}

web接口

这里就是一个controller层和web进行对接使用：

@RestController
@RequestMapping("/user")
public class UserController {

    @RpcReference
    private IUserService userService;

    @RequestMapping("/{id}")
    public String getUserById(@PathVariable int id) {
        return userService.getById(id).toString();
    }
}

就这样大功告成，只要按照启动zookeeper --> 服务提供者 -->服务消费的顺序启动，然后用浏览器访问接口即可了。