NettyTimeServer
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;
public class NettyTimeServer {
private void bind(int port) throws Exception {
/**
* NioEventLoopGroup 就是NIO的異步線程組，包含了一組NIO線程，專門處理網(wǎng)絡事件，實際上就是Reactor線程組
*
* bossGroup： 用來接受客戶端的連接
* workerGroup：用于SocketChannel的網(wǎng)絡讀寫
*
*/
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
/**
* ServerBootstrap 這個類是NIO服務端的輔助啟動類，目的是降低服務端的開發(fā)復雜度
*
*/
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup) // 兩個線程組傳入
.channel(NioServerSocketChannel.class) // 設置Channel為 NioServerSocketChannel
.option(ChannelOption.SO_BACKLOG, 1024) // 設置 tcp參數(shù)，backlog，這里為 1024
.childHandler(new TimeServerChannelHandler()) // 綁定IO事件處理類，主要用來處理網(wǎng)絡IO事件，例如記錄日志，對消息進行編碼等
.childOption(ChannelOption.SO_KEEPALIVE, true);
// 綁定端口，同步等待成功
ChannelFuture future = serverBootstrap.bind(port).sync();
// 等待服務器監(jiān)聽端口關閉
future.channel().closeFuture().sync();
} catch (Exception e) {
// TODO: handle exception
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.parseInt(args[0]);
} catch (Exception e) {
// TODO: handle exception
}
}
new NettyTimeServer().bind(port);
}
}
TimeServerChannelHandler
import java.util.Date;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* 用來處理網(wǎng)絡事件的讀寫操作
* 只需要關注 channelRead 和 exceptionCaught 方法
*/
public class TimeServerChannelHandler extends ChannelInboundHandlerAdapter {
/**
*
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // (2)
/**
* ByteBuf 類似jdk的 ByteBuffer，功能更加強大靈活
*
* buf.readableBytes()：獲取緩沖區(qū)可讀的字節(jié)數(shù)
*
*
*/
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()]; // 根據(jù)可讀的字節(jié)數(shù)創(chuàng)建byte數(shù)組
buf.readBytes(req); // 將緩沖區(qū)中的字節(jié)數(shù)組復制到新建的byte數(shù)組中
String body = new String(req, "UTF-8"); // 獲取請求的消息
System.out.println("The time server receive order : " + body);
/**
* 如果消息是 QUERY TIME ORDER， 則創(chuàng)建應答消息
*/
String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date().toString() : "BAD ORDER";
/**
* 將應答消息轉(zhuǎn)化為 ByteBuf
*/
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.write(resp); // 異步發(fā)送應答消息給客戶端
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelReadComplete");
/**
* 該方法的作用是將消息發(fā)送隊列中的消息寫入到 SocketChannel 中發(fā)送給對方
* 從性能角度考慮：為了防止頻繁地喚醒Selector進行消息發(fā)送，Netty的write方法并不直接將消息寫入SocketChannel中，
* 調(diào)用 write 方法只是把待發(fā)送的消息發(fā)到發(fā)送緩沖數(shù)組中，再通過調(diào)用 flush 方法，將發(fā)送緩沖區(qū)中的消息
* 全部寫到 SocketChannel 中
*/
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
cause.printStackTrace();
ctx.close();
}
}
NettyTimeClient
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;
public class NettyTimeClient {
private 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.SO_KEEPALIVE, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
// 發(fā)起異步連接操作
ChannelFuture future = bootstrap.connect(host, port).sync();
// 等待客戶端鏈路關閉
future.channel().closeFuture().sync();
} catch (Exception e) {
} finally {
group.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.parseInt(args[0]);
} catch (Exception e) {
// TODO: handle exception
}
}
new NettyTimeClient().connect(port, "127.0.0.1");
}
}
TimeClientHandler
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* 重點關注 channelActive channelRead exceptionCaught
*
*
*/
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
// private static final Logger.get
private final ByteBuf firstMessage;
public TimeClientHandler() {
byte[] req = "QUERY TIME ORDER".getBytes();
firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
}
/**
* 當客戶端和服務端TCP鏈路建立成功之后，Netty線程會調(diào)用 channelActive 方法
* 發(fā)送查詢事件的指令給服務端，調(diào)用 ChannelHandlerContext 的 writeAndFlush 方法將請求消息發(fā)送給服務端
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(firstMessage);
}
/**
* 當服務端返回應答消息時，channelRead 方法被調(diào)用，
*/
@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");
System.out.println("Now is :" + body);
}
/**
* 發(fā)生異常時，調(diào)用該方法，打印異常日志
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
