主頁 >  其他 > 【手寫一個RPC框架】simpleRPC-06

【手寫一個RPC框架】simpleRPC-06

2022-01-04 09:35:54 其他

在這里插入圖片描述

本專案所有代碼可見：https://github.com/weiyu-zeng/SimpleRPC

前言

本次改進我們將引入zookeeper作為RPC框架的注冊中心，服務端在zookeeper上注冊自己的服務，而客戶端呼叫服務，回去zookeeper上根據服務名尋找呼叫的服務器地址，使得我們RPC支持集群調度通信的能力，

實作

zookeeper安裝與使用

zookeeper安裝請見：

【zookeeper】windows版zookeeper安裝與啟動 可能遇到的各種問題

安裝好之后，我們打開zookeeper的server：

在這里插入圖片描述
server啟動如下：
在這里插入圖片描述

開啟zookeeper的client：

在這里插入圖片描述
如下，說明成功啟動了
在這里插入圖片描述

按回車：
在這里插入圖片描述

輸入ls /我們查看目錄：

在這里插入圖片描述

到此為止，先放在這不要關，我們寫代碼去，

專案創建

創建一個名為simpleRPC-06的module：

在這里插入圖片描述

創建com.rpc的package：

在這里插入圖片描述

依賴配置

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>SimpleRPC</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>simpleRPC-06</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.51.Final</version>
        </dependency>

        <!-- 阿里的fastjson序列化框架 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.67</version>
        </dependency>

        <!--這個jar包應該依賴log4j,不引入log4j會有控制臺會有warn，但不影響正常使用-->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.13.0</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-nop</artifactId>
            <version>1.7.30</version>
        </dependency>
    </dependencies>

</project>

請注意一下，curator必須要和zookeeper版本適配，如果curator版本太高，專案將無法運行，

我們在resources目錄下配置一下 log4j的配置，檔案名為 log4j.properties：

在這里插入圖片描述

log4j.properties

log4j.rootLogger=INFO, stdout 
log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n 
log4j.appender.logfile=org.apache.log4j.FileAppender 
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

register

我們創建一個名為register的package：

在這里插入圖片描述

創建注冊中心的注冊服務介面ServiceRegister.java：

package com.rpc.register;

import java.net.InetSocketAddress;

/**
 * @author weiyu_zeng
 *
 * 服務注冊介面，兩大基本功能，注冊：保存服務與地址， 查詢：根據服務名查找地址
 */
public interface ServiceRegister {

    void register(String serviceName, InetSocketAddress serverAddress);

    InetSocketAddress serviceDiscovery(String serviceName);
}

然后創建服務注冊實作類 ZkServiceRegister.java：

