基於Socket與Netty的RPC實現

阿新 • • 發佈：2019-02-09

前言

RPC的大多數實現方式包括兩類：Socket與Netty，接下來介紹兩種方式基於java的簡單實現。

RPC架構分為三部分：

1）服務提供者，執行在伺服器端，提供服務介面定義與服務實現類。

2）服務中心，執行在伺服器端，負責將本地服務釋出成遠端服務，管理遠端服務，提供給服務消費者使用。

3）服務消費者，執行在客戶端，通過遠端代理物件呼叫遠端服務。

1 Socket

RPC的socket實現是基於TCP/IP協議，它在通訊的兩端各建立一個Socket，從而在通訊的兩端之間形成網路虛擬鏈路。一旦建立了虛擬的網路鏈路，兩端的成員就可以通過虛擬鏈路進行通訊。Java是對基於TCP協議的網路通訊提供了良好的封裝，Java使用Socket物件來代表兩端的通訊埠，並通過Socket產生IO流來進行網路通訊，實現消費者與服務方的通訊。

1.1 Interface

首先介紹介面類中的介面方法：

SayHello.java

public interface SayHello {
    String sayHello(String name);
}

Server.java

import java.io.IOException;

public interface Server {
    void stop();

    void start() throws IOException;

    void register(Class serviceInterface, Class impl);

    boolean isRunning();

    int getPort();
}

1.2 Provider

服務端定義介面的具體實現，包括被呼叫方法以及服務中心實現。

SayHelloImpl.java

/**
 * @author jhz
 * @date 18-8-24 下午8:56
 */
public class SayHelloImpl implements SayHello{
    @Override
    public String sayHello(String name) {
        return name;
    }
}

ServerImpl.java

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author jhz
 * @date 18-8-24 下午10:00
 */
public class ServerImpl implements Server {
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static final HashMap<String, Class> serviceRegistry = new HashMap<String, Class>();

    private static boolean isRunning = false;

    private static int port;

    public ServerImpl(int port) {
        this.port = port;
    }

    public void stop() {
        isRunning = false;
        executor.shutdown();
    }

