Netty-WebSocket長連接配接推送服務

轉自：http://blog.csdn.net/sinosoft_fesco_12138/article/details/50380256

November 25, 2015

推送服務推送服務

幾種消息推送技術比較

AJAX輪詢輪詢：缺點，糟糕的使用者體驗；對伺服器壓力很大，并造成帶寬的極大浪費。

Comet：長連接配接機制，同樣由浏覽器端主動發起，但Server端以一種很慢的方式給出回應，優點是實時性好，缺點，長時間占用連結，喪失了無狀态高并發的缺點

Flash XML Socket，需用戶端安裝java虛拟機，Chrome浏覽器新版本已經預設屏蔽Flash

DWR開源架構,将java代碼轉為js檔案，引入js檔案路徑不能更改，原理也是輪詢，

WebSocket，HTML5定義，不支援html5的可以使用socket.io，高效節能，簡單易用，配合netty可以達到百萬級的連結推送。

NIO，java1.4版本，非阻塞IO

netty，基于原生NIO實作的高并發架構，配合websocket實作消息推送，netty會單獨開一個websocket端口處理請求，并不會占用中間件的連接配接數，而且一個線程可以處理幾萬個連結

NIO是什麼

NIO 是New IO 的簡稱，在jdk1.4 裡提供的新api 。Sun 官方标榜的特性如下：

為所有的原始類型提供為所有的原始類型提供(Buffer)緩存支援緩存支援。字元集編碼解碼解決方案。

Channel ：一個新的原始I/O 抽象。支援鎖和記憶體映射檔案的檔案通路接口。提供多路(non-bloking) 非阻塞式的高伸縮性網絡I/O 。

Channel 通道、Buffer 緩沖區、Selector 選擇器

其中Channel對應以前的流對應以前的流，Buffer不是什麼新東西，Selector是因為nio可以使用異步的非堵塞模式才加入的東西。以前的流總是堵塞的，一個線程隻要對它進行操作，其它操作就會被堵塞，也就相當于水管沒有閥門，你伸手接水的時候，不管水到了沒有，你就都隻能耗在接水（流）上。nio的Channel的加入，相當于增加了水龍頭（有閥門），雖然一個時刻也隻能接一個水管的水，但依賴輪換政策，在水量不大的時候，各個水管裡流出來的水，都可以得到妥善接納，這個關鍵之處就是增加了一個接水工，也就是Selector，他負責協調，也就是看哪根水管有水了的話，在目前水管的水接到一定程度的時候，就切換一下：臨時關上目前水龍頭，試着打開另一個水龍頭（看看有沒有水）。當其他人需要用水的時候，不是直接去接水，而是事前提了一個水桶給接水工，這個水桶就是Buffer。也就是，其他人雖然也可能要等，但不會在現場等，而是回家等，可以做其它事去，水接滿了，接水工會通知他們 Java NIO 實作百萬連接配接（這段代碼隻會接受連過來的連接，不做任何操作）

[java] view plain copy

ServerSocketChannel ssc = ServerSocketChannel.open();
Selector sel = Selector.open();
ssc.configureBlocking(false);
ssc.socket().bind(new InetSocketAddress(8080));
SelectionKey key = ssc.register(sel, SelectionKey.OP_ACCEPT);
while(true) {
sel.select();
Iterator it = sel.selectedKeys().iterator();
while(it.hasNext()) {
SelectionKey skey = (SelectionKey)it.next();
it.remove();
if(skey.isAcceptable()) {
ch = ssc.accept();
}
}
}

Netty是什麼

Netty: http://netty.io/ Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients

官方解釋已經夠清楚的了，其中最吸引人的就是高性能高性能了。

1，普通的伺服器10000個連接配接需要10000個線程，伺服器可能就直接卡住了，但對于netty伺服器可能幾個線程就夠了

2，netty是一套在是一套在java NIO的基礎上封裝的便于使用者開發網絡應用程式的的基礎上封裝的便于使用者開發網絡應用程式的api. 應用場景很多,諸如阿裡的消息隊列(RocketMQ),分布式rpc(Dubbo)通信層都使用到了netty(dubbo可以用服務發現自由選擇通信層)

3，netty是非阻塞事件驅動架構是非阻塞事件驅動架構, 并結合線程組(group)的概念,可以很好的支援高并發,慢連接配接的場景

4，程式設計接口非常容易,并且也較好的解決了TCP粘包/拆包的問題.netty提供了自己的ByteBuf和channel,相比于jdk的ByteBuffer和channel來說更簡便靈活操作, 并提供了pipeline的概念,并針對每個contextHandler都可以由使用者定義, 友善直接.

5，有一些web架構已經開始直接使用netty做為底層通信服務,諸如play. 這樣play就不用依賴于容器去進行部署,在沒有nginx做反向代理的情況下也能支援高并發.編解碼器可以随意擴充,今天是個web，明天就可以是一個ftp或email伺服器,個人覺得比較靈活。