package com.rpc.register;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * @author weiyu_zeng
 *
 * Curator：是Zookeeper開源的客戶端框架，封裝了很多API，使用起來非常的方便
 * CuratorFramework：連接zookeeper服務的框架，客戶端創建使用靜態工廠方式CuratorFrameworkFactory進行創建
 * tickTime：zk的心跳間隔(heartbeat interval）,也是session timeout基本單位.單位為毫秒.
 * minSessionTimeout:最小超時時間,zk設定的默認值為2*tickTime.
 * maxSessionTimeout:最大超時時間,zk設定的默認值為20*tickTime.
 * retryPolicy()重連策略：
 * Curator 四種重連策略：
 *      1.RetryUntilElapsed(int maxElapsedTimeMs, int sleepMsBetweenRetries)
 *      以sleepMsBetweenRetries的間隔重連,直到超過maxElapsedTimeMs的時間設定
 *
 *      2.RetryNTimes(int n, int sleepMsBetweenRetries)
 *      指定重連次數
 *
 *      3.RetryOneTime(int sleepMsBetweenRetry)
 *      重連一次,簡單粗暴
 *
 *      4.ExponentialBackoffRetry
 *      ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries)
 *      ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)
 *      時間間隔 = baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1)))
 *
 * namespace(): 為了避免各個應用的zk patch沖突, Curator Framework內部會給每一個Curator Framework實體分配一個namespace(可選).
 *              這樣你在create ZNode的時候都會自動加上這個namespace作為這個node path的root.
 * CuratorFramework.create()：開始創建操作，可以呼叫額外的方法(比如方式mode 或者后臺執行background) 并在最后呼叫forPath()
 *                            指定要操作的ZNode
 * CuratorFramework.checkExists(): 開始檢查ZNode是否存在的操作. 可以呼叫額外的方法(監控或者后臺處理)并在最后呼叫forPath()
 *                                指定要操作的ZNode
 * CuratorFramework.start() / close()：啟動和關閉客戶端
 * CuratorFramework(client).create().withMode(CreateMode.EPHEMERAL)：這將使用給定的資料創建臨時結點 EPHEMERAL ZNode
 * CuratorFramework.getChildren()：開始獲得ZNode的子節點串列， 以呼叫額外的方法(監控、后臺處理或者獲取狀態watch,
 *                                 background or get stat)并在最后呼叫forPath()指定要操作的ZNode
 *
 * InetSocketAddress：該類實作了可序列化介面，直接繼承自java.net.SocketAddress類，實作 IP 套接字地址（IP 地址 + 埠號），
 *                    它還可以是一個對（主機名 + 埠號），在此情況下，將嘗試決議主機名，如果決議失敗，則該地址將被視為未決議
 *                    地址，但是其在某些情形下仍然可以使用，比如通過代理連接，
 *                    構造方法：InetSocketAddress(InetAddress addr,  int port)  根據 IP 地址和埠號創建套接字地址，
 *                             InetSocketAddress(String hostname, int port) 根據主機名（IP地址指代）和埠號創建套接字地址，
 * InetSocketAddress.getHostName()：獲取 hostname，即地址的主機名部分，
 * InetSocketAddress.getPort()  獲取埠號，
 */
public class ZkServiceRegister implements ServiceRegister {

    // curator 提供的zookeeper客戶端
    private CuratorFramework client;

    // zookeeper根路徑結點
    private static final String ROOT_PATH = "MyRPC";

    // 構造方法
    // 這里負責zookeeper客戶端的初始化，并與zookeeper服務端建立連接，
    // 初始化包括指定重連策略，指定連接zookeeper的埠，指定超時時間，指定命名空間
    // 初始化完成之后start()開啟zookeeper客戶端，
    public ZkServiceRegister() {

        // 重連策略：指數時間重試
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 3);

