Netty实践与NIO原理
程序员文章站
2022-08-19 13:22:00
一、阻塞IO与非阻塞IO Linux网络IO模型（5种） （1）阻塞IO模型 所有文件操作都是阻塞的，以套接字接口为例，在进程空间中调用recvfrom，系统调用直到数据包到达且被复制到应用进程缓冲区或发生错误时才返回，期间会一直等待（阻塞）。模型如图： （2）非阻塞IO模型 recvfrom从应用 ......
一、阻塞IO与非阻塞IO
Linux网络IO模型（5种）
（1）阻塞IO模型
所有文件操作都是阻塞的，以套接字接口为例，在进程空间中调用recvfrom，系统调用直到数据包到达且被复制到应用进程缓冲区或发生错误时才返回，期间会一直等待（阻塞）。模型如图：
（2）非阻塞IO模型
recvfrom从应用层到内核时，如果该缓冲区没数据，直接返回一个EWOULDBLOCK错误，反复轮询检查这个状态，看是否有数据到来。如图：
（3）IO复用模型
Linux提高select/poll，进程通过将一个或多个fd（file descriptor）传递给select或poll系统调用，阻塞在select操作上，侦测多个fd是否处于就绪状态。select/poll顺序扫描fd是否就绪，而且支持的fd数量有限。Linux还提供了一个epoll系统调用，使用基于事件驱动的方式代替顺序扫描，性能更高。当有fd就绪时，立即回调函数rollback。如图：
（4）信号驱动IO模型
首先开启套接口信号驱动IO功能，通过系统调用sigaction执行一个信号处理函数，该函数立即返回，进程继续工作，它是非阻塞的。当数据准备就绪时，就为该进程生成一个SIGIO信号，通过信号回调通知应用程序调用recfrom来读取数据，通知主循环函数处理数据。如图：
（5）异步IO模型
告知内核启动某个操作，让内核在整个操作完成后（包括将数据从内核复制到用户自己的缓冲区）通知我们。它与信号驱动的主要区别是：信号驱动IO由内核告知我们何时开始一个IO操作，异步IO模型由内核通知我们IO操作何时已经完成。如图所示：
IO多路复用的应用：
通过把多个IO的阻塞复用到一个select的阻塞上，使系统在单线程下可处理多个客户端请求。与传统多线程模型相比，最大优势是系统开销小，不需要创建额外进程或线程。主要应用场景如下：
（1）服务器需要同时处理多个处于监听状态或连接状态的套接字
（2）服务器需要同时处理多种网络协议的套接字
Linux最终选择epoll支持IO多路复用的系统调用，优点如下：
（1）支持一个进程打开的socket描述符（FD）不受限制（select单线程默认1024太少，epoll仅受限操作系统最大文件句柄数，1GB内存机器大约10万句柄）
（2）IO效率不会随FD数目增加而线性下降（只对“活跃”的socke进行t操作，活跃socket才会去主动调用callback函数）
（3）使用mmap加速内核与用户空间消息传递（同一块内存，避免不必要复制）
（4）API简单：创建epoll描述符，添加监听事件，阻塞等待监听事件发生，关闭epoll描述符等
二、阻塞IO的例子（结合线程池）
//1.服务端
package com.xbai.io;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import com.xbai.executor.TimeServerHandlerExecutePool;
import com.xbai.handler.TimeServerHandler;
public class TimeServerExecutor {
public static void main(String[] args)throws IOException {
int port =8080;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (Exception e) {
// TODO: handle exception
}
}
ServerSocket server =null;
try {
server =new ServerSocket(port);
System.out.println("The time server is started in port : " + port);
TimeServerHandlerExecutePool singleExecutor =new TimeServerHandlerExecutePool(50,10000);
while(true){
Socket socket = server.accept();
singleExecutor.execute(new TimeServerHandler(socket));
}
}finally {
if(server !=null){
System.out.println("The time server closed");
server.close();
server =null;
}
}
}
}
//2.服务端线程池
package com.xbai.executor;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TimeServerHandlerExecutePool {
private ExecutorService executor;
public TimeServerHandlerExecutePool(int maxPoolSize,int queueSize){
executor =new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),maxPoolSize,120L,TimeUnit.SECONDS,
new ArrayBlockingQueue(queueSize));//线程池要执行的任务阻塞成一个队列，其内部的机制是等待唤醒生产者和消费者线程，有一个生产就可唤醒一个消费，去看源码的线程池原理
}
public void execute(Runnable task){
executor.execute(task);
}
}
//3.服务端处理器
package com.xbai.handler;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.Socket;
import java.sql.Date;
public class TimeServerHandler implements Runnable{
private Socketsocket;
public TimeServerHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
// TODO Auto-generated method stub
BufferedReader br =null;
PrintWriter pw =null;
try {
br =new BufferedReader(new InputStreamReader(socket.getInputStream()));
pw =new PrintWriter(socket.getOutputStream(),true);
String curTime =null;
String msg =null;
while(true){
msg = br.readLine();
if(msg ==null){
break;
}
System.out.println("The time server received order:" + msg);
curTime ="query time order".equalsIgnoreCase(msg) ?new Date(
System.currentTimeMillis()).toString() :"bad order";
pw.println(curTime);//这里不写println,就无法插入换行符，那边就不能readLine,一直阻塞，无法获取数据
}
}catch (IOException e) {
if(br !=null){
try {
br.close();
}catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
if(pw !=null){
pw.close();
pw =null;
}
if(socket !=null){
try {
socket.close();
}catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
socket =null;
}
}
}
}
//4.客户端代码
package com.xbai.io;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.UnknownHostException;
public class TimeClient {
public static void main(String[] args) {
int port =8080;
if(args !=null && args.length >0){
try {
port = Integer.valueOf(args[0]);
}catch (Exception e) {
// TODO: handle exception
}
}
Socket socket =null;
BufferedReader br =null;
PrintWriter pw =null;
try {
socket =new Socket("localhost",port);
br =new BufferedReader(new InputStreamReader(socket.getInputStream()));
pw =new PrintWriter(socket.getOutputStream(),true);
pw.println("query time order");
System.out.println("send order succeed");
String resp = br.readLine();
System.out.println("Now is :" + resp);
}catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
if(pw !=null){
pw.close();
pw =null;
}
if(br !=null){
try {
br.close();
}catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
br =null;
}
if(socket !=null){
try {
socket.close();
}catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
socket =null;
}
}
}
}
执行结果
服务端启动及收发：
客户端发送和接收：
三、非阻塞IO的例子（原生Java NIO，目前有写半包等问题，怀疑服务端没有写出去导致的客户端Selector的关闭状态异常）
//1.服务端主程序
package com.xiaobai.nio;
public class NIOServer {
public static void main(String[] args) {
MultiplexerTimeServer timeServer = new MultiplexerTimeServer();
new Thread(timeServer,"NIO-MultiplexerTimeServer-001").start();
}
}
//2.服务端timeServer
package com.xiaobai.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
public class MultiplexerTimeServer implements Runnable {
private Selector selector;
private ServerSocketChannel servChannel;
private volatile boolean stop;
public MultiplexerTimeServer() {
try {
selector = Selector.open();//建立Selector
servChannel = ServerSocketChannel.open();//建立Channel
servChannel.configureBlocking(false);
servChannel.socket().bind(new InetSocketAddress(2048), 1024);//ServerSocket绑定
servChannel.register(selector, SelectionKey.OP_ACCEPT);//向Selector注册ACCEPT事件
System.out.println("The time server is started in port 2048");
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override
public void run() {
while(!stop){
try {
selector.select(1000);//轮询Channel
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
it.remove();//移除它
try {
handleInput(key);
} catch (Exception e) {
if(key != null){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
}
if(selector != null){
try {
selector.close();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
//处理新接入的请求
if(key.isAcceptable()){//此前已向Selector注册，并已open
//获取server channel
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//获取client channel
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
//第一次捕捉到的客户端向Selector注册READ事件
sc.register(selector, SelectionKey.OP_READ);
}
//处理已注册的读事件
if(key.isReadable()){
//获取客户端Channel
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);//读到缓冲
if(readBytes > 0){
readBuffer.flip();
// Buffer java.nio.Buffer.flip()
//
//
// Flips this buffer. The limit is set to the current position and then the position is set to zero. If the mark is defined then it is discarded.
//
// After a sequence of channel-read or put operations, invoke this method to prepare for a sequence of channel-write or relative get operations. For example:
//
// buf.put(magic); // Prepend header
// in.read(buf); // Read data into rest of buffer
// buf.flip(); // Flip buffer
// out.write(buf); // Write header + data to channel
byte[] bytes = new byte[readBuffer.remaining()];//缓冲中有多少个字节数据
readBuffer.get(bytes);
String body = new String(bytes,"UTF-8");
System.out.println("The time server receive order : " + body);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(//
System.currentTimeMillis()).toString() : "BAD ORDER";
doWrite(sc,currentTime);
}else if(readBytes < 0){
//贵在坚持！
//对端链路关闭
// key.cancel();
// sc.close();
}else{
;//读到0字节，忽略
}
}
}
}
private void doWrite(SocketChannel channel, String response) throws IOException{
if(response != null && response.trim().length() > 0){
byte[] bytes = response.getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);//根据字节数组容量创建ByteBuffer
writeBuffer.put(bytes);//字节数组复制到缓冲区
writeBuffer.flip();
channel.write(writeBuffer);//SocketChannel是异步非阻塞的，不保证一次发送完，出现“写半包”问题，
//这里缺少注册写操作，不断轮询Selector将没有发送完的ByteBuffer发送完毕
//TODO 这里有问题，没有写出去，导致客户端无法收到消息，显示Selector关闭状态异常
}
}
}
//3.客户端主程序
package com.xiaobai.nio;
public class NIOClient {
public static void main(String[] args) {
TimeClientHandle timeClientHandle = new TimeClientHandle("127.0.0.1",2048);
new Thread(timeClientHandle,"NIO-TimeClient-001").start();
}
}
//4.客户端timeClient
package com.xiaobai.nio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
public class TimeClientHandle implements Runnable {
private String host;
private int port;
private Selector selector;
private SocketChannel socketChannel;
private volatile boolean stop;
public TimeClientHandle(String host,int port) {
this.host = host==null?"127.0.0.1":host;
this.port = port;
try {
selector = Selector.open();
socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
} catch (Exception e) {
// TODO: handle exception
}
}
@Override
public void run() {
try {
doConnect();
} catch (Exception e) {
// TODO: handle exception
}
while(!stop){
try {
selector.select(3000);
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
SelectionKey key = null;
while(it.hasNext()){
key = it.next();
it.remove();
try {
handleInput(key);
} catch (Exception e) {
if(key != null){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
if(selector != null){
try {
selector.close();
} catch (Exception e) {
// TODO: handle exception
}
}
}
}
private void handleInput(SelectionKey key) throws Exception{
if(key.isValid()){
//判断是否连接成功
//连接方法中已有连接不成功注册连接事件的逻辑，反复尝试连接，这里判断，如果成功，注册该客户连接的read事件准备接收数据
SocketChannel sc = (SocketChannel) key.channel();
if(key.isConnectable()){
if(sc.finishConnect()){
sc.register(selector, SelectionKey.OP_READ);
doWrite(sc);//本客户向外写东西
}
}
//下面是从服务器接收数据
if(key.isReadable()){
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
int readBytes = sc.read(readBuffer);//读到缓冲
if(readBytes > 0){
readBuffer.flip();
// Buffer java.nio.Buffer.flip()
//
//
// Flips this buffer. The limit is set to the current position and then the position is set to zero. If the mark is defined then it is discarded.
//
// After a sequence of channel-read or put operations, invoke this method to prepare for a sequence of channel-write or relative get operations. For example:
//
// buf.put(magic); // Prepend header
// in.read(buf); // Read data into rest of buffer
// buf.flip(); // Flip buffer
// out.write(buf); // Write header + data to channel
byte[] bytes = new byte[readBuffer.remaining()];//缓冲中有多少个字节数据
readBuffer.get(bytes);
String body = new String(bytes,"UTF-8");
System.out.println("Now is : " + body);
this.stop = true;
}else if(readBytes < 0){
//贵在坚持！
//对端链路关闭
key.cancel();
sc.close();
}else{
;//读到0字节，忽略
}
}
}
}
private void doConnect() throws IOException {
//如果连接成功，则直接注册到多路复用器上，发送请求消息，读应答
if(socketChannel.connect(new InetSocketAddress(host, port))){//异步连接，直至成功
socketChannel.register(selector, SelectionKey.OP_READ);
doWrite(socketChannel);
}else{//注册连接事件，轮询直至连接成功
//异步，到底是什么概念？底层是什么原理？TCP/IP层面
socketChannel.register(selector, SelectionKey.OP_CONNECT);
}
}
private void doWrite(SocketChannel sc) throws IOException {
//本客户向外写东西
byte[] req = "QUERY TIME ORDER".getBytes();
ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
writeBuffer.put(req);
writeBuffer.flip();
sc.write(writeBuffer);
if(!writeBuffer.hasRemaining()){
System.out.println("Send order 2 server succeed.");
}
}
}
四、TCP与UDP
五、网络传输粘包与拆包问题
六、Netty入门案例与原理分析、Reactor模式
第一个例子：
//1.NettyServer
package com.xiaobai.server.netty;
import io.netty.bootstrap.ServerBootstrap;
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.NioServerSocketChannel;
import java.net.InetSocketAddress;
public class NettyServer {
private final int port;
public NettyServer(int port) {
this.port = port;
}
public static void main(String[] args) throws InterruptedException {
if(args.length != 1) {
System.err.println("Usage:" + NettyServer.class.getSimpleName() + " <port>");
return;
}
int port = Integer.parseInt(args[0]);
new NettyServer(port).start();
}
private void start() throws InterruptedException {
final NettyServerHandler serverHandler = new NettyServerHandler();
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(group).channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(serverHandler);
}
});
ChannelFuture f = b.bind().sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully().sync();
}
}
}
//2.NettyServerHandler
package com.xiaobai.server.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
@ChannelHandler.Sharable
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf in = (ByteBuf) msg;
System.out.println("Server received:" + in.toString(CharsetUtil.UTF_8));
ctx.write(in);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
.addListener(ChannelFutureListener.CLOSE);//关闭该Channel
}
}
//3.NettyClient
package com.xiaobai.server.netty;
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;
import java.net.InetSocketAddress;
public class NettyClient {
private final String host;
private final int port;
public NettyClient(String host, int port) {
this.host = host;
this.port = port;
}
public static void main(String[] args) throws InterruptedException {
if(args.length != 2) {
System.err.println("Usage:" + NettyServer.class.getSimpleName() + " <host> <port>");
return;
}
String host = args[0];
int port = Integer.parseInt(args[1]);
new NettyClient(host,port).start();
}
private void start() throws InterruptedException {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(host,port))
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new NettyClientHandler());
}
});
ChannelFuture f = b.connect().sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully().sync();
}
}
}
//4.NettyClientHandler
package com.xiaobai.server.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelHandlerInvoker;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.concurrent.EventExecutorGroup;
public class NettyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
@Override
protected void messageReceived(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
System.out.println("Client received: " + byteBuf.toString(CharsetUtil.UTF_8));
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!",CharsetUtil.UTF_8));
}
}
执行结果：
服务端：
客户端：
第二个例子：
//服务端启动spring配置文件applicationContext.xml
<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
<bean id="nettyServer" class="com.xiaobai.netty.server.NettyServer" init-method="start">
<constructor-arg index="0" type="int">
<value>8888</value>
</constructor-arg>
</bean>
</beans>
//服务端启动类
package com.xiaobai.netty.spring;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class SpringStart {
public static void main(String[] args) {
ApplicationContext application = new ClassPathXmlApplicationContext("com/xiaobai/netty/spring/applicationContext.xml");
}
}
//NettyServer
package com.xiaobai.netty.server;
import com.xiaobai.netty.handlers.ChildChannelHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import org.apache.log4j.Logger;
public class NettyServer {
private static final Logger logger = Logger.getLogger(NettyServer.class);
//无参
public NettyServer() {
}
//用于spring管理构造函数初始化bean
public NettyServer(int port) {
this.port = port;
}
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
private ServerBootstrap bootstrap;
private int port;
public void start() {
bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
try {
bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG,1024)
.childHandler(new ChildChannelHandler());
//同步等待绑定端口成功
ChannelFuture future = bootstrap.bind(port).sync();
System.out.println("NettyServer Successfully Started in port:" + port);
logger.info("NettyServer Successfully Started in port:" + port);
//同步等待服务器端口关闭
future.channel().closeFuture().sync();//经实验，这是阻塞方法，一直阻塞
}catch (Exception e) {
//优雅退出，释放线程池资源
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
//ChildChannelHandler
package com.xiaobai.netty.handlers;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
//注意参数化类型不要引错包
public class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
//原理来自平时读书实践（nio,aio,同步/异步原理，底层）
//接收到客户端连接（可连接）后的初始化动作，使用责任链模式绑定一系列数据读写操作，用于可读可写时的操作
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new TimeServerHandler());//由简入难，不断调试、琢磨的框架
}
}
//TimeServerHandler
package com.xiaobai.netty.handlers;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import java.util.Date;
public class TimeServerHandler extends ChannelHandlerAdapter {
private int counter;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// ByteBuf buf = (ByteBuf) msg;
// byte[] req = new byte[buf.readableBytes()];
// buf.readBytes(req);
// String body = new String(req,"UTF-8").substring(0,req.length - System.getProperty("line.separator").length());
String body = (String)msg;
System.out.println("The time server receive order:" + body + " ; the counter is : " + ++counter);
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() :"BAD ORDER";
currentTime = currentTime + System.getProperty("line.separator");
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
//这些都是主程序代码中出现这些情况后调用的接口代码
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
}
//客户端启动类
package com.xiaobai.netty.client;
import org.apache.log4j.Logger;
public class NettyTest {
private static final Logger logger = Logger.getLogger(NettyTest.class);
public static void main(String[] args) {
try {
logger.info("Netty communication start!");
NettyClient client = new NettyClient("127.0.0.1",8888);
client.send();
}catch (Exception e) {
logger.error("Client connected failure!");
}
}
}
//NettyClient
package com.xiaobai.netty.client;
import com.xiaobai.netty.handlers.ChildChannelHandler;
import com.xiaobai.netty.handlers.TimeClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import org.apache.log4j.Logger;
public class NettyClient {
private static final Logger logger = Logger.getLogger(NettyClient.class);
//无参
public NettyClient() {
}
//用于spring管理构造函数初始化bean
public NettyClient(String host,int port) {
this.host = host;
this.port = port;
}
private EventLoopGroup group;
private Bootstrap bootstrap;
private int port;
private String host;
public void send() {
try {
group = new NioEventLoopGroup();
if(host != null && host.trim() != "") {
bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY,true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
//发起异步连接操作
System.out.println("NettyClient connecting " + host +":" + port);
ChannelFuture future = bootstrap.connect(host,port).sync();
logger.info("NettyClient connected " + host +":" + port);
//等待客户端关闭
future.channel().closeFuture().sync();
}else{
logger.info("accessing nowhere!");
}
}catch (Exception e) {
//优雅退出，释放线程池资源
group.shutdownGracefully();
}
}
}
//TimeClientHandler
package com.xiaobai.netty.handlers;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import org.apache.log4j.Logger;
public class TimeClientHandler extends ChannelHandlerAdapter {
private static final Logger logger = Logger.getLogger(TimeClientHandler.class);
private int counter;
private byte[] req;
public TimeClientHandler() {
req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
logger.warn("Unexpected exception from downstream:" + cause.getMessage());
ctx.close();
}
//已连接后发送消息
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//不要反复用一个ByteBuf,经测试会出现无法发布的问题--上网多查，研究其中原理！！
ByteBuf message = null;
for(int i = 0;i < 100; i++) {
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// ByteBuf buf = (ByteBuf) msg;
// byte[] req = new byte[buf.readableBytes()];
// buf.readBytes(req);
// String body = new String(req,"UTF-8");
String body = (String)msg;
System.out.println("Now is:" + body + " ; the counter is : " + ++counter);
}
}
执行结果：
服务端：
客户端：
原理与源码分析：
NioEventLoopGroup:一个线程组，包含了一组NIO线程，专门用于网络事件处理，实际上它们就是Reactor线程组。这里创建两个，一个用于服务端接受客户端的连接，另一个用于SocketChannel的网络读写。
ServerBootstrap:Netty用于启动NIO服务端的辅助启动类，降低开发复杂度。它的group方法将两个NIO线程组当作入参传递到ServerBootstrap中。backlog:TCP参数，这里设置为1024.
NioServerSocketChannel:功能对应于JDK NIO类库中的ServerSocketChannel类。
ChildChannelHandler:绑定I/O事件的处理类，作用类似于Reactor模式中的Handler类，主要用于处理网络I/O事件，例如记录日志、对消息进行编解码等。
服务端启动辅助类配置完成后，调用它的bind方法绑定监听端口，随后调用它的同步阻塞方法sync等待绑定操作完成。完成之后Netty会返回一个ChannelFuture,它的功能类似于JDK的java.util.concurrent.Future,主要用于异步操作的通知回调。
future.channel().closeFuture.sync()是阻塞方法（一直阻塞，直到服务关闭），等待服务端链路关闭后才退出。
shutdownGracefully方法：优雅退出，释放关联资源。
ByteBuf:类似于JDK中的java.nio.ByteBuffer对象，不过它提供了更加强大和灵活的功能。通过ByteBuf的readableBytes方法可获取缓冲区可读字节数，根据可读字节数创建byte数组，通过ByteBuf的readBytes方法将缓冲区字节数组复制到新建byte数组中，通过ChannelHandlerContext的write方法异步发送应答消息给客户端。
ChannelHandlerContext的flush方法：将消息发送队列中的消息写入到SocketChannel中发送给对方。从性能角度考虑，为了防止频繁唤醒Selector进行消息发送（Writable事件），Netty的write方法并不直接将消息写入SocketChannel中，只是把待发送的消息放到发送缓冲数组中，再通过调用flush方法，将发送缓冲区中的消息全部写到SocketChannel中。
七、Netty对粘包拆包的解决方案
八、编码与解码
九、序列化与反序列化
十、网络传输私有协议制定与聊天室业务实现
上一篇： 雨霖玲（爆笑版 绝对经典）
下一篇： 打DOTA的境界