6，相對于Tomcat這種Web Server（顧名思義主要是提供Web協定相關的服務的），Netty是一個是一個Network Server，是處于Web Server更下層的網絡框架，也就是說你可以使用Netty模仿Tomcat做一個提供HTTP服務的Web容器。簡而言之，Netty通過使用NIO的很多新特性，對TCP/UDP程式設計進行了簡化和封裝，提供了更容易使用的網絡程式設計接口，讓你可以根據自己的需要封裝獨特的HTTP Server活着FTP Server等.

Netty更多的連結更多的連結

非阻塞IO

其實無論是用Java NIO 還是用Netty，達到百萬連結都沒有任何難度，因為他們都是非阻塞的IO，不需要為每個連結建立一個線程了。

預知詳情，可以搜尋一下BIO,NIO,AIO的相關知識。

Netty 實作百萬連接配接（這段代碼隻會接受連過來的連接）

[java] view plain copy

NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workerGroup= new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup);
bootstrap.channel( NioServerSocketChannel.class);
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
//todo: add handler
}});
bootstrap.bind(8080).sync();

WebSocket是什麼

WebSocket是HTML5出的東西（協定），也就是說HTTP協定沒有變化，或者說沒關系，但HTTP是不支援持久連接配接的（長連接配接，循環連接配接的不算）

首先，Websocket是一個持久化的協定，相對于HTTP這種非持久的協定來說。簡單的舉個例子吧，用目前應用比較廣泛的PHP生命周期來解釋。

1) HTTP的生命周期通過Request來界定，也就是一個Request 一個Response，那麼在HTTP1.0中，這次HTTP請求就結束了。在HTTP1.1中進行了改進，使得有一個keep-alive，也就是說，在一個HTTP連接配接中，可以發送多個Request，接收多個Response。但是請記住 Request = Response ，在HTTP中永遠是這樣，也就是說一個request隻能有一個response。而且這個response也是被動的，不能主動發起。

Websocket的作用在講Websocket之前，我就順帶着講下 long poll 和 ajax輪詢的原理。

首先是 ajax輪詢，ajax輪詢的原理非常簡單，讓浏覽器隔個幾秒就發送一次請求，詢問伺服器是否有新資訊。

場景再現：用戶端：啦啦啦，有沒有新資訊(Request)

服務端：沒有（Response）

用戶端：啦啦啦，有沒有新資訊(Request)

服務端：沒有。。（Response）

用戶端：啦啦啦，有沒有新資訊(Request)

服務端：你好煩啊，沒有啊。。（Response）

用戶端：啦啦啦，有沒有新消息（Request）

服務端：好啦好啦，有啦給你。（Response）

用戶端：啦啦啦，有沒有新消息（Request）

服務端：。。。。。沒。。。。沒。。。沒有（Response） ---- loop

long poll long poll 其實原理跟 ajax輪詢差不多，都是采用輪詢的方式，不過采取的是阻塞模型（一直打電話，沒收到就不挂電話），也就是說，用戶端發起連接配接後，如果沒消息，就一直不傳回Response給用戶端。直到有消息才傳回，傳回完之後，用戶端再次建立連接配接，周而複始。

場景再現用戶端：啦啦啦，有沒有新資訊，沒有的話就等有了才傳回給我吧（Request）

服務端：額。。等待到有消息的時候。。來給你（Response）

用戶端：啦啦啦，有沒有新資訊，沒有的話就等有了才傳回給我吧（Request） -loop

從上面可以看出其實這兩種方式，都是在不斷地建立HTTP連接配接，然後等待服務端處理，可以展現HTTP協定的另外一個特點，被動性。何為被動性呢，其實就是，服務端不能主動聯系用戶端，隻能有用戶端發起。簡單地說就是，伺服器是一個很懶的冰箱（這是個梗）（不會、不能主動發起連接配接），但是上司有指令，如果有客戶來，不管多麼累都要好好接待。

說完這個，我們再來說一說上面的缺陷從上面很容易看出來，不管怎麼樣，上面這兩種都是非常消耗資源的。

ajax輪詢需要伺服器有很快的處理速度和資源。（速度） long poll 需要有很高的并發，也就是說同時接待客戶的能力。（場地大小）是以ajax輪詢和long poll 都有可能發生這種情況。

用戶端：啦啦啦啦，有新資訊麼？

服務端：月線正忙，請稍後再試（服務端：月線正忙，請稍後再試（503 Server Unavailable））

用戶端：。。。。好吧，啦啦啦，有新資訊麼？

服務端：月線正忙，請稍後再試（服務端：月線正忙，請稍後再試（503 Server Unavailable））