        // zookeeper的地址固定，不管是服務提供者還是消費者，都要與之建立連接
        // sessionTimeoutMs 與 zoo.cfg中的tickTime 有關系，
        // zk還會根據minSessionTimeout與maxSessionTimeout兩個引數重新調整最后的超時值，默認分別為tickTime 的2倍和20倍
        // 使用心跳監聽狀態
        this.client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181")
                                             .sessionTimeoutMs(40000)
                                             .retryPolicy(policy)
                                             .namespace(ROOT_PATH)
                                             .build();
        this.client.start();
        System.out.println("zookeeper 連接成功");
    }

    // 注冊：傳入服務方法名(String)，傳入主機名和埠號的套接字地址(InetSocketAddress)
    @Override
    public void register(String serviceName, InetSocketAddress serverAddress) {
        try {
            // serviceName創建成永久節點，服務提供者下線時，不刪服務名，只刪地址
            Stat stat = client.checkExists().forPath("/" + serviceName);
            if (stat == null) {
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath("/" + serviceName);
            }
            // 路徑地址，一個/代表一個節點
            String path = "/" + serviceName + "/" + getServiceAddress(serverAddress);
            // 臨時節點，服務器下線就洗掉節點
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);
        } catch (Exception e) {
            System.out.println("此服務已存在");
        }
    }

    // 根據服務名回傳地址
    @Override
    public InetSocketAddress serviceDiscovery(String serviceName) {
        try {
            List<String> strings = client.getChildren().forPath("/" + serviceName);
            // 這里默認用的第一個，后面加負載均衡
            String string = strings.get(0);
            return parseAddress(string);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    // 地址 -> XXX.XXX.XXX.XXX:port 字串
    private String getServiceAddress(InetSocketAddress serverAddress) {
        return serverAddress.getHostName() + ":" + serverAddress.getPort();
    }

    // 字串決議為地址：按照":"切分開，前半是host(String)，后半是port(int)
    private InetSocketAddress parseAddress(String address) {
        String[] result = address.split(":");
        return new InetSocketAddress(result[0], Integer.parseInt(result[1]));
    }
}

接下來可以對service，client和server進行修改，

client

NettyRPCClient.java 做一點修改：

package com.rpc.client;

import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import com.rpc.register.ServiceRegister;
import com.rpc.register.ZkServiceRegister;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;

import java.net.InetSocketAddress;

/**
 * @author zwy
 *
 * 實作RPCClient介面
 */
public class NettyRPCClient implements RPCClient {
    private static final Bootstrap bootstrap;
    private static final EventLoopGroup eventLoopGroup;
    private String host;
    private int port;
    private ServiceRegister serviceRegister; // ServiceRegister介面類class

    // 建構式：初始化zookeeper
    public NettyRPCClient() {
        this.serviceRegister = new ZkServiceRegister();
    }
    // netty客戶端初始化，重復使用
    static {
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                .handler(new NettyClientInitializer());
    }

    /**
     * 這里需要操作一下，因為netty的傳輸都是異步的，你發送request，會立刻回傳一個值， 而不是想要的相應的response
     */
    @Override
    public RPCResponse sendRequest(RPCRequest request) {
        InetSocketAddress address = serviceRegister.serviceDiscovery(request.getInterfaceName());
        host = address.getHostName();
        port = address.getPort();
        try {
            ChannelFuture channelFuture  = bootstrap.connect(host, port).sync();
            Channel channel = channelFuture.channel();
            // 發送資料
            channel.writeAndFlush(request);
            channel.closeFuture().sync();
            // 阻塞的獲得結果，通過給channel設計別名，獲取特定名字下的channel中的內容（這個在hanlder中設定）
            // AttributeKey是，執行緒隔離的，不會由執行緒安全問題，
            // 實際上不應通過阻塞，可通過回呼函式，后面可以再進行優化
            AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
            RPCResponse response = channel.attr(key).get();

            System.out.println(response);
            return response;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
}

TestClient.java 也做相應的修改：

package com.rpc.client;

import com.rpc.common.Blog;
import com.rpc.common.User;
import com.rpc.service.BlogService;
import com.rpc.service.UserService;


/**
 * @author zwy
 */
public class TestClient {
    public static void main(String[] args) {
        // 不需傳host，port
        RPCClient rpcClient = new NettyRPCClient();
        // 把這個客戶端傳入代理客戶端
        RPCClientProxy rpcClientProxy = new RPCClientProxy(rpcClient);
        // 代理客戶端根據不同的服務，獲得一個代理類， 并且這個代理類的方法以或者增強（封裝資料，發送請求）
        UserService userService = rpcClientProxy.getProxy(UserService.class);

        // 服務的方法1
        User userByUserId = userService.getUserByUserId(10);
        System.out.println("從服務器端得到的user為：" + userByUserId);

        // 服務的方法2
        User user = User.builder().userName("張三").id(100).sex(true).build();
        Integer integer = userService.insertUserId(user);
        System.out.println("向服務器端插入資料" + integer);

        // 服務的方法3
        BlogService blogService = rpcClientProxy.getProxy(BlogService.class);
        Blog blogById = blogService.getBlogById(10000);
        System.out.println("從服務端得到的blog為：" + blogById);
    }
}

client中的其他代碼和simpleRPC-05一樣，可以直接從simpleRPC-05復制粘貼過來，為了完整，我還是把代碼放下面：

RPCClient.java

package com.rpc.client;


import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;

/**
 * @author zwy
 *
 * RPC客戶端：發送請求，獲得response
 */
public interface RPCClient {
    RPCResponse sendRequest(RPCRequest request);
}

RPCClientProxy.java

package com.rpc.client;

import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import lombok.AllArgsConstructor;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
 * @author zwy
 *
 * 客戶端代理：把動態代理封裝request物件（這里和simpleRPC-02的ClientProxy函式一樣，保留了動態代理的設計）
 */
@AllArgsConstructor
public class RPCClientProxy implements InvocationHandler {
    private RPCClient client;

    // jdk動態代理，每一次代理物件呼叫方法，會經過此方法增強（反射獲取request物件，socket發送至客戶端）
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // request的構建，使用了lombok中的builder，更加簡潔
        RPCRequest request = RPCRequest.builder().interfaceName(method.getDeclaringClass().getName())
                                       .methodName(method.getName())
                                       .params(args)
                                       .paramsTypes(method.getParameterTypes())
                                       .build();
        // 資料傳輸
        RPCResponse response = client.sendRequest(request);
//        System.out.println(response);
        return response.getData();
    }

    <T> T getProxy(Class<T> clazz) {
        Object o = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, this);
        return (T)o;
    }
}

NettyClientInitializer.java

package com.rpc.client;

import com.rpc.codec.JsonSerializer;
import com.rpc.codec.MyDecode;
import com.rpc.codec.MyEncode;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

/**
 * @author zwy
 *
 * 同樣的與服務端解碼和編碼格式
 */
public class NettyClientInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 使用自定義的編解碼器
        pipeline.addLast(new MyDecode());
        // 編碼需要傳入序列化器，這里是json，還支持ObjectSerializer，也可以自己實作其他的
        pipeline.addLast(new MyEncode(new JsonSerializer()));
        pipeline.addLast(new NettyClientHandler());
    }
}

