Netty框架学习01-Netty服务端开发和Netty客户端开发
环境准备java
依赖：数据库
<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>5.0.0.Alpha1</version>
</dependency>
回顾一下NIO进行服务端开发的步骤bootstrap
- 建立ServerSocketchannel，配置他为非阻塞模式
- 绑定监听，配置TCP参数，好比backlog大小
- 建立一个独立的IO线程，用于轮询多路复用器Selector
- 建立Selector，将以前建立的ServerSocketChannel注册到Selector上，监听Selectionkey.Accept;
- 启动IO线程，在循环中执行Selector。select()方法，轮询就绪的channle;
- 当轮询到了处于就绪状态的Channel时，须要对其进行判断，若是是OP_Accept状态，说明是新的客户端接入，则调用ServerSocketChannel。accept（）方法接受新的客户端；
- 设置新接入的客户端链路SocketChannel为非阻塞模式，配置其余的一些TCP参数
- 将SocketChannel注册到Selector，监听OP_READ操做位；
- 若是轮询的Channel为Op_READ，则说明SocketChannel中有新的就绪的数据包须要读取，则构造ByteBuffer对象，读取数据包；
- 若是轮询的Channel为Op_WRITE，说明还有数据没有发送完成，须要继续发送。
一个简单NIO服务端程序，须要十几步，而使用Netty就简单了。数组
Netty服务端有专门的线程组，用于服务端接收客户端的链接，另外一个用于接收SocketChannel的网络链接读写，它的实现原理和nio的传统方式很相似，nio是经过ServerSocketChannel进行做为数据的传输通道，他也有本身传输通道NioServerSocketChannel，他须要一个IO事件处理器，这个处理器主要用于处理io事件，记录日志，消息编码等操做。缓存
Netty服务端代码实现以下：服务器
package com.soecode.lyf.demo.test.netty.server;
import io.netty.bootstrap.ServerBootstrap;
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;
/**
* @author 魏文思
* @date 2019/11/22$ 9:48$
*/
public class TimeServer {
public void bind(int port) throws Exception {
//配置服务端的nio线程组
/**
* NioEventLoopGroup 是个线程组，它包含了一组nio线程，专门用于服务端接受客户端网络事件的处理，
* 实际上他们呢就是Reactor线程组。建立两个的缘由一个是用于服务端接受客户端的链接，另外一个用于进行SocketChannel的网络读写。
*/
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
/**
* Serverboostrap是netty用于启动服务辅助启动类，目的是下降服务端的开发复杂度。
*/
ServerBootstrap b = new ServerBootstrap();
/**
* 调用group方法，将两个nio线程组看成传入参数传递到ServcerbootStrap中。
* 接着设置建立的Channel为NioServerSocketChannel，他的功能对应于JDK nio类库中的ServerSocketChannel类。
* 而后配置NioServerSocketChannel的tcp参数，此处将他的backlog设置为1024，
* 最后绑定IO事件的处理类childrenChannelHandler，他的做用相似于Reactor模式中Handler类，主要用于从处理网络IO事件，例如记录日志，对消息进行编码
*/
b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChildChannelHanler());
//绑定客户端，同步等待成功
/**
* 服务端启动辅助类配置完成以后，调用它的bind方法绑定监听器端口，
* 随后调用它的 同步阻塞方法sync等到绑定操做完成。
* 完成以后Netty会返回一个ChannelFuture，它的功能相似于jdkjdk current 包里面的Future,主要用于异步操做的通知回调。
*/
ChannelFuture f = b.bind(port).sync();
//等待服务端监听
// 端口关闭
/**
* 调用f.channel().closeFuture().sync()来进行阻塞，等待服务端链路关闭以后main函数才退出
*/
System.out.println("1");
f.channel().closeFuture().sync();
System.out.println("2");
} catch (Exception e) {
e.printStackTrace();
} finally {
//优雅退出
/**
* 它会释放跟踪shutdownGracefully相关联的资源。
*/
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
//IO事件处理器 用于异步操做的通知回调
private class ChildChannelHanler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new TimeServerHandler());
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
Integer post = Integer.valueOf(args[0]);
} catch (Exception e) {
//采用默认值
e.printStackTrace();
}
}
new TimeServer().bind(port);
}
}
nettyIo事件处理器以下：网络
package com.soecode.lyf.demo.test.netty.openthedoor;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import java.util.Date;
/**
* Netty时间服务器服务端
* <p>
* 主要用于对网络事件进行读写操做，
* 一般咱们只关心channelRead和exceptionCaught方法。
*
* @author 魏文思
* @date 2019/11/22$ 10:06$
*/
public class TimeServerHandler extends ChannelHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
/**
* 将msg转换成了Netty的ByteBuf对象。Bytebuf相似于jdk中的Java.nio.ByteBuffer对象，不过他提供 了更增强大和灵活的功能。
* 经过ByteBuf的readablebytes方法能够获取缓冲区可读取的字节数，根据可读的字节数建立byte数组，经过ByteBuf的readBytes方法能够获取缓冲区可读的字节数，
* 根据可读的字节数建立byte数组，经过Bytebuf的readBytes方法将缓冲区中的字节数组复制到新建的byte数组中，最后经过new String的构造函数获取请求消息。
* 这时对请求消息进行判断，若是是QUERY TIME ORDER 则建立应答消息，经过ChannelHandlerContent的write方法异步发送应答消息给客户端。
*/
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "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";
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
/**
* 调用flush（）方法它的做用是将消息发送队列中的消息写入到SocketChannel中发送给对方。
* 从性能角度考虑，为了防止频繁你唤醒Selector进行消息发送，Netty的write方法并不直接将消息写入SocketChannel中,调用write方法只是把待发送的消息发送到缓存数组中，
* 再经过调用flush方法，将发送缓存区中的消息所有写道Socketchannel中。
*
* 这种涉及思想在批量插入数据库的有相似操做。提升性能
*
*/
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
/**
* 当发生异常时，关闭ChannelHandlerContext，释放和ChannelhanlerContext相关联的句柄资源。
*/
ctx.close();
}
}
IO事件处理器这里使用了适配器模式，在ChannelHandlerAdapter这个适配器里面有不少Netty提供了IO处理器能够作的事情，一般选择一些必要的方法进行实现。异步
实现很简单，很传统的nio很相似，它封装了一个本身的启动类，ServerBootStrap，这个启动类再进行一些配置，配置服务通讯须要的一些必要条件，通道，处理器，回调等等。socket
NIO客户端的实现tcp
package com.soecode.lyf.demo.test.netty.client;
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.NioSocketChannel;
/**
* netty的客户端
*
* @author 魏文思
* @date 2019/11/22$ 13:59$
*/
public class TimeClient {
public void connect(int port, String host) throws Exception {
//配置客户端线程组
EventLoopGroup group = new NioEventLoopGroup();
try {
//配置启动工具
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) {
channel.pipeline().addLast(new TimeClientHandler());
}
});
//发起异步链接操做
ChannelFuture sync = bootstrap.connect(host, port).sync();
//等待客户端链路关闭
sync.channel().closeFuture().sync();
} catch (Throwable e) {
//Unable to create Channel from class class io.netty.channel.socket.nio.NioServerSocketChannel
e.printStackTrace();
} finally {
//优雅退出，释放nio线程组
group.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
//采用默认值
}
}
new TimeClient().connect(port, "127.0.0.1");
System.out.println("----------------end");
}
}
package com.soecode.lyf.demo.test.netty.client;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
/**
* @author 魏文思
* @date 2019/11/22$ 14:07$
*/
@Slf4j
public class TimeClientHandler extends ChannelHandlerAdapter {
private final ByteBuf firstMessage;
public TimeClientHandler() {
byte[] req = "QUERY TIME ORDER".getBytes();
firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
}
/**
* 当客户端和服务端tcp链路创建链接以后，Netty的NIO线程会调用channelActive，
* 发送查询时间的指令给服务端，调用ChannelhandlerContext的writeAndflush方法将请求消息发送给服务端
* @param ctx
* @throws Exception
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(firstMessage);
}
/**
* 当服务端返回应答消息时，channelRead方法被调用，将消息打印
* @param ctx
* @param msg
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("channelRead");
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println("Now is :" + body);
}
/**
* 当发生异常时，打印异常日志，释放客户端资源
* @param ctx
* @param cause
* @throws Exception
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("exceptionCaught");
//释放资源
log.warn("Unexceped exception from downstream :" + cause.getMessage());
ctx.close();
}
}
这里我发犯了个错误，
将NioSocketChannel写成了NIOServerSocketChannel结果启动的时候报错：
启动服务端，启动客户端
控制打印以下：
服务端：
客户端以下：
至此实现netty客户端和服务端的链接。
本篇文章参考《Netty 权威指南》 第二版。