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 读取 |
|