NettyClientHandler.java

package com.rpc.client;

import com.rpc.common.RPCResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;

/**
 * @author zwy
 */
public class NettyClientHandler extends SimpleChannelInboundHandler<RPCResponse> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RPCResponse msg) throws Exception {
        // 接收到response, 給channel設計別名，讓sendRequest里讀取response
        AttributeKey<RPCResponse> key = AttributeKey.valueOf("RPCResponse");
        ctx.channel().attr(key).set(msg);
        ctx.channel().close();
    }

    // 跟NettyRPCServerHandler一樣
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

service

服務暴露類加入注冊的功能，ServiceProvider.java 做相應的修改：

package com.rpc.service;


import com.rpc.register.ServiceRegister;
import com.rpc.register.ZkServiceRegister;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;

/**
 * @author zwy
 */
public class ServiceProvider {
    /**
     * 一個實作類可能實作多個服務介面，
     */
    private Map<String, Object> interfaceProvider;

    private ServiceRegister serviceRegister;
    private String host;
    private int port;

    public ServiceProvider(String host, int port){
        // 需要傳入服務端自身的服務的網路地址
        this.host = host;
        this.port = port;
        this.interfaceProvider = new HashMap<>();
        this.serviceRegister = new ZkServiceRegister();
    }

    public void provideServiceInterface(Object service) throws Exception {
        Class<?>[] interfaces = service.getClass().getInterfaces();

        for(Class clazz : interfaces){
            // 本機的映射表
            interfaceProvider.put(clazz.getName(),service);
            // 在注冊中心注冊服務
            serviceRegister.register(clazz.getName(), new InetSocketAddress(host, port));
        }
    }

    public Object getService(String interfaceName){
        return interfaceProvider.get(interfaceName);
    }
}

service中的其他代碼和simpleRPC-05一樣，可以直接從simpleRPC-05復制粘貼過來，為了完整，我還是把代碼放下面：

BlogService.java

package com.rpc.service;

import com.rpc.common.Blog;

public interface BlogService {

    Blog getBlogById(Integer id);
}

BlogServiceImpl.java

package com.rpc.service;

import com.rpc.common.Blog;

public class BlogServiceImpl implements BlogService {

    @Override
    public Blog getBlogById(Integer id) {
        Blog blog = Blog.builder()
                        .id(id)
                        .title("我的博客")
                        .useId(22).build();
        System.out.println("客戶端查詢了" + id + "博客");
        return blog;
    }
}

UserService.java

package com.rpc.service;

import com.rpc.common.User;

/**
 * @author zwy
 */
