基於Netty的時間服務器程序代碼
阿新 • • 發佈：2018-02-11
異步 大小 fin warning ole 網絡 字節數 日誌 oot [toc]
基於Netty的時間服務器程序代碼
程序代碼來自於《Netty權威指南》第三章，不過我都加了註釋，所以看起來會非常好理解。需要註意的是，《Netty權威指南》中TimeServerHandler類繼承的是ChannelHandlerAdapter，因為其使用的是Netty 5.x的版本，該版本現在已經被官方廢棄，而我使用的是Netty 4.x的，所以如果繼承ChannelHandlerAdapter，啟動程序時不會報錯，但是無法正常工作，即handler不生效這點尤其需要註意。
服務端代碼
TimeServer.java:
package cn.xpleaf.netty; 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; public class TimeServer { /** * 綁定端口號，啟動Netty服務端 * @param port * @throws Exception */ public void bind(int port) throws Exception { // NioEventLoopGroup是個線程組，它包含了一組NIO線程，專門用於網絡事件的處理 // 實際上它們就是Reactor線程組，關於Reactor，這是一種設計模型， // Reactor模型就是將消息放到了一個隊列中，通過異步線程池對其進行消費 // 在Netty中就是，NioEventLoopGroup就是Reactor，ChannelHandler就是Reactor模型中的handler // 可以參考文章：http://www.ivaneye.com/2016/07/23/iomodel.html // 這裏創建了兩個線程組，一個用於服務端接受客戶端的連接，一個用於進行SocketChannel的網絡讀寫 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { // 創建NIO服務端的輔助啟動類，目的是降低服務端的開發復雜度 ServerBootstrap b = new ServerBootstrap(); // 將兩個NIO線程組作為參數傳遞到ServerBootstrap中 b.group(bossGroup, workerGroup) // NioServerSocketChannel對應JDK NIO類庫中的ServerSocketChannel .channel(NioServerSocketChannel.class) // 設置NioServerSocketChannel的TCP參數 .option(ChannelOption.SO_BACKLOG, 1024) // 綁定I/O事件的處理類ChildChannelHandler，它的作用類似於Reactor模式中的Handler類 // 主要用於處理網絡I/O事件，例如記錄日誌、對消息進行編解碼等 .childHandler(new ChildChannelHandler()); // 綁定端口，同步等待成功，該方法是同步阻塞的，綁定成功後返回一個ChannelFuture ChannelFuture f = b.bind(port).sync(); // 等待服務端監聽端口關閉，阻塞，等待服務端鏈路關閉之後main函數才退出 f.channel().closeFuture().sync(); } finally { // 優雅退出，釋放線程池資源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } /** * 實現initChannel方法，其作用是當創建NioServerSocketChannel成功之後， * 在進行初始化是，將它的ChannelHandler設置到ChannelPipeline中，用於處理網絡I/O事件 * @author yeyonghao * */ private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new TimeServerHandler()); } } public static void main(String[] args) throws Exception { int port = 8080; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // TODO: handle exception } } new TimeServer().bind(port); } }
TimeServerHandler.java:
package cn.xpleaf.netty; import java.sql.Date; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; /** * 這裏需要註意，《Netty權威指南》中TimeServerHandler類繼承的是ChannelHandlerAdapter， * 因為其使用的是Netty 5.x的版本，該版本現在已經被官方廢棄，而我使用的是Netty 4.x的， * 所以如果繼承ChannelHandlerAdapter，啟動程序時不會報錯，但是無法正常工作，即handler不生效 * 這點尤其需要註意。 * @author yeyonghao * */ public class TimeServerHandler extends ChannelInboundHandlerAdapter { /** * 對網絡事件進行寫操作 */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { // Netty中的ByteBuf類似於nio類庫中的ByteBuffer，不過功能更強大 ByteBuf buf = (ByteBuf) msg; // 創建與buf一樣大小的字節數組 byte[] req = new byte[buf.readableBytes()]; // 將數據從buf中復制到req中 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對象 ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes()); // 通過ChannelHandlerContext的write方法異步發送給客戶端 ctx.write(resp); } /** * 寫操作完成之後進行的操作 */ @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { // ctx.write()只是將消息放到發送緩沖數組中，調用flush就將其發送緩沖區中的消息全部寫到SocketChannel中 ctx.flush(); } /** * 釋放資源 */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ctx.close(); } }
客戶端代碼
TimeClient.java:
package cn.xpleaf.netty; 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 TimeClient { public void connect(int port, String host) throws Exception { // 配置客戶端NIO線程組 EventLoopGroup group = new NioEventLoopGroup(); try { // 客戶端是Bootstrap，註意客戶端的bootstrap只能傳入一個NIO線程組 Bootstrap b = new Bootstrap(); // 客戶端是NioSocketChannel b.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) // 與服務端一樣的，只不過這裏使用了匿名內部類 .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ch.pipeline().addLast(new TimeClientHandler()); } }); // 發起異步連接操作（註意服務端是bind，客戶端則需要connect） ChannelFuture f = b.connect(host, port).sync(); // 等待客戶端鏈路關閉 f.channel().closeFuture().sync(); } 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(port); } catch (NumberFormatException e) { // 采用默認值 } } new TimeClient().connect(port, "localhost"); } }
TimeClientHandler.java:
package cn.xpleaf.netty;
import java.util.logging.Logger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
/**
* 該類的方法與功能與服務端的handler類似
* @author yeyonghao
*
*/
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger logger = Logger.getLogger(TimeServerHandler.class.getName());
private ByteBuf firstMessage;
/**
* 初始化需要發送的數據
*/
public TimeClientHandler() {
byte[] req = "QUERY TIME ORDER".getBytes();
firstMessage = Unpooled.buffer(req.length);
firstMessage.writeBytes(req);
}
/**
* 當客戶端和服務端TCP鏈路建立成功之後，Netty的NIO線程會調用channelActive方法
*/
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 將請求消息發送給服務端
ctx.writeAndFlush(firstMessage);
}
@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);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
logger.warning("Unexpected exception from downstream : ");
ctx.close();
}
}基於Netty的時間服務器程序代碼