通過上面這個例子，我們可以看出，這兩種方式都不是最好的方式，需要很多資源。一種需要更快的速度，一種需要更多的'電話'。這兩種都會導緻'電話'的需求越來越高。哦對了，忘記說了HTTP還是一個狀态協定。通俗的說就是，伺服器因為每天要接待太多客戶了，是個健忘鬼健忘鬼，你一挂電話，他就把你的東西全忘光了，把你的東西全丢掉了。你第二次還得再告訴服務器一遍。

是以在這種情況下出現了，Websocket出現了。他解決了HTTP的這幾個難題。首先，被動性被動性，當伺服器完成協定更新後（HTTP->Websocket），服務端就可以主動推送資訊給用戶端啦。是以上面的情景可以做如下修改。

用戶端：啦啦啦，我要建立Websocket協定，需要的服務：chat，Websocket協定版本：17（HTTP Request）

服務端：ok，确認，已更新為Websocket協定（HTTP Protocols Switched）

用戶端：麻煩你有資訊的時候推送給我噢。。

服務端：ok，有的時候會告訴你的。

服務端：balabalabalabala

服務端：哈哈哈哈哈啊哈哈哈哈

服務端：笑死我了哈哈哈哈哈哈哈

浏覽器打開浏覽器打開websocket代碼代碼

[javascript] view plain copy

var socket;
if(!window.WebSocket){
window.WebSocket = window.MozWebSocket;
}
if(window.WebSocket){
socket = new WebSocket("ws://10.0.53.219:7397/websocket");
socket.onmessage = function(event){
var ta = document.getElementById('responseText');
ta.value += event.data+"\r\n";
};
socket.onopen = function(event){
var ta = document.getElementById('responseText');
ta.value = "這裡顯示伺服器推送資訊"+"\r\n";
};
socket.onclose = function(event){
var ta = document.getElementById('responseText');
ta.value = "";
ta.value = "WebSocket 關閉"+"\r\n";
};
}else{
alert("您的浏覽器不支援WebSocket協定！");
}

不支援websocket的浏覽器處理方法

點選打開連結 https://github.com/gimite/web-socket-js/

複制 swfobject.js, web_socket.js, WebSocketMain.swf到項目中

[javascript] view plain copy

<span style="font-size:10px;"><script type="text/javascript" src="swfobject.js"></script>
<script type="text/javascript" src="web_socket.js"></script>
<script type="text/javascript">
// Let the library know where WebSocketMain.swf is:
WEB_SOCKET_SWF_LOCATION = "WebSocketMain.swf";
// Write your code in the same way as for native WebSocket:
var ws = new WebSocket("ws://example.com:10081/");
ws.onopen = function() {
ws.send("Hello"); // Sends a message.
};
ws.onmessage = function(e) {
// Receives a message.
alert(e.data);
};
ws.onclose = function() {
alert("closed");
};
</script></span>

附錄：附錄：netty+websocket即時聊天代碼即時聊天代碼

Global.java

[java] view plain copy

