黑马程序员技术交流社区
标题:
【广州校区】+【原创】+基于Netty的高性能RPC框架-Dubbo
[打印本页]
作者:
小李哥
时间:
2019-9-12 11:13
标题:
【广州校区】+【原创】+基于Netty的高性能RPC框架-Dubbo
Netty概述：
1、netty是基于Java NIO的网络应用框架，client-server框架
2、Netty是一个高性能、异步事件驱动的NIO框架，它提供了对TCP、UDP和文件传输的支持，
作为一个异步NIO框架，Netty的所有IO操作都是异步非阻塞的，
通过Future-Listener机制，用户可以方便的主动获取或者通过通知机制获得IO操作结果。
3、作为当前最流行的NIO框架，Netty在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用，
一些业界著名的开源组件也基于Netty的NIO框架构建。
Netty创建步骤：
NIO通讯服务端步骤：
1、创建ServerSocketChannel,为它配置非阻塞模式
2、绑定监听，配置TCP参数，录入backlog大小等
3、创建一个独立的IO线程，用于轮询多路复用器Selector
4、创建Selector，将之前的ServerSocketChannel注册到Selector上，并设置监听标识位SelectionKey.ACCEPT
5、启动IO线程，在循环体中执行Selector.select()方法，轮询就绪的通道
6、当轮询到处于就绪的通道时，需要进行判断操作位，如果是ACCEPT状态，说明是新的客户端介入，则调用accept方法接受新的客户端。
7、设置新接入客户端的一些参数，并将其通道继续注册到Selector之中。设置监听标识等
8、如果轮询的通道操作位是READ，则进行读取，构造Buffer对象等
9、更细节的还有数据没发送完成继续发送的问题
.在Netty中包含下面几个主要的组件：
Bootstrap：netty的组件容器，用于把其他各个部分连接起来；如果是TCP的Server端，则为ServerBootstrap.
Channel：代表一个Socket的连接
EventLoopGroup：一个Group包含多个EventLoop，可以理解为线程池
EventLoop：处理具体的Channel，一个EventLoop可以处理多个Channel
ChannelPipeline：每个Channel绑定一个pipeline，在上面注册处理逻辑handler
Handler：具体的对消息或连接的处理，有两种类型，Inbound和Outbound。分别代表消息接收的处理和消息发送的处理。
ChannelFuture：注解回调方法
通过轮训监听的方式去监听客户端的消息，客户端进行编码，服务端接收的消息进行解码。这样一个小例子。
创建服务端：
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
public class EchoServer {
private final int port;
public EchoServer(int port) {
this.port = port;
}
public void start() throws Exception {
EventLoopGroup eventLoopGroup = null;
try {
//创建ServerBootstrap实例来引导绑定和启动服务器
ServerBootstrap serverBootstrap = new ServerBootstrap();
//创建NioEventLoopGroup对象来处理事件，如接受新连接、接收数据、写数据等等
eventLoopGroup = new NioEventLoopGroup();
//指定通道类型为NioServerSocketChannel，设置InetSocketAddress让服务器监听某个端口已等待客户端连接。
serverBootstrap.group(eventLoopGroup).channel(NioServerSocketChannel.class).localAddress("localhost",port).childHandler(new ChannelInitializer<Channel>() {
//设置childHandler执行所有的连接请求
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new EchoServerHandler());
}
});
// 最后绑定服务器等待直到绑定完成，调用sync()方法会阻塞直到服务器完成绑定,然后服务器等待通道关闭，因为使用sync()，所以关闭操作也会被阻塞。
ChannelFuture channelFuture = serverBootstrap.bind().sync();
System.out.println("开始监听，端口为：" + channelFuture.channel().localAddress());
channelFuture.channel().closeFuture().sync();
} finally {
eventLoopGroup.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoServer(20000).start();
}
}
服务端监听器：package nio.netty.demo.EchoServer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.util.Date;
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
System.out.println("server 读取数据……");
//读取数据
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println("接收客户端数据:" + body);
//向客户端写数据
System.out.println("server向client发送数据");
String currentTime = new Date(System.currentTimeMillis()).toString();
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
System.out.println("server 读取数据完毕..");
ctx.flush();//刷新后才将数据发出到SocketChannel
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
cause.printStackTrace();
ctx.close();
}
}
客户端创建：package nio.netty.demo.EchoServer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
//xx
import java.net.InetSocketAddress;
//aa
public class EchoClient {
private final String host;
private final int port;
public EchoClient(String host, int port) {
this.host = host;
this.port = port;
}
public void start() throws Exception {
EventLoopGroup nioEventLoopGroup = null;
try {
//创建Bootstrap对象用来引导启动客户端
Bootstrap bootstrap = new Bootstrap();
//创建EventLoopGroup对象并设置到Bootstrap中，EventLoopGroup可以理解为是一个线程池，这个线程池用来处理连接、接受数据、发送数据
nioEventLoopGroup = new NioEventLoopGroup();
//创建InetSocketAddress并设置到Bootstrap中，InetSocketAddress是指定连接的服务器地址
bootstrap.group(nioEventLoopGroup).channel(NioSocketChannel.class).remoteAddress(new InetSocketAddress(host, port))
.handler(new ChannelInitializer<SocketChannel>() {
//添加一个ChannelHandler，客户端成功连接服务器后就会被执行
@Override
protected void initChannel(SocketChannel ch)
throws Exception {
ch.pipeline().addLast(new EchoClientHandler());
}
});
// • 调用Bootstrap.connect()来连接服务器
ChannelFuture f = bootstrap.connect().sync();
// • 最后关闭EventLoopGroup来释放资源
f.channel().closeFuture().sync();
} finally {
nioEventLoopGroup.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws Exception {
new EchoClient("localhost", 20000).start();
}
}
客户端监听：package nio.netty.demo.EchoServer;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
//客户端连接服务器后被调用
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("客户端连接服务器，开始发送数据……");
byte[] req = "小李哥上綫了！".getBytes();
ByteBuf firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
ctx.writeAndFlush(firstMessage);
}
//• 从服务器接收到数据后调用 监听服务器端接受数据的状态
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
System.out.println("client 读取server数据..");
//服务端返回消息后
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println("服务端数据为 :" + body);
/*ByteBuf firstMessage = Unpooled.buffer("XXX".getBytes().length);
firstMessage.writeBytes("XXX".getBytes());
ctx.writeAndFlush(firstMessage);*/
}
//• 发生异常时被调用
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("client exceptionCaught..");
// 释放资源
ctx.close();
}
}
服务端控制台输出："C:\Program Files\Java\jdk1.8.0_111\bin\java" "-javaagent:C:\installSoft\developTools\intellij idea\IntelliJ IDEA 2017.2.4\lib\idea_rt.jar=57261:C:\installSoft\developTools\intellij idea\IntelliJ IDEA 2017.2.4\bin" -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_111\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\rt.jar;E:\mayun\rpc-svn\NettyDemo\target\classes;C:\Users\Administrator\.m2\repository\io\netty\netty-all\4.1.24.Final\netty-all-4.1.24.Final.jar" nio.netty.demo.EchoServer.EchoServer开始监听，端口为：/127.0.0.1:20000server 读取数据……接收客户端数据:小李哥上綫了！server向client发送数据server 读取数据完毕..
客户端控制台输出：
"C:\Program Files\Java\jdk1.8.0_111\bin\java" "-javaagent:C:\installSoft\developTools\intellij idea\IntelliJ IDEA 2017.2.4\lib\idea_rt.jar=57298:C:\installSoft\developTools\intellij idea\IntelliJ IDEA 2017.2.4\bin" -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_111\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_111\jre\lib\rt.jar;E:\mayun\rpc-svn\NettyDemo\target\classes;C:\Users\Administrator\.m2\repository\io\netty\netty-all\4.1.24.Final\netty-all-4.1.24.Final.jar" nio.netty.demo.EchoServer.EchoClient客户端连接服务器，开始发送数据……client 读取
欢迎光临 黑马程序员技术交流社区 (http://bbs.itheima.com/)
黑马程序员IT技术论坛 X3.2