public interface UserService {

    // 客戶端通過這個介面呼叫服務端的實作類
    User getUserByUserId(Integer id);

    // 給這個服務增加一個功能
    Integer insertUserId(User user);
}

UserServiceImpl.java

package com.rpc.service;

import com.rpc.common.User;


/**
 * @author zwy
 */
public class UserServiceImpl implements UserService {

    @Override
    public User getUserByUserId(Integer id) {
        // 模擬從資料庫中取用戶的行為
        User user = User.builder()
                        .id(id)
                        .userName("he2121")
                        .sex(true).build();
        System.out.println("客戶端查詢了" + id + "的用戶");
        return user;
    }

    @Override
    public Integer insertUserId(User user) {
        System.out.println("插入資料成功: " + user);
        return 1;
    }
}

server

TestServer.java 做相應的修改：

package com.rpc.server;

import com.rpc.service.*;

public class TestServer {
    public static void main(String[] args) throws Exception {
        UserService userService = new UserServiceImpl();
        BlogService blogService = new BlogServiceImpl();

        // 這里重用了服務暴露類，順便在注冊中心注冊，實際上應分開，每個類做各自獨立的事
        ServiceProvider serviceProvider = new ServiceProvider("127.0.0.1", 8899); // 8899
        serviceProvider.provideServiceInterface(userService);
        serviceProvider.provideServiceInterface(blogService);

        RPCServer RPCServer = new NettyRPCServer(serviceProvider);
        RPCServer.start(8899);
    }
}

server中的其他代碼和simpleRPC-05一樣，可以直接從simpleRPC-05復制粘貼過來，為了完整，我還是把代碼放下面：

RPCServer.java

package com.rpc.server;

/**
 * @author zwy
 */
public interface RPCServer {
    void start(int port);
    void stop();
}

NettyRPCServer.java

package com.rpc.server;

import com.rpc.service.ServiceProvider;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.AllArgsConstructor;


/**
 * @author zwy
 */
@AllArgsConstructor
public class NettyRPCServer implements RPCServer {
    private ServiceProvider serviceProvider;

