用Netty实现RPC

一、RPC 基本介绍

RPC（Remote Procedure Call）— 远程过程调用，是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程
两个或多个应用程序都分布在不同的服务器上，它们之间的调用都像是本地方法调用一样(如图)

3）常见的 RPC 框架有: 比较知名的如阿里的Dubbo、google 的gRPC、Go 语言的rpcx、Apache 的thrift，Spring 旗下的 Spring Cloud。

二、 RPC 调用流程图

三、 PRC 调用流程说明

服务消费方(client)以本地调用方式调用服务
client stub 接收到调用后负责将方法、参数等封装成能够进行网络传输的消息体
client stub 将消息进行编码并发送到服务端
server stub 收到消息后进行解码
server stub 根据解码结果调用本地的服务
本地服务执行并将结果返回给 server stub
server stub 将返回导入结果进行编码并发送至消费方
client stub 接收到消息并进行解码
服务消费方(client)得到结果

小结：RPC 的目标就是将2-8这些步骤都封装起来，用户无需关心这些细节，可以像调用本地方法一样即可完成远程服务调用

四、自己实现 dubbo RPC(基于 Netty)

【1】需求说明

dubbo 底层使用了 Netty 作为网络通讯框架，要求用 Netty 实现一个简单的 RPC 框架
模仿 dubbo，消费者和提供者约定接口和协议，消费者远程调用提供者的服务，提供者返回一个字符串，消费者打印提供者返回的数据。底层网络通信使用 Netty 4.1.20

【2】设计说明

创建一个接口，定义抽象方法。用于消费者和提供者之间的约定。
创建一个提供者，该类需要监听消费者的请求，并按照约定返回数据。
创建一个消费者，该类需要透明的调用自己不存在的方法，内部需要使用 Netty 请求提供者返回数据
开发的分析图

【3】代码实现

NettyServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;public class NettyServer {public static void startServer(String hostName, int port) {startServer0(hostName, port);}// 编写一个方法，完成对NettyServer的初始化和启动private static void startServer0(String hostname, int port) {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new StringDecoder());pipeline.addLast(new StringEncoder());pipeline.addLast(new NettyServerHandler()); // 业务处理器}});ChannelFuture channelFuture = serverBootstrap.bind(hostname, port).sync();System.out.println("服务提供方开始提供服务~~");channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

NettyServerHandler

import com.atguigu.netty.dubborpc.customer.ClientBootstrap;
import com.atguigu.netty.dubborpc.provider.HelloServiceImpl;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;//服务器这边handler比较简单
public class NettyServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {//获取客户端发送的消息，并调用服务System.out.println("msg=" + msg);//客户端在调用服务器的api 时，我们需要定义一个协议//比如我们要求 每次发消息是都必须以某个字符串开头 "HelloService#hello#你好"if (msg.toString().startsWith(ClientBootstrap.providerName)) {String result = new HelloServiceImpl().hello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));ctx.writeAndFlush(result);}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}
}

NettyClient

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.Proxy;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class NettyClient {// 创建线程池private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());private static NettyClientHandler client;private int count = 0;// 编写方法使用代理模式，获取一个代理对象public Object getBean(final Class<?> serivceClass, final String providerName) {return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] { serivceClass },(proxy, method, args) -> {System.out.println("(proxy, method, args) 进入...." + (++count) + " 次");// {} 部分的代码，客户端每调用一次 hello, 就会进入到该代码if (client == null) {initClient();}// 设置要发给服务器端的信息// providerName 协议头 args[0] 就是客户端调用api hello(???), 参数client.setPara(providerName + args[0]);//return executor.submit(client).get();});}// 初始化客户端private static void initClient() {client = new NettyClientHandler();// 创建EventLoopGroupNioEventLoopGroup group = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new StringDecoder());pipeline.addLast(new StringEncoder());pipeline.addLast(client);}});try {bootstrap.connect("127.0.0.1", 7000).sync();} catch (Exception e) {e.printStackTrace();}}
}

NettyClientHandler

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.concurrent.Callable;public class NettyClientHandler extends ChannelInboundHandlerAdapter implements Callable {private ChannelHandlerContext context;// 上下文private String result; // 返回的结果private String para; // 客户端调用方法时，传入的参数// 与服务器的连接创建后，就会被调用, 这个方法是第一个被调用(1)@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(" channelActive 被调用  ");context = ctx; // 因为我们在其它方法会使用到 ctx}// 收到服务器的数据后，调用方法 (4)//@Overridepublic synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println(" channelRead 被调用  ");result = msg.toString();notify(); // 唤醒等待的线程}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();}// 被代理对象调用, 发送数据给服务器，-> wait -> 等待被唤醒(channelRead) -> 返回结果 (3)-》5@Overridepublic synchronized Object call() throws Exception {System.out.println(" call1 被调用  ");context.writeAndFlush(para);// 进行waitwait(); // 等待channelRead 方法获取到服务器的结果后，唤醒System.out.println(" call2 被调用  ");return result; // 服务方返回的结果}// (2)void setPara(String para) {System.out.println(" setPara  ");this.para = para;}
}

HelloServiceImpl

public class HelloServiceImpl implements HelloService{private static int count = 0;//当有消费方调用该方法时， 就返回一个结果@Overridepublic String hello(String mes) {System.out.println("收到客户端消息=" + mes);//根据mes 返回不同的结果if(mes != null) {return "你好客户端, 我已经收到你的消息 [" + mes + "] 第" + (++count) + " 次";} else {return "你好客户端, 我已经收到你的消息 ";}}
}

HelloService

//这个是接口，是服务提供方和 服务消费方都需要
public interface HelloService {String hello(String mes);
}

ServerBootstrap

//ServerBootstrap 会启动一个服务提供者，就是 NettyServer
public class ServerBootstrap {public static void main(String[] args) {//代码代填..NettyServer.startServer("127.0.0.1", 7000);}
}

ClientBootstrap

import com.atguigu.netty.dubborpc.netty.NettyClient;
import com.atguigu.netty.dubborpc.publicinterface.HelloService;public class ClientBootstrap {// 这里定义协议头public static final String providerName = "HelloService#hello#";public static void main(String[] args) throws Exception {// 创建一个消费者NettyClient customer = new NettyClient();// 创建代理对象HelloService service = (HelloService) customer.getBean(HelloService.class, providerName);for (;;) {Thread.sleep(2 * 1000);// 通过代理对象调用服务提供者的方法(服务)String res = service.hello("你好 dubbo~");System.out.println("调用的结果 res= " + res);}}
}

用Netty实现RPC相关推荐

Netty和RPC框架线程模型分析
<Netty 进阶之路>.<分布式服务框架原理与实践>作者李林锋深入剖析Netty和RPC框架线程模型.李林锋已在 InfoQ 上开设 Netty 专题持续出稿,感兴趣的同学可 ...
Netty 和 RPC 框架线程模型分析
https://www.infoq.cn/article/9Ib3hbKSgQaALj02-90y 1. 背景 1.1 线程模型的重要性对于 RPC 框架而言,影响其性能指标的主要有三个要素: I/ ...
基于Netty的RPC简易实现
代码地址如下: http://www.demodashi.com/demo/13448.html 可以给你提供思路也可以让你学到Netty相关的知识当然,这只是一种实现方式需求看下图,其实这个 ...
基于Netty的RPC架构实战演练
基于Netty的RPC架构实战演练 NIO netty服务端 netty客户端 netty线程模型源码分析(一) netty线程模型源码分析(二) netty5案例学习 netty学习之心跳 prot ...
使用Netty实现RPC
Netty作为一个异步事件驱动的网络应用框架,可以用于快速开发可维护的高性能服务器和客户端.国内著名的RPC框架Dubbo底层使用的是Netty作为网络通信的.本篇文章我们来探索一下RPC框架的本质以 ...
Java编写基于netty的RPC框架
一简单概念RPC: ( Remote Procedure Call),远程调用过程,是通过网络调用远程计算机的进程中某个方法,从而获取到想要的数据,过程如同调用本地的方法一样.阻塞IO :当阻塞I/ ...
基于Netty的RPC框架
概述 RPC(Remote Procedure Call),远程过程调用,是一个计算机通信协议,该协议允许运行一个进程调用另一个进程,而程序员无需额外为这个交互作用编程. 两个或者多个应用程序分布在不 ...
手动实现一个基于netty的RPC框架（模拟dubble）
轻量级RPC框架开发内容安排: 1.掌握RPC原理 2.掌握nio操作 3.掌握netty简单的api 4.掌握自定义RPC框架 RPC原理学习什么是RPC RPC(Remote Procedur ...
谈谈如何使用Netty开发实现高性能的RPC服务器
RPC(Remote Procedure Call Protocol)远程过程调用协议,它是一种通过网络,从远程计算机程序上请求服务,而不必了解底层网络技术的协议.说的再直白一点,就是客户端在不必知道 ...

用Netty实现RPC

用Netty实现RPC

一、RPC 基本介绍

二、 RPC 调用流程图

三、 PRC 调用流程说明

四、自己实现 dubbo RPC(基于 Netty)

【1】需求说明

【2】设计说明

【3】代码实现

NettyServer

NettyServerHandler

NettyClient

NettyClientHandler

HelloServiceImpl

HelloService

ServerBootstrap

ClientBootstrap

用Netty实现RPC相关推荐

最新文章

热门文章