    public void start() throws IOException {
        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress(port));
        System.out.println("start server");
        try {
            while (true) {
                // 1.監聽客戶端的TCP連線，接到TCP連線後將其封裝成task，由執行緒池執行
                executor.execute(new ServiceTask(server.accept()));
            }
        } finally {
            server.close();
        }
    }

    public void register(Class serviceInterface, Class impl) {
        serviceRegistry.put(serviceInterface.getName(), impl);
    }

    public boolean isRunning() {
        return isRunning;
    }

    public int getPort() {
        return port;
    }

    private static class ServiceTask implements Runnable {
        Socket clent = null;

        public ServiceTask(Socket client) {
            this.clent = client;
        }

        public void run() {
            ObjectInputStream input = null;
            ObjectOutputStream output = null;
            try {
                // 2.將客戶端傳送的碼流反序列化成物件，反射呼叫服務實現者，獲取執行結果
                input = new ObjectInputStream(clent.getInputStream());
                String serviceName = input.readUTF();
                String methodName = input.readUTF();
                Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();
                Class serviceClass = serviceRegistry.get(serviceName);
                if (serviceClass == null) {
                    throw new ClassNotFoundException(serviceName + " not found");
                }
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);

                // 3.將執行結果反序列化，通過socket傳送給客戶端
                output = new ObjectOutputStream(clent.getOutputStream());
                output.writeObject(result);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (output != null) {
                    try {
                        output.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (input != null) {
                    try {
                        input.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (clent != null) {
                    try {
                        clent.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

        }
    }
}

ProviderRunner.java(服務端啟動)

import java.io.IOException;

/**
 * @author jhz
 * @date 18-8-24 下午10:40
 */
public class ProviderRunner {
    public static void main(String[] args) {
        new Thread(new Runnable() {
            public void run() {
                try {
                    Server serviceServer = new ServerImpl(8088);
                    serviceServer.register(SayHello.class, SayHelloImpl.class);
                    serviceServer.start();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

1.3 Consumer

RPCClient.java(客戶端代理類)

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;

/**
 * @author jhz
 * @date 18-8-24 下午10:10
 */
public class RPCClient {
    public static <T> T getRemoteProxyObj(final Class<?> serviceInterface, final InetSocketAddress addr) {
        // 1.將本地的介面呼叫轉換成JDK的動態代理，在動態代理中實現介面的遠端呼叫
        return (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class<?>[]{serviceInterface},
                new InvocationHandler() {
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        Socket socket = null;
                        ObjectOutputStream output = null;
                        ObjectInputStream input = null;
                        try {
                            // 2.建立Socket客戶端，根據指定地址連線遠端服務提供者
                            socket = new Socket();
                            socket.connect(addr);

                            // 3.將遠端服務呼叫所需的介面類、方法名、引數列表等編碼後傳送給服務提供者
                            output = new ObjectOutputStream(socket.getOutputStream());
                            output.writeUTF(serviceInterface.getName());
                            output.writeUTF(method.getName());
                            output.writeObject(method.getParameterTypes());
                            output.writeObject(args);

                            // 4.同步阻塞等待伺服器返回應答，獲取應答後返回
                            input = new ObjectInputStream(socket.getInputStream());
                            return input.readObject();
                        } finally {
                            if (socket != null) socket.close();
                            if (output != null) output.close();
                            if (input != null) input.close();
                        }
                    }
                });
    }
}

可以看到遠端的呼叫是在動態處理器中的invoke()方法（匿名類中的invoke）中實現的，且是同步阻塞的實現方式，當客戶端發起呼叫時，在invoke()中發起建立socket連線，等待服務端返回呼叫結果。

SocketRunner.java(呼叫類)

import java.io.IOException;
import java.net.InetSocketAddress;

/**
 * @author jhz
 * @date 18-8-24 下午10:19
 */
public class SocketRunner {
    public static void main(String[] args) throws IOException {
        SayHello service = RPCClient.getRemoteProxyObj(SayHello.class, new InetSocketAddress("localhost", 8088));
        System.out.println(service.sayHello("hello jhz!"));
    }

}

執行結果

首先啟動服務端

再啟動客戶端：

2 Netty

接著我們使用Netty來代替這種BIO的實現：

2.1 Interface

介面類只需要定義介面方法（SayHello），服務端中的實現方法也不用變

2.2 Provider

NettyServer.java（Netty服務端）

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
/**
 * @author jhz
 * @date 18-8-24 下午10:48
 */
public class NettyServer {
    public static void startServer(String hostName, int port) {
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
            bootstrap.group(eventLoopGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new StringDecoder());
                            p.addLast(new StringEncoder());
                            p.addLast(new HelloServerHandler());
                        }
                    });
            bootstrap.bind(hostName, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在NettyServer中定義ServerBootStrap,並新增自定義的handler，具體的實現如下：

HelloServerHandler.java

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * @author jhz
 * @date 18-8-26 上午12:40
 */
public class HelloServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {

        // 如何符合約定，則呼叫本地方法，返回資料
        if (msg.toString().startsWith("SayHello")) {
            String result = new SayHelloImpl()
                    .sayHello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
            ctx.writeAndFlush(result);
        }
    }
}

該類重寫channelRead以實現對channel中訊息的過濾以及具體方法的呼叫。

NettyServerStarter.java

/**
 * @author jhz
 * @date 18-8-26 上午12:23
 */
public class NettyServerStarter {
    public static void main(String[] args) {
        NettyServer.startServer("localhost", 8088);
        System.out.println("Netty Server Start!");
    }
}

2.2 Consumer

同樣，我們需要建立客戶端代理類

RPCConsumer.java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author jhz
 * @date 18-8-26 上午12:19
 */
public class RPCConsumer {
    private static ExecutorService executor = Executors
            .newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static SayHelloHandler client;

    /**
     * 建立一個代理物件
     */
    public Object createProxy(final Class<?> serviceClass,
                              final String providerName) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{serviceClass}, (proxy, method, args) -> {
                    if (client == null) {
                        initClient();
                    }
                    // 設定引數
                    client.setPara(providerName + args[0]);
                    return executor.submit(client).get();
                });
    }

    /**
     * 初始化客戶端
     */
    private static void initClient() {
        client = new SayHelloHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new StringDecoder());
                        p.addLast(new StringEncoder());
                        p.addLast(client);
                    }
                });
        try {
            b.connect("localhost", 8088).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

在代理中實現客戶端的初始化，定義ClientBootStrap並新增客戶端的自定義handler(SayHelloHandler)，這個handler同時還能在收到回覆後完成callback，SayHelloHandler的具體實現如下。

ClientHandler.java

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.Callable;

public class SayHelloHandler extends ChannelInboundHandlerAdapter implements Callable {

    private ChannelHandlerContext context;
    private String result;
    private String para;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        context = ctx;
    }

    /**
     * 收到服務端資料，喚醒等待執行緒
     */
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
        result = msg.toString();
        notify();
    }

    /**
     * 寫出資料，開始等待喚醒
     */
    @Override
    public synchronized Object call() throws InterruptedException {
        context.writeAndFlush(para);
        wait();
        return result;
    }

    void setPara(String para) {
        this.para = para;
    }
}

定義了ChannelHandlerContext context：快取ctx，可以用於下次使用，整個流程如下，客戶端發起呼叫->呼叫call()方法，將請求訊息寫入channel並等待喚醒->收到服務端返回的訊息呼叫channelRead()並喚醒call上等待的執行緒，返回結果。

ClientBootStrap.java(客戶端測試類)

/**
 * @author jhz
 * @date 18-8-26 上午12:12
 */
public class ClientBootStrap {
    public static final String providerName = "SayHello#sayHello#";

    public static void main(String[] args) throws InterruptedException {

        RPCConsumer consumer = new RPCConsumer();
        // 建立一個代理物件
        SayHello service = (SayHello) consumer
                .createProxy(SayHello.class, providerName);
        System.out.println(service.sayHello("hello jhz"));
    }
}

測試結果

啟動Netty服務端：

客戶端呼叫：

採用Netty實現的RPC是執行緒級別的阻塞，並不會影響到其他的執行緒。