關?.NETty
Netty 是一個利用 JAVA 的高級網絡的能力,隱藏其背后的復雜性而提供一個易于使用的 API 的客戶端/服務器框架。
MAVEN依賴
<dependencies>
<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.36.Final</version>
</dependency>
</dependencies>
SpringBootApplication
啟動器中需要new一個NettyServer,并顯式調用啟動netty。
@SpringBootApplication
public class SpringCloudStudyDemoApplication {
public static void main(String[] args) {
SpringApplication.run(SpringCloudStudyDemoApplication.class,args);
try {
new NettyServer(12345).start();
System.out.println("https://blog.csdn.net/moshowgame");
System.out.println("http://127.0.0.1:6688/netty-websocket/index");
}catch(Exception e) {
System.out.println("NettyServerError:"+e.getMessage());
}
}
}
NettyServer
啟動的NettyServer,這里進行配置
/**
* NettyServer Netty服務器配置
* @author zhengkai.blog.csdn.net
*/
public class NettyServer {
private final int port;
public NettyServer(int port) {
this.port = port;
}
public void start() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap sb = new ServerBootstrap();
sb.option(ChannelOption.SO_BACKLOG, 1024);
sb.group(group, bossGroup) // 綁定線程池
.channel(NIOServerSocketChannel.class) // 指定使用的channel
.localAddress(this.port)// 綁定監聽端口
.childHandler(new ChannelInitializer<SocketChannel>() { // 綁定客戶端連接時候觸發操作
@Override
protected void initChannel(SocketChannel ch) throws Exception {
System.out.println("收到新連接");
//websocket協議本身是基于http協議的,所以這邊也要使用http解編碼器
ch.pipeline().addLast(new HttpServerCodec());
//以塊的方式來寫的處理器
ch.pipeline().addLast(new ChunkedWriteHandler());
ch.pipeline().addLast(new HttpObjectAggregator(8192));
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
ch.pipeline().addLast(new MyWebSocketHandler());
}
});
ChannelFuture cf = sb.bind().sync(); // 服務器異步創建綁定
System.out.println(NettyServer.class + " 啟動正在監聽: " + cf.channel().localAddress());
cf.channel().closeFuture().sync(); // 關閉服務器通道
} finally {
group.shutdownGracefully().sync(); // 釋放線程池資源
bossGroup.shutdownGracefully().sync();
}
}
}
MyChannelHandlerPool
通道組池,管理所有websocket連接
/**
* MyChannelHandlerPool
* 通道組池,管理所有websocket連接
* @author zhengkai.blog.csdn.net
*/
public class MyChannelHandlerPool {
public MyChannelHandlerPool(){}
public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}
MyWebSocketHandler
處理ws一下幾種情況:
- channelActive與客戶端建立連接
- channelInactive與客戶端斷開連接
- channelRead0客戶端發送消息處理
/**
* NettyServer Netty服務器配置
* @author zhengkai.blog.csdn.net
*/
public class NettyServer {
private final int port;
public NettyServer(int port) {
this.port = port;
}
public void start() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup group = new NioEventLoopGroup();
try {
ServerBootstrap sb = new ServerBootstrap();
sb.option(ChannelOption.SO_BACKLOG, 1024);
sb.group(group, bossGroup) // 綁定線程池
.channel(NioServerSocketChannel.class) // 指定使用的channel
.localAddress(this.port)// 綁定監聽端口
.childHandler(new ChannelInitializer<SocketChannel>() { // 綁定客戶端連接時候觸發操作
@Override
protected void initChannel(SocketChannel ch) throws Exception {
System.out.println("收到新連接");
//websocket協議本身是基于http協議的,所以這邊也要使用http解編碼器
ch.pipeline().addLast(new HttpServerCodec());
//以塊的方式來寫的處理器
ch.pipeline().addLast(new ChunkedWriteHandler());
ch.pipeline().addLast(new HttpObjectAggregator(8192));
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
ch.pipeline().addLast(new MyWebSocketHandler());
}
});
ChannelFuture cf = sb.bind().sync(); // 服務器異步創建綁定
System.out.println(NettyServer.class + " 啟動正在監聽: " + cf.channel().localAddress());
cf.channel().closeFuture().sync(); // 關閉服務器通道
} finally {
group.shutdownGracefully().sync(); // 釋放線程池資源
bossGroup.shutdownGracefully().sync();
}
}
}
socket.html
主要是連接ws,發送消息,以及消息反饋
<!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>Netty-Websocket</title>
<script type="text/JavaScript">
// by zhengkai.blog.csdn.net
var socket;
if(!window.WebSocket){
window.WebSocket = window.MozWebSocket;
}
if(window.WebSocket){
socket = new WebSocket("ws://127.0.0.1:12345/ws");
socket.onmessage = function(event){
var ta = document.getElementById('responseText');
ta.value += event.data+"rn";
};
socket.onopen = function(event){
var ta = document.getElementById('responseText');
ta.value = "Netty-WebSocket服務器。。。。。。連接 rn";
};
socket.onclose = function(event){
var ta = document.getElementById('responseText');
ta.value = "Netty-WebSocket服務器。。。。。。關閉 rn";
};
}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;">
<label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br />
<label>TEXT</label><input type="text" name="message" value="這里輸入消息" /> <br />
<br /> <input type="button" value="發送ws消息"
onClick="send(this.form.uid.value+':'+this.form.message.value)" />
<hr color="black" />
<h3>服務端返回的應答消息</h3>
<textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
</form>
</body>
</html>
Controller
寫好了html當然還需要一個controller來引導頁面。
@RestController
public class IndexController {
@GetMapping("/index")
public ModelAndView index(){
ModelAndView mav=new ModelAndView("socket");
mav.addObject("uid", RandomUtil.randomNumbers(6));
return mav;
}
}
效果演示
思路優化
由于netty不能像默認的websocket一樣設置一些PathVariable例如{uid}等參數(暫未發現可以,如果有發現歡迎補充),所以很多時候發送到后臺的報文可以設置一些特殊的格式,例如上文的004401:大家好,可以分解為userid:text,當然userid也可以是加密的一些報文,甚至可以學習其他報文一樣設置加密區,這取決于大家的業務需要. (已更新解決方案)
后言
項目已經整合進開源項目spring-cloud-study的子模塊spring-cloud-study-netty-websocket,作為對websocket體系的補充,對SpringBoot2.0集成WebSocket,實現后臺向前端推送信息 的完善。
改造netty支持url參數
最新改造的項目代碼已經上傳,克服了使用url會導致連接斷開的問題,詳情請看spring-cloud-study
首先,調整一下加載handler的順序,優先MyWebSocketHandler在WebSocketServerProtocolHandler之上。
ch.pipeline().addLast(new MyWebSocketHandler());
ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
其次,改造MyWebSocketHandler 的channelRead方法,首次連接會是一個FullHttpRequest類型,可以通過FullHttpRequest.uri()獲取完整ws的URL地址,之后接受信息的話,會是一個TextWebSocketFrame類型。
public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("與客戶端建立連接,通道開啟!");
//添加到channelGroup通道組
MyChannelHandlerPool.channelGroup.add(ctx.channel());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("與客戶端斷開連接,通道關閉!");
//添加到channelGroup 通道組
MyChannelHandlerPool.channelGroup.remove(ctx.channel());
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//首次連接是FullHttpRequest,處理參數 by zhengkai.blog.csdn.net
if (null != msg && msg instanceof FullHttpRequest) {
FullHttpRequest request = (FullHttpRequest) msg;
String uri = request.uri();
Map paramMap=getUrlParams(uri);
System.out.println("接收到的參數是:"+JSON.toJSONString(paramMap));
//如果url包含參數,需要處理
if(uri.contains("?")){
String newUri=uri.substring(0,uri.indexOf("?"));
System.out.println(newUri);
request.setUri(newUri);
}
}else if(msg instanceof TextWebSocketFrame){
//正常的TEXT消息類型
TextWebSocketFrame frame=(TextWebSocketFrame)msg;
System.out.println("客戶端收到服務器數據:" +frame.text());
sendAllMessage(frame.text());
}
super.channelRead(ctx, msg);
}
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {
}
private void sendAllMessage(String message){
//收到信息后,群發給所有channel
MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
}
private static Map getUrlParams(String url){
Map<String,String> map = new HashMap<>();
url = url.replace("?",";");
if (!url.contains(";")){
return map;
}
if (url.split(";").length > 0){
String[] arr = url.split(";")[1].split("&");
for (String s : arr){
String key = s.split("=")[0];
String value = s.split("=")[1];
map.put(key,value);
}
return map;
}else{
return map;
}
}
}
html中的ws地址也進行改造
socket = new WebSocket("ws://127.0.0.1:12345/ws?uid=666&gid=777");
改造后控制臺輸出情況
收到新連接
與客戶端建立連接,通道開啟!
接收到的參數是:{"uid":"666","gid":"777"}
/ws
客戶端收到服務器數據:142531:這里輸入消息
客戶端收到服務器數據:142531:這里輸入消息
客戶端收到服務器數據:142531:這里輸入消息
failed: WebSocket opening handshake timed out
聽說是ssl wss的情況下才會出現,來自 @around-gao 的解決方法:
把MyWebSocketHandler和WebSocketServerProtocolHandler調下順序就好了。
原文鏈接:
https://mp.weixin.qq.com/s/19ykCBRkg5J8JAyTPPNncg