    @Override
    public void start(int port) {
        // netty服務執行緒組負責建立連接(TCP/IP連接)，work負責具體的請求
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        System.out.println("Netty服務端啟動了");

        try {
            // 啟動Netty服務器
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 初始化
            serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
                           .childHandler(new NettyServerInitializer(serviceProvider));
            // 同步阻塞
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            // 死回圈監聽
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

    @Override
    public void stop() {

    }
}

NettyServerInitializer.java

package com.rpc.server;

import com.rpc.codec.JsonSerializer;
import com.rpc.codec.MyDecode;
import com.rpc.codec.MyEncode;
import com.rpc.service.ServiceProvider;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import lombok.AllArgsConstructor;

/**
 * @author zwy
 */
@AllArgsConstructor
public class NettyServerInitializer extends ChannelInitializer<SocketChannel> {
    private ServiceProvider serviceProvider;

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 使用自定義的解碼器
        pipeline.addLast(new MyDecode());
        // 使用自定義的編碼器，而且解碼器需要傳入序列化器，這里是json，還支持ObjectSerializer，也可以自己實作其他的
        pipeline.addLast(new MyEncode(new JsonSerializer()));
        pipeline.addLast(new NettyRPCServerHandler(serviceProvider));
    }
}

NettyRPCServerHandler.java

package com.rpc.server;

import com.rpc.common.RPCRequest;
import com.rpc.common.RPCResponse;
import com.rpc.service.ServiceProvider;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.AllArgsConstructor;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

/**
 * @author zwy
 */
@AllArgsConstructor
public class NettyRPCServerHandler extends SimpleChannelInboundHandler<RPCRequest> {
    private ServiceProvider serviceProvider;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RPCRequest msg) throws Exception {
        // System.out.println(msg);
        RPCResponse response = getResponse(msg);
        ctx.writeAndFlush(response);
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    // 這里和WorkThread里的getResponse差不多
    RPCResponse getResponse(RPCRequest request) {
        // 得到服務名
        String interfaceName = request.getInterfaceName();
        // 得到服務器相應類
        Object service = serviceProvider.getService(interfaceName);
        // 反射呼叫方法
        Method method = null;
        try {
            method = service.getClass().getMethod(request.getMethodName(), request.getParamsTypes());
            Object invoke = method.invoke(service, request.getParams());
            return RPCResponse.success(invoke);
        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
            e.printStackTrace();
            System.out.println("方法執行錯誤");
            return RPCResponse.fail();
        }
    }
}

common

和simpleRPC-05一樣，可以直接復制過來，

在這里插入圖片描述

codec

和simpleRPC-05一樣，可以直接復制過來，

在這里插入圖片描述

檔案結構

simpleRPC-06的檔案結構如下：

在這里插入圖片描述

在這里插入圖片描述

運行

啟動TestServer.java ：
在這里插入圖片描述
然后啟動TestClient.java：

在這里插入圖片描述
在這里插入圖片描述

我們來看看我們最開始開的zookeeper客戶端：

現在輸入ls /

在這里插入圖片描述

發現我們多了一個結點 MyRPC：

輸入ls /MyRPC：

在這里插入圖片描述
可以看到我們注冊的服務都在這里，成功！

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/402694.html

標籤：其他

上一篇：【無標題】一定要走，走到燈火通明

下一篇：如何在IDEA編譯器中連接HDFS，運行MapReduce程式

標籤雲
其他(157675) Python(38076) JavaScript(25376) Java(17977) C(15215) 區塊鏈(8255) C＃(7972) AI(7469) 爪哇(7425) MySQL(7132) html(6777) 基礎類(6313) sql(6102) 熊猫(6058) PHP(5869) 数组(5741) R(5409) Linux(5327) 反应(5209) 腳本語言(PerlPython)(5129) 非技術區(4971) Android(4554) 数据框(4311) css(4259) 节点.js(4032) C語言(3288) json(3245) 列表(3129) 扑(3119) C++語言(3117) 安卓(2998) 打字稿(2995) VBA(2789) Java相關(2746) 疑難問題(2699) 细绳(2522) 單片機工控(2479) iOS(2429) ASP.NET(2402) MongoDB(2323) 麻木的(2285) 正则表达式(2254) 字典(2211) 循环(2198) 迅速(2185) 擅长(2169) 镖(2155) 功能(1967) .NET技术(1958) Web開發(1951) python-3.x(1918) HtmlCss(1915) 弹簧靴(1913) C++(1909) xml(1889) PostgreSQL(1872) .NETCore(1853) 谷歌表格(1846) Unity3D(1843) for循环(1842)

熱門瀏覽
  • 網閘典型架構簡述

    網閘架構一般分為兩種：三主機的三系統架構網閘和雙主機的2+1架構網閘。 三主機架構分別為內端機、外端機和仲裁機。三機無論從軟體和硬體上均各自獨立。首先從硬體上來看，三機都用各自獨立的主板、記憶體及存盤設備。從軟體上來看，三機有各自獨立的作業系統。這樣能達到完全的三機獨立。對于“2+1”系統，“2”分為 ......

    uj5u.com 2020-09-10 02:00:44 more
  • 如何從xshell上傳檔案到centos linux虛擬機里

    如何從xshell上傳檔案到centos linux虛擬機里及：虛擬機CentOs下執行 yum -y install lrzsz命令，出現錯誤：鏡像無法找到軟體包 前言 一、安裝lrzsz步驟 二、上傳檔案 三、遇到的問題及解決方案 總結 前言 提示：其實很簡單，往虛擬機上安裝一個上傳檔案的工具 ......

    uj5u.com 2020-09-10 02:00:47 more
  • 一、SQLMAP入門

    一、SQLMAP入門 1、判斷是否存在注入 sqlmap.py -u 網址/id=1 id=1不可缺少。當注入點后面的引數大于兩個時。需要加雙引號， sqlmap.py -u "網址/id=1&uid=1" 2、判斷文本中的請求是否存在注入 從文本中加載http請求，SQLMAP可以從一個文本檔案中 ......

