1、rpc基本介绍
RPC ( Remote Procedure Call) -远程过程调用，是一个计算机通信协议。该协议允许运行于一台计算机的程序调
用另一台计算机的子程序，两个或多个应用程序分布不同的服务器上，而他们之间的调用就像调用本地方法一样
常见的 RPC框架有:比较知名的如阿里的Dubbo、google的gRPC、Go语言的rpex、Apache的thrift，Spring的Spring Cloud。
2、rpc流程
1)服务消费方(client)以本地调用方式调用服务
2)clientstub接收到调用后负责将方法、参数等封装成能够进行网络传输的消息体
3)clientstub将消息进行编码并发送到服务端
4) server stub收到消息后进行解码
5)serverstub根据解码结果调用本地的服务
6)本地服务执行并将结果返回给server stub
7) server stub将返回导入结果进行编码并发送至消费方
8)clientstub接收到消息并进行解码
9)服 务消费方(client)得到结果
3、用Netty 实现一个简单的RPC框架
需求：消费者远程调用提供者的服务，提供者返回一个字符串，消费者接受提供者返回的数据
1》创建一个公共接口
2》创建一个提供者
3》创建一个消费者
示意图：
公共接口：
//公共接口，提供者和消费者都需要
public interface HelloService { public String hello(String msg); }
provider：服务端
实现类：
public class HelloImp implements HelloService { //消费方调用此方法时，返回值 @Override public String hello(String msg) { System.out.println("收到客户端信息："+msg); if(msg != null){ return "收到客户端的信息"+msg+"，我是服务端"; }else{ return "收到客户端的信息，我是服务端"; } } }
启动类：
public class MyServerBootstrap { public static void main(String[] args) {
//启动服务提供者 new MyNettyServer().startServer("127.0.0.1",7000); } }
public class MyNettyServer { public void startServer(String host , int port){
startServer0(host,port); } //初始化并启动服务端 public void startServer0(String host, int port){ EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workGroup = new NioEventLoopGroup(); ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup,workGroup) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new StringDecoder()); pipeline.addLast(new StringEncoder()); pipeline.addLast(new MyServerRpcHandler()); } }); try { ChannelFuture channelFuture = serverBootstrap.bind(host, port).sync(); channelFuture.addListener((sc)->{ if(sc.isSuccess()){ System.out.println("服务器启动成功"); }else{ System.out.println("服务器启动失败"); } }); channelFuture.channel().closeFuture().sync(); } catch (InterruptedException e) { e.printStackTrace(); }finally { bossGroup.shutdownGracefully(); workGroup.shutdownGracefully(); } } }
业务处理handler:
public class MyServerRpcHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("收到的客户端信息："+msg); String message = msg.toString().substring(msg.toString().lastIndexOf("#")+1); String result = new HelloImp().hello(message); ctx.writeAndFlush(result); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println(cause.getMessage()); ctx.close(); } }
consumer:客户端
启动类：
public class MyClientBootStrap { private static final String providerName = "helloWorld#"; public static void main(String[] args) {
//创建消费者对象 MyNettyClient consumer = new MyNettyClient();
//创建代理对象 HelloService helloService = (HelloService)consumer.getBean(HelloService.class, providerName);
//通过代理对象调用方法 String res = helloService.hello("hello 你好 服务端"); System.out.println("从客户端返回的结果["+res+"]"); } }
public class MyNettyClient { private static MyClientRpcHandler clientRpcHandler;
//创建自定义的线程池 ExecutorService executorService = new ThreadPoolExecutor( 2, 5, 20, TimeUnit.SECONDS, new LinkedBlockingDeque<>(3), new ThreadPoolExecutor.CallerRunsPolicy()); //使用代理模式，创建一个代理对象 public Object getBean(final Class<?> serviceClass , final String providerName){ System.out.println("getBean。。。。被调用"); return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),new Class<?>[]{serviceClass},(proxy, method, args) -> { if(clientRpcHandler == null){ initClient(); } System.out.println("clientRpcHandler..."+clientRpcHandler); clientRpcHandler.setPara(providerName+args[0]); return executorService.submit(clientRpcHandler).get(); }); }; public void initClient(){ clientRpcHandler = new MyClientRpcHandler(); EventLoopGroup group = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY,true) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new StringDecoder()); pipeline.addLast(new StringEncoder()); pipeline.addLast(clientRpcHandler); } }); try { bootstrap.connect("127.0.0.1", 7000).sync(); } catch (InterruptedException e) { e.printStackTrace(); } } }
业务handler:
public class MyClientRpcHandler extends ChannelInboundHandlerAdapter implements Callable{ private ChannelHandlerContext context;//上下文对象 private String result;//返回的结果 private String para;//客户端传给服务端的信息
//与服务端连接后第一个被调用 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("channelActive被调用。。。"); context = ctx; System.out.println("context"+context); }
//读取服务器的数据 @Override public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("MyClientRpcHandler。。。。channelRead被调用"); result = msg.toString(); notify();//唤醒等待的线程 } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println(cause.getMessage()); ctx.close(); } @Override public synchronized Object call() throws Exception { System.out.println("call被调用。。。"); context.writeAndFlush(para);//发送数据到服务端 wait();//等待channelRead获取到服务端返回的结果数据后，唤醒线程 return result;//服务端返回的数据 } public void setPara(String para){ System.out.println("setPara。。。被调用"); this.para = para; } }