package com.fesco.netty.websocket.common;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
public final class Global {
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}
ChildChannelHandler.java
package com.fesco.netty.websocket;
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;
public class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("http-codec", new HttpServerCodec());
ch.pipeline().addLast("aggregator",new HttpObjectAggregator(65535));
ch.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
ch.pipeline().addLast("handler",new MyWebSocketServerHandler());
}
}
MyWebSocketServerHandler.java
package com.fesco.netty.websocket;
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;
import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.logging.Level;
import java.util.logging.Logger;
import com.fesco.netty.websocket.common.Global;
public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object>{
private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName());
private WebSocketServerHandshaker handshaker;
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
Global.group.add(ctx.channel());
System.out.println("用戶端與伺服器段開啟");
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
Global.group.remove(ctx.channel());
System.out.println("用戶端與伺服器連結關閉！");
}
@Override
protected void messageReceived(ChannelHandlerContext ctx, Object msg)
throws Exception {
if(msg instanceof FullHttpRequest){
handleHttpRequest(ctx, ((FullHttpRequest) msg));
}else if(msg instanceof WebSocketFrame){
handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
}
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
private void handlerWebSocketFrame(ChannelHandlerContext ctx,
WebSocketFrame frame) {
if(frame instanceof CloseWebSocketFrame){
handshaker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
}
if(frame instanceof PingWebSocketFrame){
ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
}
if(!(frame instanceof TextWebSocketFrame)){
System.err.println("本例僅支援文本！");
throw new UnsupportedOperationException(String.format(
"%s frame types not supported", frame.getClass().getName()));
}
String request = ((TextWebSocketFrame)frame).text();
System.out.println("伺服器收到:" + request);
if (logger.isLoggable(Level.FINE)) {
logger.fine(String.format("%s received %s", ctx.channel(),request));
}
TextWebSocketFrame tws = new TextWebSocketFrame(getDateTime()
+ "(" +ctx.channel().remoteAddress() + ") ：" + request);
// 群發
Global.group.writeAndFlush(tws);
// 傳回【誰發的發給誰】
// ctx.channel().writeAndFlush(tws);
}
private void handleHttpRequest(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("ws://localhost:7397/websocket", null, false);
handshaker = wsFactory.newHandshaker(req);
if(null == handshaker){
WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
}else{
handshaker.handshake(ctx.channel(), req);
}
}
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();
}
ChannelFuture future = ctx.channel().writeAndFlush(res);
if (!isKeepAlive(req) || res.getStatus().code() != 200) {
future.addListener(ChannelFutureListener.CLOSE);
}
}
private static boolean isKeepAlive(FullHttpRequest req){
return false;
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
throws Exception {
cause.printStackTrace();
ctx.close();
}
private String getDateTime() {
// Calendar calendar = Calendar.getInstance();
Calendar calendar = new GregorianCalendar();
java.util.Date date = new java.util.Date();
calendar.setTime(date);
String sHour = null;
String sMinute = null;
String sSecond = null;
String sYear = null;
String sMonth = null;
String sDay = null;
int year = calendar.get(Calendar.YEAR);
int month = calendar.get(Calendar.MONTH) + 1;
int day = calendar.get(Calendar.DATE);
int hour = calendar.get(Calendar.HOUR_OF_DAY);
int minute = calendar.get(Calendar.MINUTE);
int second = calendar.get(Calendar.SECOND);
int milliSecond = calendar.get(Calendar.MILLISECOND);
sYear = String.valueOf(year);
if (month < 10) {
sMonth = "0" + month;
} else
sMonth = String.valueOf(month);
if (day < 10) {
sDay = "0" + day;
} else
sDay = String.valueOf(day);
if (hour < 10) {
sHour = "0" + hour;
} else {
sHour = String.valueOf(hour);
}
if (minute < 10) {
sMinute = "0" + minute;
} else {
sMinute = String.valueOf(minute);
}
if (second < 10) {
sSecond = "0" + second;
} else {
sSecond = String.valueOf(second);
}
return sYear + "-" + sMonth + "-" + sDay + " " + sHour + ":" + sMinute + ":" + sSecond;
}
}

NettyServer.java

[java] view plain copy

package com.fesco.netty.websocket;
import java.net.InetSocketAddress;
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;
public class NettyServer {
public void run(){
EventLoopGroup boosGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boosGroup, workGroup);
bootstrap.channel(NioServerSocketChannel.class);
bootstrap.childHandler(new ChildChannelHandler());
System.err.println("伺服器開啟待用戶端連結.....");
Channel ch = bootstrap.bind(new InetSocketAddress("10.0.53.219", 7397)).sync().channel();
ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally{
boosGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
public static void main(String[] args){
new NettyServer().run();
}
}

ClinetHtml.html

[html] view plain copy

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
<meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
<title>無标題文檔</title>
<script type="text/javascript">
var socket;
if(!window.WebSocket){
window.WebSocket = window.MozWebSocket;
}
if(window.WebSocket){
socket = new WebSocket("ws://10.0.53.219:7397/websocket");
socket.onmessage = function(event){
var ta = document.getElementById('responseText');
ta.value += event.data+"\r\n";
};
socket.onopen = function(event){
var ta = document.getElementById('responseText');
ta.value = "這裡顯示伺服器推送資訊"+"\r\n";
};
socket.onclose = function(event){
var ta = document.getElementById('responseText');
ta.value = "";
ta.value = "WebSocket 關閉"+"\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="發送 WebSocket 請求消息"
onClick="send(this.form.message.value)" />
<hr color="blue" />
<h3>服務端傳回的應答消息</h3>
<textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
</form>
</body>
</html>

Netty-WebSocket長連接配接推送服務

繼續閱讀

【超級推薦】HTTPS原理及WebSocket 原理

websocket簡單執行個體

websocket簡單介紹（一）

WebSocket簡單介紹（屬性+方法）

WebSocket 的簡單應用入門

WebSocket實際應用

基于websocket實作的一個簡單的聊天室

Java WebSocket實作網絡聊天室一、建立WebChatCopy項目并配置pom檔案

關于websocket小聊天室

springBoot整合WebSocket實作的聊天室demo，可向指定用戶端發送消息

websocket 實戰——好友聊天JS websocket 實戰——好友聊天

websocket redis實作叢集即時消息聊天websocket與redismq實作叢集消息聊天

SpringBoot整合Websocket實作即時聊天功能

WebSocket code 200 報錯

python websocket 有多快啟動服務socket伺服器 sample_server.py前端頁面 index.html效果圖連結挺可靠，10分鐘沒斷過平均每秒 10000 QOS 吞吐django http 不做優化好像 1000 不到