基于netty的websocket例子

nettyServer

package com.atguigu.netty.websocket;
import javax.annotation.PostConstruct;
import org.springframework.stereotype.Service;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;
/**
* ClassName:NettyServer 注解式随spring啟動
* Function: TODO ADD FUNCTION.
* @author fwz
*/
@Service
public class NettyServer {
public static void main(String[] args) {
new NettyServer().run();
}
@PostConstruct
public void initNetty(){
new Thread(){
public void run() {
new NettyServer().run();
}
}.start();
}
public void run(){
System.out.println("===========================Netty端口啟動========");
// Boss線程：由這個線程池提供的線程是boss種類的，用于建立、連接配接、綁定socket， （有點像門衛）然後把這些socket傳給worker線程池。
// 在伺服器端每個監聽的socket都有一個boss線程來處理。在用戶端，隻有一個boss線程來處理所有的socket。
EventLoopGroup bossGroup = new NioEventLoopGroup();
// Worker線程：Worker線程執行所有的異步I/O，即處理操作
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
// ServerBootstrap 啟動NIO服務的輔助啟動類,負責初始話netty伺服器，并且開始監聽端口的socket請求
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workGroup);
// 設定非阻塞,用它來建立新accept的連接配接,用于構造serversocketchannel的工廠類
b.channel(NioServerSocketChannel.class);
// ChildChannelHandler 對出入的資料進行的業務操作,其繼承ChannelInitializer
b.childHandler(new ChildChannelHandler());
System.out.println("服務端開啟等待用戶端連接配接 ... ...");
Channel ch = b.bind(8081).sync().channel();
ch.closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally{
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}


private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
@Override
protected void initChannel(SocketChannel e) throws Exception {
// 設定30秒沒有讀到資料，則觸發一個READER_IDLE事件。
// pipeline.addLast(new IdleStateHandler(30, 0, 0));
// HttpServerCodec：将請求和應答消息解碼為HTTP消息
e.pipeline().addLast("http-codec",new HttpServerCodec());
// HttpObjectAggregator：将HTTP消息的多個部分合成一條完整的HTTP消息
e.pipeline().addLast("aggregator",new HttpObjectAggregator(65536));
// ChunkedWriteHandler：向用戶端發送HTML5檔案
e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
// 在管道中添加我們自己的接收資料實作方法
e.pipeline().addLast("handler",new MyWebSocketServerHandler());
}
}

}

MyWebSocketServerHandler

package com.atguigu.netty.websocket;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
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.HttpHeaderNames;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
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.AttributeKey;
import io.netty.util.CharsetUtil;

/**
 * ClassName:MyWebSocketServerHandler Function: TODO ADD FUNCTION.
 *
 * @author fwz
 */