    uj5u.com 2020-09-10 02:00:50 more
  • Metasploit 簡單使用教程

    metasploit 簡單使用教程 浩先生， 2020-08-28 16:18:25 分類專欄： kail 網路安全 linux 文章標簽： linux資訊安全 編輯 著作權 metasploit 使用教程 前言 一、Metasploit是什么？ 二、準備作業 三、具體步驟 前言 Msfconsole ......

    uj5u.com 2020-09-10 02:00:53 more
  • 游戲逆向之驅動層與用戶層通訊

    驅動層代碼： #pragma once #include <ntifs.h> #define add_code CTL_CODE(FILE_DEVICE_UNKNOWN,0x800,METHOD_BUFFERED,FILE_ANY_ACCESS) /* 更多游戲逆向視頻www.yxfzedu.com ......

    uj5u.com 2020-09-10 02:00:56 more
  • 北斗電力時鐘（北斗授時服務器）讓網路資料更精準

    北斗電力時鐘（北斗授時服務器）讓網路資料更精準 北斗電力時鐘（北斗授時服務器）讓網路資料更精準 京準電子科技官微——ahjzsz 近幾年，資訊技術的得了快速發展，互聯網在逐漸普及，其在人們生活和生產中都得到了廣泛應用，并且取得了不錯的應用效果。計算機網路資訊在電力系統中的應用，一方面使電力系統的運行 ......

    uj5u.com 2020-09-10 02:01:03 more
  • 【CTF】CTFHub 技能樹 彩蛋 writeup

    ?碎碎念 CTFHub：https://www.ctfhub.com/ 筆者入門CTF時時剛開始刷的是bugku的舊平臺，后來才有了CTFHub。 感覺不論是網頁UI設計，還是題目質量，賽事跟蹤，工具軟體都做得很不錯。 而且因為獨到的金幣制度的確讓人有一種想去刷題賺金幣的感覺。 個人還是非常喜歡這個 ......

    uj5u.com 2020-09-10 02:04:05 more
  • 02windows基礎操作

    我學到了一下幾點 Windows系統目錄結構與滲透的作用 常見Windows的服務詳解 Windows埠詳解 常用的Windows注冊表詳解 hacker DOS命令詳解（net user / type /md /rd/ dir /cd /net use copy、批處理 等） 利用dos命令制作 ......

    uj5u.com 2020-09-10 02:04:18 more
  • 03.Linux基礎操作

    我學到了以下幾點 01Linux系統介紹02系統安裝，密碼啊破解03Linux常用命令04LAMP 01LINUX windows： win03 8 12 16 19 配置不繁瑣 Linux：redhat,centos(紅帽社區版)，Ubuntu server,suse unix:金融機構，證券，銀 ......

    uj5u.com 2020-09-10 02:04:30 more
  • 05HTML

    01HTML介紹 02頭部標簽講解03基礎標簽講解04表單標簽講解 HTML前段語言 js1.了解代碼2.根據代碼 懂得挖掘漏洞 （POST注入/XSS漏洞上傳）3.黑帽seo 白帽seo 客戶網站被黑帽植入劫持代碼如何處理4.熟悉html表單 <html><head><title>TDK標題，描述 ......

    uj5u.com 2020-09-10 02:04:36 more
最新发布
  • 2023年最新微信小程式抓包教程

    01 開門見山 隔一個月發一篇文章，不過分。 首先回顧一下《微信系結手機號資料庫被脫庫事件》，我也是第一時間得知了這個訊息，然后跟蹤了整件事情的經過。下面是這起事件的相關截圖以及近日流出的一萬條資料樣本： 個人認為這件事也沒什么，還不如關注一下之前45億快遞資料查詢渠道疑似在近日復活的訊息。 訊息是 ......

    uj5u.com 2023-04-20 08:48:24 more
  • web3 產品介紹：metamask 錢包 使用最多的瀏覽器插件錢包

    Metamask錢包是一種基于區塊鏈技術的數字貨幣錢包，它允許用戶在安全、便捷的環境下管理自己的加密資產。Metamask錢包是以太坊生態系統中最流行的錢包之一，它具有易于使用、安全性高和功能強大等優點。 本文將詳細介紹Metamask錢包的功能和使用方法。 一、 Metamask錢包的功能 數字資 ......

