netty入門(一)
Netty是一個(gè)高性能 事件驅(qū)動(dòng)的異步的非堵塞的IO(NIO)框架腊嗡，用于建立TCP等底層的連接春塌，基于Netty可以建立高性能的Http服務(wù)器唬涧。
1、首先來復(fù)習(xí)下非堵塞IO(NIO)
NIO這個(gè)庫是在JDK1.4中才引入的约炎。NIO和IO有相同的作用和目的，但實(shí)現(xiàn)方式不同蟹瘾，NIO主要用到的是塊圾浅，所以NIO的效率要比IO高很多。
在Java API中提供了兩套NIO憾朴，一套是針對(duì)標(biāo)準(zhǔn)輸入輸出NIO狸捕，另一套就是網(wǎng)絡(luò)編程N(yùn)IO。
****Buffer和Channel是標(biāo)準(zhǔn)NIO中的核心對(duì)象
Channel是對(duì)原IO中流的模擬伊脓，任何來源和目的數(shù)據(jù)都必須通過一個(gè)Channel對(duì)象府寒。一個(gè)Buffer實(shí)質(zhì)上是一個(gè)容器對(duì)象魁衙，發(fā)給Channel的所有對(duì)象都必須先放到Buffer中；同樣的株搔，從Channel中讀取的任何數(shù)據(jù)都要讀到Buffer中剖淀。
網(wǎng)絡(luò)編程N(yùn)IO中還有一個(gè)核心對(duì)象Selector，它可以注冊(cè)到很多個(gè)Channel上纤房，監(jiān)聽各個(gè)Channel上發(fā)生的事件纵隔，并且能夠根據(jù)事件情況決定Channel讀寫。這樣炮姨，通過一個(gè)線程管理多個(gè)Channel捌刮，就可以處理大量網(wǎng)絡(luò)連接了。
Selector 就是注冊(cè)對(duì)各種 I/O 事件興趣的地方舒岸，而且當(dāng)那些事件發(fā)生時(shí)绅作，就是這個(gè)對(duì)象告訴你所發(fā)生的事件。
Selector selector = Selector.open(); //創(chuàng)建一個(gè)selector
為了能讓Channel和Selector配合使用蛾派，我們需要把Channel注冊(cè)到Selector上俄认。通過調(diào)用channel.register（）方法來實(shí)現(xiàn)注冊(cè)：
channel.configureBlocking(false); //設(shè)置成異步IO
SelectionKey key =channel.register(selector,SelectionKey.OP_READ); //對(duì)所關(guān)心的事件進(jìn)行注冊(cè)（connet,accept,read,write）
SelectionKey 代表這個(gè)通道在此 Selector 上的這個(gè)注冊(cè)。
2洪乍、異步
CallBack：回調(diào)是異步處理經(jīng)常用到的編程模式眯杏，回調(diào)函數(shù)通常被綁定到一個(gè)方法上，并且在方法完成之后才執(zhí)行壳澳，這種處理方式在javascript當(dāng)中得到了充分的運(yùn)用岂贩。回調(diào)給我們帶來的難題是當(dāng)一個(gè)問題處理過程中涉及很多回調(diào)時(shí)巷波，代碼是很難讀的萎津。
Futures：Futures是一種抽象，它代表一個(gè)事情的執(zhí)行過程中的一些關(guān)鍵點(diǎn)褥紫，我們通過Future就可以知道任務(wù)的執(zhí)行情況姜性，比如當(dāng)任務(wù)沒完成時(shí)我們可以做一些其它事情。它給我們帶來的難題是我們需要去判斷future的值來確定任務(wù)的執(zhí)行狀態(tài)髓考。
3部念、netty到底怎么工作的呢？
直接來看個(gè)最簡單的實(shí)例
服務(wù)器端
public class EchoServer {
private final static int port = 8007;
public void start() throws InterruptedException{
ServerBootstrap bootstrap = new ServerBootstrap(); //引導(dǎo)輔助程序
EventLoopGroup group = new NioEventLoopGroup(); //通過nio的方式接受連接和處理連接
try {
bootstrap.group(group)
.channel(NioServerSocketChannel.class) //設(shè)置nio類型的channel
.localAddress(new InetSocketAddress(port)) //設(shè)置監(jiān)聽端口
.childHandler(new ChannelInitializer<SocketChannel>() { //有連接到達(dá)時(shí)會(huì)創(chuàng)建一個(gè)channel
// pipline 管理channel中的handler,在channel隊(duì)列中添加一個(gè)handler來處理業(yè)務(wù)
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("myHandler", new EchoServerHandler());
//ch.pipeline().addLast("idleStateHandler",new IdleStateHandler(0, 0, 180));
}
});
ChannelFuture future = bootstrap.bind().sync(); //配置完成氨菇，綁定server儡炼，并通過sync同步方法阻塞直到綁定成功
System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
future.channel().closeFuture().sync(); //應(yīng)用程序會(huì)一直等待，直到channel關(guān)閉
} catch (Exception e) {
e.getMessage();
}finally {
group.shutdownGracefully().sync();
}
}
- 創(chuàng)建一個(gè)ServerBootstrap實(shí)例
- 創(chuàng)建一個(gè)EventLoopGroup來處理各種事件查蓉，如處理鏈接請(qǐng)求乌询，發(fā)送接收數(shù)據(jù)等。
- 定義本地InetSocketAddress( port)好讓Server綁定
- 創(chuàng)建childHandler來處理每一個(gè)鏈接請(qǐng)求
- 所有準(zhǔn)備好之后調(diào)用ServerBootstrap.bind()方法綁定Server
handler 處理核心業(yè)務(wù)
@Sharable //注解@Sharable可以讓它在channels間共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
@Override
public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception {
ByteBuf buf = ctx.alloc().buffer();
buf.writeBytes("Hello World".getBytes());
ctx.write(buf);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
客戶端 連接
public class EchoClient {
private final int port;
private final String hostIp;
public EchoClient(int port, String hostIp) {
this.port = port;
this.hostIp = hostIp;
}
public void start() throws InterruptedException {
Bootstrap bootstrap = new Bootstrap();
EventLoopGroup group = new NioEventLoopGroup();
try {
bootstrap.group(group).channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(hostIp, port))
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new EchoClientHandler());
}
});
ChannelFuture future = bootstrap.connect().sync();
future.addListener(new ChannelFutureListener() {
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) {
System.out.println("client connected");
} else {
System.out.println("server attemp failed");
future.cause().printStackTrace();
}
}
});
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully().sync();
}
}
客戶端handler
@Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf>{
/**
*此方法會(huì)在連接到服務(wù)器后被調(diào)用
* */
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("Netty rocks!");
ctx.write(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));
}
/**
* 接收到服務(wù)器數(shù)據(jù)時(shí)調(diào)用
*/
@Override
protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
System.out.println("Client received: " + ByteBufUtil.hexDump(msg.readBytes(msg.readableBytes())));
}
/**
*捕捉到異常
* */
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