public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName());
    private WebSocketServerHandshaker handshaker;

    /**
     * channel 通道 action 活躍的 當用戶端主動連結服務端的連結後，這個通道就是活躍的了。也就是用戶端與服務端建立了通信通道并且可以傳輸資料
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
// 添加
        Global.group.add(ctx.channel());
        System.out.println("用戶端與服務端連接配接開啟：" + ctx.channel().remoteAddress().toString());
    }

    /**
     * channel 通道 Inactive 不活躍的
     * 當用戶端主動斷開服務端的連結後，這個通道就是不活躍的。也就是說用戶端與服務端關閉了通信通道并且不可以傳輸資料
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
// 移除
        Global.group.remove(ctx.channel());
        System.out.println("用戶端與服務端連接配接關閉：" + ctx.channel().remoteAddress().toString());
    }

    /**
     * 接收用戶端發送的消息 channel 通道 Read 讀
     * 簡而言之就是從通道中讀取資料，也就是服務端接收用戶端發來的資料。但是這個資料在不進行解碼時它是ByteBuf類型的
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// 傳統的HTTP接入
        if (msg instanceof FullHttpRequest) {
            handleHttpRequest(ctx, ((FullHttpRequest) msg));
// WebSocket接入
        } else if (msg instanceof WebSocketFrame) {
            System.out.println(handshaker.uri());
            if ("anzhuo".equals(ctx.channel().attr(AttributeKey.valueOf("type")).get())) {
                handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
            } else {
                handlerWebSocketFrame2(ctx, (WebSocketFrame) msg);
            }
        }
    }

    /**
     * channel 通道 Read 讀取 Complete 完成 在通道讀取完成後會在這個方法裡通知，對應可以做重新整理操作 ctx.flush()
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
// 判斷是否關閉鍊路的指令
        if (frame instanceof CloseWebSocketFrame) {
            System.out.println(1);
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
// 判斷是否ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
// 本例程僅支援文本消息，不支援二進制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            System.out.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(new Date().toString() + ctx.channel().id() + "：" + request);
// 群發
        Global.group.writeAndFlush(tws);
// 傳回【誰發的發給誰】
// ctx.channel().writeAndFlush(tws);
    }

    private void handlerWebSocketFrame2(ChannelHandlerContext ctx, WebSocketFrame frame) {
// 判斷是否關閉鍊路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
// 判斷是否ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
// 本例程僅支援文本消息，不支援二進制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            System.out.println("本例程僅支援文本消息，不支援二進制消息");
            throw new UnsupportedOperationException(
                    String.format("%s frame types not supported", frame.getClass().getName()));
        }
// 傳回應答消息
        String request = ((TextWebSocketFrame) frame).text();
        System.out.println("服務端2收到：" + request);
        if (logger.isLoggable(Level.FINE)) {
            logger.fine(String.format("%s received %s", ctx.channel(), request));
        }
        TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);
// 群發
        Global.group.writeAndFlush(tws);
// 傳回【誰發的發給誰】
// ctx.channel().writeAndFlush(tws);
    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
// 如果HTTP解碼失敗，傳回HHTP異常
        if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req,
                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
//擷取url後置參數
        String uri = req.uri();
    
// 構造握手響應傳回，本機測試
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://localhost:8081/websocket" , null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
// 傳回應答給用戶端
        if (res.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }
// 如果是非Keep-Alive，關閉連接配接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     * exception 異常 Caught 抓住 抓住異常，當發生異常的時候，可以做一些相應的處理，比如列印日志、關閉連結
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        // TODO Auto-generated method stub

    }
}

Global

package com.atguigu.netty.websocket;

import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
* ClassName:Global
* Function: TODO ADD FUNCTION.
* @author fwz
*/
public class Global {
public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
}

controller

package com.atguigu.netty.websocket;

import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;



@Controller
public class DeptController_Consumer
{

    
    

    @RequestMapping(value = "/socketHtml")
    public String html()
    {
        return"WebSocketServer";
    }

    

}

WebSocketServer

<html>
<head>
<meta charset="UTF-8">
Netty WebSocket 時間伺服器
</head>
<br>
<body>
<br>
    <script type="text/javascript">
        var socket;
        if (!window.WebSocket) {
            window.WebSocket = window.MozWebSocket;
        }
        if (window.WebSocket) {
            socket = new WebSocket(
                    "ws://localhost:8081/websocket");
            socket.onmessage = function(event) {
                var ta = document.getElementById("responseText");
                ta.value="";
                ta.value=event.data;
            };
            socket.onopen = function(event) {
                
                
                var ta = document.getElementById("responseText");
                ta.value="";
                ta.value="打開WebSocket服務正常，浏覽器支援WebSocket！";
            };
            socket.onclose = function(event) {

                var ta = document.getElementById("responseText");
                ta.value="";
                ta.value="WebSocket關閉！";
            };
        }else{
            alert("抱歉，您的浏覽器不支援WebSocket")
        }

        function send(message) {
            if (!window.WebSocket) {
                return;
            }
            if (socket.readyState == WebSocket.OPEN) {
                socket.send(message);
            } else {
                alert("The socket is not open.");
            }
        }
    </script>
    <form οnsubmit="return false;">
        <input type="text" name="message" value="Hello,Netty-WebSocket" />
        <br>
        <br>
         <input
            type="button" value="發送 WebSocket 請求消息"
            onclick="send(this.form.message.value)" />
            <hr color="blue"/>
            <h3>服務端傳回的應答消息</h3>
            <textarea style="width: 500px;height: 300px;" id="responseText" ></textarea>
    </form>
</body>
</html>

啟動後如下：

基于netty的websocket例子

轉載于:https://www.cnblogs.com/fengwenzhee/p/10414899.html

基于netty的websocket例子

繼續閱讀

nginx location中斜線的位置的重要性

XX系統實施過程問題總結

nginx 安裝錯誤資訊解決

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method