    uj5u.com 2023-04-20 08:47:46 more
  • vulnhub_Earth

    前言 靶機地址->>>vulnhub_Earth 攻擊機ip：192.168.20.121 靶機ip：192.168.20.122 參考文章 https://www.cnblogs.com/Jing-X/archive/2022/04/03/16097695.html https://www.cnb ......

    uj5u.com 2023-04-20 07:46:20 more
  • 從4k到42k，軟體測驗工程師的漲薪史，給我看哭了

    清明節一過，盲猜大家已經無心上班，在數著日子準備過五一，但一想到銀行卡里的余額……瞬間心情就不美麗了。最近，2023年高校畢業生就業調查顯示，本科畢業月平均起薪為5825元。調查一出，便有很多同學表示自己又被平均了。看著這一資料，不免讓人想到前不久中國青年報的一項調查：近六成大學生認為畢業10年內會 ......

    uj5u.com 2023-04-20 07:44:00 more
  • 最新版本 Stable Diffusion 開源 AI 繪畫工具之中文自動提詞篇

    🎈 標簽生成器 由于輸入正向提示詞 prompt 和反向提示詞 negative prompt 都是使用英文，所以對學習母語的我們非常不友好 使用網址：https://tinygeeker.github.io/p/ai-prompt-generator 這個網址是為了讓大家在使用 AI 繪畫的時候 ......

    uj5u.com 2023-04-20 07:43:36 more
  • 漫談前端自動化測驗演進之路及測驗工具分析

    隨著前端技術的不斷發展和應用程式的日益復雜，前端自動化測驗也在不斷演進。隨著 Web 應用程式變得越來越復雜，自動化測驗的需求也越來越高。如今，自動化測驗已經成為 Web 應用程式開發程序中不可或缺的一部分，它們可以幫助開發人員更快地發現和修復錯誤，提高應用程式的性能和可靠性。 ......

    uj5u.com 2023-04-20 07:43:16 more
  • CANN開發實踐：4個DVPP記憶體問題的典型案例解讀

    摘要：由于DVPP媒體資料處理功能對存放輸入、輸出資料的記憶體有更高的要求（例如，記憶體首地址128位元組對齊），因此需呼叫專用的記憶體申請介面，那么本期就分享幾個關于DVPP記憶體問題的典型案例，并給出原因分析及解決方法。 本文分享自華為云社區《FAQ_DVPP記憶體問題案例》，作者：昇騰CANN。 DVPP ......

    uj5u.com 2023-04-20 07:43:03 more
  • msf學習

    msf學習 以kali自帶的msf為例 一、msf核心模塊與功能 msf模塊都放在/usr/share/metasploit-framework/modules目錄下 1、auxiliary 輔助模塊，輔助滲透（埠掃描、登錄密碼爆破、漏洞驗證等） 2、encoders 編碼器模塊，主要包含各種編碼 ......

    uj5u.com 2023-04-20 07:42:59 more
  • Halcon軟體安裝與界面簡介

    1. 下載Halcon17版本到到本地 2. 雙擊安裝包后 3. 步驟如下 1.2 Halcon軟體安裝 界面分為四大塊 1. Halcon的五個助手 1) 影像采集助手：與相機連接，設定相機引數，采集影像 2) 標定助手：九點標定或是其它的標定，生成標定檔案及內參外參，可以將像素單位轉換為長度單位 ......

    uj5u.com 2023-04-20 07:42:17 more
  • 在MacOS下使用Unity3D開發游戲

    第一次發博客，先發一下我的游戲開發環境吧。 去年2月份買了一臺MacBookPro2021 M1pro(以下簡稱mbp)，這一年來一直在用mbp開發游戲。我大致分享一下我的開發工具以及使用體驗。 1、Unity 官網鏈接： https://unity.cn/releases 我一般使用的Apple ......

    uj5u.com 2023-04-20 07:40:19 more

Copyright © 2010-2020 有解無憂