居于Netty实现的一个简单的RPC

Netty：基于java NIO 网络通信框架，具有高效、简单、快速的应用特点。在当下互联网高并发场景下得到很好地应用，现在用java写的高并发产品(如dubbo 、zookeeper、hadoop、rocketmq)大都应用了netty作为底层的通信技术。

RPC：远程调用，通过RPC框架，使得我们可以像调用本地方法一样地调用远程机器上的方法。

1.定义常量类，包含主机地址，端口号。

/**
 * @author hsl 2020-07-02 13:26
 * 一个常亮工具类 主机地址，端口号
 */
public class Constants {
    public static String REMOTE_HOST = "localhost";
    public static int PORT = 6789;
}

2.RPC框架，使得我们可以像调用本地方法一样地调用远程机器上的方法，那么我们就定义接口并实现。

package com.netty.nettyOne.service.dateTimeService;

/**
 * @author hsl 2020-07-02 13:29
 */
public interface DateTimeService {
    public String getNow(String param);
}

package com.netty.nettyOne.service.dateTimeService;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author hsl 2020-07-02 13:29
 */
public class DateTimeServiceImpl implements DateTimeService {

    @Override
    public String getNow(String param) {
        return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date());
    }
}

3.NettyServer

package com.netty.nettyOne.server;

import com.netty.nettyOne.service.dateTimeService.DateTimeServiceImpl;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * @author hsl 2020-07-02 14:00
 * 服务处理类实现
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //System.out.println("Server has received Message: [" + msg + "]");
        //处理客户端发送来的信息
        System.out.println("客户端发送的信息：" + msg.toString());
        //响应
        String result = new DateTimeServiceImpl().getNow(msg.toString());
        System.out.println("服务器响应的信息：" + result);
        ctx.writeAndFlush(result);
    }

//    @Override
//    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
//
//    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();    //抛出关闭通信channel
    }
}

package com.netty.nettyOne.server;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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 hsl 2020-07-02 13:39
 */
public class NettyServer {
    //主机名称 端口号
    private String host;
    private int port;

    public NettyServer(String host, int port) {
        this.host = host;
        this.port = port;
    }

    /**
     * 对外公开的方法
     */
    public void serverStart() {
        serverStart0(host, port);
    }

    /**
     * @param host 主机名称
     * @param port     端口号
     */
    private void serverStart0(String host, int port) {
        //处理 ACCEPT 事件的线程工作组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);

        //处理 READ/WRITER 时间的线程工作组
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());

        ServerBootstrap server = new ServerBootstrap();

        server.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new StringEncoder())
                                .addLast(new StringDecoder())
                                .addLast(new NettyServerHandler());  // 此处添加业务处理Handler
                    }
                });

        try {
            //启动服务
            ChannelFuture channelFuture = server.bind(host, port).sync();
            System.out.println("NettyServer is started ..... ");

            //关闭服务
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

netty服务器启动类

package com.netty.nettyOne.server;

import com.netty.nettyOne.util.Constants;

/**
 * @author hsl 2020-07-02 14:29
 */
public class NettyServerBootstrap {
    public static void main(String[] args) {
        //初始化服务器
        NettyServer nettyServer = new NettyServer(Constants.REMOTE_HOST, Constants.PORT);
        nettyServer.serverStart();
    }
}

4.NettyClient

package com.netty.nettyOne.client;

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

import java.util.concurrent.Callable;

/**
 * @author hsl 2020-07-02 14:48
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {
    //上下文对象
    /**
     * {@link #channelActive}
     */
    private ChannelHandlerContext context;
    //存放服务器返回结果
    private String result;
    //存放请求参数
    private String requestParam;

    public void setRequestParam(String requestParam) {
        this.requestParam = requestParam;
    }

    @Override
    public synchronized Object call() throws Exception {
        System.out.println("NettyClientHandler call() Invoacated ... ");
        //发送请求
        context.writeAndFlush(requestParam);
        //发送完了请求之后，等待....
        wait();
        //被唤醒之后，继续处理（返回结果即可）
        return result;
    }

    /**
     * @param ctx
     * @throws Exception
     * 当 该 NettyClientHandler 被初始化的时候，就会调用 该 channelActive 方法一次
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(" channelActive 被调用  ");
        // 向服务端发送请求。
        this.context = ctx;
    }

    /**
     * @param ctx
     * @param msg
     * @throws Exception
     * channelRead 用来处理服务端返回来的数据。
     */
    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        result = msg.toString();
        System.out.println("Client received result: " + result);

        //唤醒线程
        notify();
    }

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

package com.netty.nettyOne.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
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.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author hsl 2020-07-02 14:43
 */
public class NettyClient {
    private String host;
    private int port;

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
        //启动nettyCLient
        clientStart();
    }

    /**
     * 初始化一个线程池，用来处理服务端返回的结果数据
     */
    private static int cpu_cores = Runtime.getRuntime().availableProcessors();
    private static ExecutorService threadPool = Executors.newFixedThreadPool(cpu_cores);

    private NettyClientHandler clientHandler;

    /**
     * 客户端的启动方法
     */
    private void clientStart() {
        clientStart0(host, port);
    }

    private void clientStart0(String host, int port) {
        //服务工作组
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        //客户端服务启动类
        Bootstrap client = new Bootstrap();
        clientHandler = new NettyClientHandler();

        //绑定服务参数
        client.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new StringEncoder())
                                .addLast(new StringDecoder())
                                .addLast(clientHandler);
                    }
                });

        try {
            //客户端启动
            client.connect(host, port).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * NettyClient 端提供一个 代理方法，获取服务类的一个代理类
     *
     * @param serviceClass
     */
    public Object getProxy(final 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 {
                        System.out.println("客户端发送的请求参数是：" + args[0].toString());
                        clientHandler.setRequestParam(args[0].toString());
                        return threadPool.submit(clientHandler).get();
                    }
                });
    }
}

package com.netty.nettyOne.client;

import com.netty.nettyOne.service.dateTimeService.DateTimeService;
import com.netty.nettyOne.util.Constants;

/**
 * @author hsl 2020-07-02 15:30
 */
public class NettyClientBootstrap {
    public static void main(String[] args) {
        //启动客户端
        NettyClient client = new NettyClient(Constants.REMOTE_HOST, Constants.PORT);

        //获取一个服务端 服务对象的 代理对象
        DateTimeService dateTimeService = (DateTimeService) client.getProxy(DateTimeService.class);
        //调用服务
        String result = dateTimeService.getNow("Netty牛逼");
        // 输出结果
        System.out.println("Client Received Result From Server: [" + result + "]");
    }
}

6.实现结果

服务器打印

客户端打印