Netty是由JBOSS提供的一款基于NIO的客戶嫡锌、服務(wù)器端編程的Java開源框架，用以快速開發(fā)高性能琳钉、高可靠性的網(wǎng)絡(luò)服務(wù)器和客戶端程序势木。
轉(zhuǎn)載：http://www.reibang.com/p/331ee864bda9
BIO：同步阻塞IO
服務(wù)端采用一個(gè)獨(dú)立的Acceptor線程監(jiān)聽客戶端的連接，當(dāng)接收到客戶端連接請(qǐng)求后歌懒，為每個(gè)客戶端創(chuàng)建一個(gè)新的線程進(jìn)行鏈路處理啦桌，處理完成后通過輸出流返回給客戶端，然后銷毀線程及皂。屬于典型的一請(qǐng)求一應(yīng)答通信模型甫男，缺乏彈性伸縮能力且改。當(dāng)客戶端并發(fā)訪問量增加后，服務(wù)端的線程數(shù)同步增加板驳，由于線程是jvm非常寶貴的系統(tǒng)資源钾虐，線程數(shù)增加，系統(tǒng)的性能急劇下降笋庄，甚至?xí)斐啥褩Ｒ绯鲂ǎ瑒?chuàng)建線程失敗等問題。
偽異步IO
當(dāng)有新的客戶端接入的時(shí)候直砂，將客戶端socket封裝成Task到線程池中處理菌仁。線程池可以設(shè)置消息隊(duì)列的大小和最大線程數(shù)，因此它的資源占用是可控的静暂。但是當(dāng)有大量客戶端接入的時(shí)候會(huì)造成線程池阻塞济丘。
NIO：同步非阻塞IO
- 緩沖區(qū)Buffer：一塊連續(xù)的內(nèi)存塊，是 NIO 數(shù)據(jù)讀或?qū)懙闹修D(zhuǎn)地洽蛀。
- 通道Channel：數(shù)據(jù)的源頭或者數(shù)據(jù)的目的地摹迷，用于向 buffer 提供數(shù)據(jù)或者讀取 buffer 數(shù)據(jù) ,buffer 對(duì)象的唯一接口。支持異步 I/O 郊供。
- 多路復(fù)用器Selector：具有選擇就緒任務(wù)的能力峡碉。
Buffer作為IO流中數(shù)據(jù)的緩沖區(qū)，而Channel則作為socket的IO流與Buffer的傳輸通道驮审■昙模客戶端socket與服務(wù)端socket之間的IO傳輸不直接把數(shù)據(jù)交給CPU使用，而是先經(jīng)過Channel通道把數(shù)據(jù)保存到Buffer疯淫，然后CPU直接從Buffer區(qū)讀寫數(shù)據(jù)地来，一次可以讀寫更多的內(nèi)容。
nio.png
NIO是對(duì)BIO的改進(jìn)熙掺，基于Reactor模型未斑。NIO的最重要的地方是當(dāng)一個(gè)連接創(chuàng)建后，不需要對(duì)應(yīng)一個(gè)線程币绩，這個(gè)連接會(huì)被注冊(cè)到多路復(fù)用器上面蜡秽，所以所有的連接只需要一個(gè)線程就可以搞定，當(dāng)這個(gè)線程中的多路復(fù)用器進(jìn)行輪詢的時(shí)候类浪，發(fā)現(xiàn)連接上有請(qǐng)求的話载城，才開啟一個(gè)線程進(jìn)行處理肌似。
AIO：異步阻塞IO
AIO是對(duì)NIO的改進(jìn)费就，是基于Proactor模型。與NIO不同川队，當(dāng)進(jìn)行讀寫操作時(shí)力细，只須直接調(diào)用API的read或write方法即可睬澡。這兩種方法均為異步的，對(duì)于讀操作而言眠蚂，當(dāng)有流可讀取時(shí)煞聪，操作系統(tǒng)會(huì)將可讀的流傳入read方法的緩沖區(qū)，并通知應(yīng)用程序逝慧；對(duì)于寫操作而言昔脯，當(dāng)操作系統(tǒng)將write方法傳遞的流寫入完畢時(shí)，操作系統(tǒng)主動(dòng)通知應(yīng)用程序笛臣。較之NIO而言云稚，AIO一方面簡(jiǎn)化了程序出的編寫，流的讀取和寫入都由操作系統(tǒng)來代替完成沈堡；另一方面省去了NIO中程序要遍歷事件通知隊(duì)列（selector）的代價(jià)静陈。Windows基于[IOCP(http://en.wikipedia.org/wiki/Input/output_completion_port)實(shí)現(xiàn)了AIO，Linux目前只有基于epoll實(shí)現(xiàn)的AIO诞丽。
Java對(duì)BIO鲸拥、NIO、AIO的支持：
- BIO ： 同步并阻塞僧免，服務(wù)器實(shí)現(xiàn)模式為一個(gè)連接一個(gè)線程刑赶，即客戶端有連接請(qǐng)求時(shí)服務(wù)器端就需要啟動(dòng)一個(gè)線程進(jìn)行處理，如果這個(gè)連接不做任何事情會(huì)造成不必要的線程開銷懂衩，當(dāng)然可以通過線程池機(jī)制改善角撞。
- NIO ： 同步非阻塞，服務(wù)器實(shí)現(xiàn)模式為一個(gè)請(qǐng)求一個(gè)線程勃痴，即客戶端發(fā)送的連接請(qǐng)求都會(huì)注冊(cè)到多路復(fù)用器上谒所，多路復(fù)用器輪詢到連接有I/O請(qǐng)求時(shí)才啟動(dòng)一個(gè)線程進(jìn)行處理。
- AIO(NIO.2) ： 異步非阻塞沛申，服務(wù)器實(shí)現(xiàn)模式為一個(gè)有效請(qǐng)求一個(gè)線程劣领，客戶端的I/O請(qǐng)求都是由OS先完成了再通知服務(wù)器應(yīng)用去啟動(dòng)線程進(jìn)行處理。
BIO铁材、NIO尖淘、AIO適用場(chǎng)景分析:
- BIO方式適用于連接數(shù)目比較小且固定的架構(gòu)，這種方式對(duì)服務(wù)器資源要求比較高著觉，并發(fā)局限于應(yīng)用中村生，JDK1.4以前的唯一選擇，但程序直觀簡(jiǎn)單易理解饼丘。
- NIO方式適用于連接數(shù)目多且連接比較短（輕操作）的架構(gòu)趁桃，比如聊天服務(wù)器，并發(fā)局限于應(yīng)用中，編程比較復(fù)雜卫病，JDK1.4開始支持油啤。
- AIO方式使用于連接數(shù)目多且連接比較長(zhǎng)（重操作）的架構(gòu)，比如相冊(cè)服務(wù)器蟀苛，充分調(diào)用OS參與并發(fā)操作益咬，編程比較復(fù)雜，JDK7開始支持帜平。
WebSocket
WebSocket是HTML5開始提供的一種在單個(gè) TCP 連接上進(jìn)行全雙工通訊的協(xié)議幽告。在WebSocket API中，瀏覽器和服務(wù)器只需要做一個(gè)握手的動(dòng)作裆甩，然后评腺，瀏覽器和服務(wù)器之間就形成了一條快速通道。兩者之間就直接可以數(shù)據(jù)互相傳送淑掌。瀏覽器通過 JavaScript 向服務(wù)器發(fā)出建立 WebSocket 連接的請(qǐng)求蒿讥，連接建立以后，客戶端和服務(wù)器端就可以通過 TCP 連接直接交換數(shù)據(jù)抛腕。當(dāng)你獲取 Web Socket 連接后芋绸，你可以通過 send() 方法來向服務(wù)器發(fā)送數(shù)據(jù)，并通過 onmessage 事件來接收服務(wù)器返回的數(shù)據(jù)担敌。
當(dāng)WebSocket的客戶端與服務(wù)端通信以后摔敛，就不需要之前握手請(qǐng)求HTTP協(xié)議的參與了
WebSocket的優(yōu)點(diǎn)
- 節(jié)省通信開銷（HttpRequest中的head很長(zhǎng)，占用帶寬和資源）
- 服務(wù)器可以主動(dòng)傳送數(shù)據(jù)給客戶端
- 實(shí)時(shí)通信
因?yàn)镠TTP 協(xié)議是一種無狀態(tài)的全封、無連接的马昙、單向的應(yīng)用層協(xié)議。它采用了請(qǐng)求/響應(yīng)模型刹悴。通信請(qǐng)求只能由客戶端發(fā)起行楞，服務(wù)端對(duì)請(qǐng)求做出應(yīng)答處理。這種通信模型有一個(gè)弊端：HTTP 協(xié)議無法實(shí)現(xiàn)服務(wù)器主動(dòng)向客戶端發(fā)起消息土匀。如果服務(wù)器有連續(xù)的狀態(tài)變化子房，客戶端要獲知就非常麻煩。大多數(shù) Web 應(yīng)用程序?qū)⑼ㄟ^頻繁的異步JavaScript和XML（AJAX）請(qǐng)求實(shí)現(xiàn)長(zhǎng)輪詢就轧。輪詢的效率低证杭，非常浪費(fèi)資源（因?yàn)楸仨毑煌＿B接，或者 HTTP 連接始終打開）妒御。
WebSocket的生命周期
WebSocket端點(diǎn)生命周期的第一個(gè)事件是打開通知解愤，它用來指示到WebSocket會(huì)話另一端的連接已經(jīng)建立。一旦打開通知被WebSocket對(duì)話的兩端都接收到乎莉，參與的任意WebSocket后續(xù)就可以發(fā)送消息了送讲。在WebSocket對(duì)話期間奸笤，可能會(huì)出現(xiàn)一些消息傳遞的錯(cuò)誤。接受消息的WebSocket端點(diǎn)本身就可能產(chǎn)生錯(cuò)誤李茫，或者WebSocket實(shí)現(xiàn)本身在某些情況下也會(huì)產(chǎn)生錯(cuò)誤。要注意對(duì)錯(cuò)誤的處理肥橙。不管在WebSocket對(duì)話的哪一端準(zhǔn)備結(jié)束對(duì)話魄宏，他都可以初始化關(guān)閉事件。下面從Java組件的視角來看看其生命周期如何呈現(xiàn)存筏。
- 打開事件：＠OnOpen 此事件發(fā)生在端點(diǎn)上建立新連接時(shí)并且在任何其他事件發(fā)生之前
- 消息事件：@OnMessage 此事件接收WebSocket對(duì)話中另一端發(fā)送的消息宠互。
- 錯(cuò)誤事件：@OnError 此事件在WebSocket連接或者端點(diǎn)發(fā)生錯(cuò)誤時(shí)產(chǎn)生
- 關(guān)閉事件：@OnClose 此事件表示W(wǎng)ebSocket端點(diǎn)的連接目前部分地關(guān)閉，它可以由參與連接的任意一個(gè)端點(diǎn)發(fā)出
Netty實(shí)現(xiàn)WebSocket
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
/**
* 存儲(chǔ)整個(gè)工程的全局配置
*/
public class NettyConfig {
//存儲(chǔ)每一個(gè)客戶端接入進(jìn)來時(shí)的channel對(duì)象
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}
import java.util.Date;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;
/**
* 接收/處理/響應(yīng)客戶端websocket請(qǐng)求的核心業(yè)務(wù)處理類
*/
public class MyWebSocketHandler extends SimpleChannelInboundHandler<Object> {
private WebSocketServerHandshaker handshaker;
private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket";
//客戶端與服務(wù)端創(chuàng)建連接的時(shí)候調(diào)用
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
NettyConfig.group.add(ctx.channel());
System.out.println("客戶端與服務(wù)端連接開啟...");
}
//客戶端與服務(wù)端斷開連接的時(shí)候調(diào)用
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
NettyConfig.group.remove(ctx.channel());
System.out.println("客戶端與服務(wù)端連接關(guān)閉...");
}
//服務(wù)端接收客戶端發(fā)送過來的數(shù)據(jù)結(jié)束之后調(diào)用
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
//工程出現(xiàn)異常的時(shí)候調(diào)用
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
//服務(wù)端處理客戶端websocket請(qǐng)求的核心方法
@Override
protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception {
//處理客戶端向服務(wù)端發(fā)起http握手請(qǐng)求的業(yè)務(wù)
if (msg instanceof FullHttpRequest) {
handHttpRequest(context, (FullHttpRequest)msg);
}else if (msg instanceof WebSocketFrame) { //處理websocket連接業(yè)務(wù)
handWebsocketFrame(context, (WebSocketFrame)msg);
}
}
// 處理客戶端與服務(wù)端之前的websocket業(yè)務(wù)
private void handWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){
//判斷是否是關(guān)閉websocket的指令
if (frame instanceof CloseWebSocketFrame) {
handshaker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
}
//判斷是否是ping消息
if (frame instanceof PingWebSocketFrame) {
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
return;
}
//判斷是否是二進(jìn)制消息椭坚，如果是二進(jìn)制消息予跌，拋出異常
if( ! (frame instanceof TextWebSocketFrame) ){
System.out.println("目前我們不支持二進(jìn)制消息");
throw new RuntimeException("【"+this.getClass().getName()+"】不支持消息");
}
//返回應(yīng)答消息
//獲取客戶端向服務(wù)端發(fā)送的消息
String request = ((TextWebSocketFrame) frame).text();
System.out.println("服務(wù)端收到客戶端的消息====>>>" + request);
TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + " ===>>> " + request);
//服務(wù)端向每個(gè)連接上來的客戶端群發(fā)消息
NettyConfig.group.writeAndFlush(tws);
}
// 處理客戶端向服務(wù)端發(fā)起http握手請(qǐng)求的業(yè)務(wù)
private void handHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req){
if (!req.getDecoderResult().isSuccess()
|| ! ("websocket".equals(req.headers().get("Upgrade")))) {
sendHttpResponse(ctx, req,
new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
return;
}
WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
WEB_SOCKET_URL, null, false);
handshaker = wsFactory.newHandshaker(req);
if (handshaker == null) {
WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
}else{
handshaker.handshake(ctx.channel(), req);
}
}
// 服務(wù)端向客戶端響應(yīng)消息
private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req,
DefaultFullHttpResponse res){
if (res.getStatus().code() != 200) {
ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
res.content().writeBytes(buf);
buf.release();
}
//服務(wù)端向客戶端發(fā)送數(shù)據(jù)
ChannelFuture f = ctx.channel().writeAndFlush(res);
if (res.getStatus().code() != 200) {
f.addListener(ChannelFutureListener.CLOSE);
}
}
}
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;
/**
* 初始化連接時(shí)候的各個(gè)組件
*/
public class MyWebSocketChannelHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel e) throws Exception {
e.pipeline().addLast("http-codec", new HttpServerCodec());
e.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
e.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
e.pipeline().addLast("handler", new MyWebSocketHandler());
}
}
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
/**
* 程序的入口，負(fù)責(zé)啟動(dòng)應(yīng)用
*/
public class Main {
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new MyWebSocketChannelHandler());
System.out.println("服務(wù)端開啟等待客戶端連接....");
Channel ch = b.bind(8888).sync().channel();
ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally{
//優(yōu)雅的退出程序
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
<html>
<head>
<meta http-equiv="Content-Type" content="text/html; charset = utf-8"/>
<title>WebSocket客戶端</title>
<script type="text/javascript">
var socket;
if(!window.WebSocket){
window.WebSocket = window.MozWebSocket;
}
if(window.WebSocket){
socket = new WebSocket("ws://localhost:8888/websocket");
socket.onmessage = function(event){
var ta = document.getElementById('responseContent');
ta.value += event.data + "\r\n";
};
socket.onopen = function(event){
var ta = document.getElementById('responseContent');
ta.value = "你當(dāng)前的瀏覽器支持WebSocket,請(qǐng)進(jìn)行后續(xù)操作\r\n";
};
socket.onclose = function(event){
var ta = document.getElementById('responseContent');
ta.value = "";
ta.value = "WebSocket連接已經(jīng)關(guān)閉\r\n";
};
}else{
alert("您的瀏覽器不支持WebSocket");
}
function send(message){
if(!window.WebSocket){
return;
}
if(socket.readyState == WebSocket.OPEN){
socket.send(message);
}else{
alert("WebSocket連接沒有建立成功Ｉ凭ァ券册！");
}
}
</script>
</head>
<body>
<form onSubmit="return false;">
<input type = "text" name = "message" value = ""/>
<br/><br/>
<input type = "button" value = "發(fā)送WebSocket請(qǐng)求消息" onClick = "send(this.form.message.value)"/>
<hr color="red"/>
<h2>客戶端接收到服務(wù)端返回的應(yīng)答消息</h2>
<textarea id = "responseContent" style = "width:1024px; height:300px"></textarea>
</form>
</body>
</html>
參考
netty-all-5.0.0.Alpha1.jar
WebSocket 詳解教程
SpringBoot 開發(fā)WebSocket
