源码时代JAVA干货分享|带你用Netty框架实现WebSocket通信

客官°小女子只卖身不卖艺 2023-06-13 06:02 26阅读 0赞

**功能介绍**

*  Netty开发服务端
 *  HTML实现客户端
 *  实现服务端与客户端时实时交互

**开发步骤**

1.**导包**

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty‐all</artifactId>
    <version>5.0.0.Alpha2</version>
    </dependency>

2.**工程配置文件：NettyConfig**

/**
    
    * 这里放的是工程中相应的配置
    
    */
    
    public class NettyConfig{
    
    
    
    /**
    
    * 用于存储每一个客户端接入进来时的channel对象
    
    */
    
    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    }

3.**MyWebSocketHandler**

public class MyWebSocketHandler extends SimpleChannelInboundHandler<Object> { private WebSocketServerHandshaker handshaker;
    
    //请求路径
    
    private static final String WEB_SOCKET_URL = "ws://localhost:8888/websocket";
    
    
    
    //工程出现异常的时候调用@Override
    
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    
    cause.printStackTrace(); ctx.close(); //关闭当前连接
    
    }
    
    
    
    //客户端与服务器创建连接的时候调用@Override
    
    public void channelActive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.add(ctx.channel());
    
    System.out.println("客户端与服务器连接开启");
    
    }
    
    //客户端与服务器断开连接的时候调用@Override
    
    public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyConfig.group.remove(ctx.channel());
    
    System.out.println("客户端与服务器断开连接");
    
    }
    
    //服务端接收客户端发送过来的数据结束之后调用@Override
    
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception{ ctx.flush(); //进行数据清除
    
    }
    
    
    
    //服务端处理客户端调用的核心方法
    
    protected void messageReceived(ChannelHandlerContext context, Object msg) throws Exception {
    
    //处理客户端向服务端发起http握手请求业务if(msg instanceof FullHttpRequest){
    
    handHttpRequest(context,(FullHttpRequest) msg);
    
    }else if(msg instanceof WebSocketFrame){
    
    //处理websocket连接业务handWebSocketFrme(context,(WebSocketFrame)msg);
    
    }
    
    }
    
    
    
    /**
    
    处理客户端与服务端之前的websocket业务
    @param ctx
    @param frame
    */
    
    
    private void handWebSocketFrme(ChannelHandlerContext ctx,WebSocketFrame frame){
    
    //判断是否是关闭websocket的指令
    
    if(frame instanceof CloseWebSocketFrame){ handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
    
    }
    
    //判断是否是Ping消息
    
    if(frame instanceof PingWebSocketFrame){
    
    //返回一个胖信息
    
    ctx.channel().write(newPongWebSocketFrame(frame.content().retain())); return;
    
    }
    
    //判断是否是二进制消息，如果是二进制消息，抛出异常if(!(frame instanceof TextWebSocketFrame)){
    
    System.out.println("目前我们不支持二进制消息");
    
    throw new RuntimeException(this.getClass().getName()+"不支持的消息");
    
    }
    
    //返回应答消息
    
    //获取客户端向服务端发送的消息
    
    String request = ((TextWebSocketFrame)frame).text(); System.out.println("服务端收到客户端的消息===>"+request); TextWebSocketFrame tws = new TextWebSocketFrame(new
    
    Date().toString()+ctx.channel().id()+"===>"+request);
    
    //群发，服务端向每个上来的客户端群发消息NettyConfig.group.writeAndFlush(tws);
    
    }
    
    
    
    /**
    
    处理客户端向服务器发起http握手请求的业务
    @param ctx
    @param req
    */
    
    private void handHttpRequest(ChannelHandlerContext ctx,FullHttpRequest req){
    
    //        req.headers().get()
    
    //        req.decoderResult().isSuccess()
    
    //不是websocket,就不是客户端发给服务器的Http请求if(!req.decoderResult().isSuccess()|| !
    
    ("websocket").equals(req.headers().get("Upgrade"))){
    
    sendHttpResponse(ctx,req,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
    
    return;
    
    }
    
    WebSocketServerHandshakerFactory wsFactory =
    
    newWebSocketServerHandshakerFactory(WEB_SOCKET_URL,null,false); handshaker = wsFactory.newHandshaker(req);
    
    if(handshaker==null){
    
    
    
    WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
    
    }else{
    
    handshaker.handshake(ctx.channel(),req);
    
    }
    
    }
    
    
    
    /**
    
    服务端向客户端响应消息
    @param ctx
    @param req
    @param res
    
    */
    
    private 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();
    
    }
    
    //服务器向客户端发送数据
    
    ChannelFuture f = ctx.channel().writeAndFlush(res); if(res.status().code()!=200){
    
    f.addListener(ChannelFutureListener.CLOSE);
    
    }
    
    }
    
    
    
    }

# 4 初始化连接各个组件:MyWebSocketChannelHandler #

/**
    
    * 初始化连接各个组件
    
    */
    
    public class MyWebSocketChannelHandlerextends ChannelInitializer<SocketChannel>{
    
    
    
    protected void initChannel(SocketChannel e) throws Exception { e.pipeline().addLast("http‐codec",new HttpServerCodec()); e.pipeline().addLast("aggregator",new HttpObjectAggregator(66636)); e.pipeline().addLast("http‐chunked",new ChunkedWriteHandler()); e.pipeline().addLast("handler",new MyWebSocketHandler());
    
    }
    
    }
    
    
    
    
    
    5 程序入口
    
    /**
    
    * 程序的入口，负责启动应用
    
    */
    
    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("服务端开启等待客户端连接");
    
    Channel ch = b.bind(8888).sync().channel(); ch.closeFuture().sync();
    
    
    
    
    } catch (Exception e) {
    
    e.printStackTrace();
    
    } finally {
    
    //优雅的退出程序bossGroup.shutdownGracefully(); workGroup.shutdownGracefully();
    
    }
    
    }
    
    }
    
    
    
    6 web端测试代码
    
    
    
    <!DOCTYPE html>
    
    <htmllang="en">
    
    <head>
    
    <meta charset="UTF‐8">
    
    <title>WebSocket客户端</title>
    
    </head>
    
    <script>
    
    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("responseContext"); ta.value += event.data +"\r\n";
    
    }
    
    socket.onopen = function (event) {
    
    var ta = document.getElementById("responseContext"); ta.value = "您浏览器支持webSoceket，请进行后续 操作\r\n";
    
    }
    
    
    
    socket.onclose = function (event) {
    
    var ta = document.getElementById("responseContext"); ta.value = "";
    
    ta.value = "webSoceket连接已经关闭";
    
    }
    
    }else{
    
    console.debug("浏览器不支持webSocket")
    
    }
    
    
    
    function send() {
    
    var message = document.getElementById("message").value; if(!window.WebSocket){
    
    return;
    
    }
    
    if(socket.readyState == WebSocket.OPEN){ socket.send(message)
    
    }else{
    
    alert("连接没有建立成功！");
    
    }
    
    }
    
    </script>
    
    
    <body>
    
    
    
    <form action=""onsubmit="return false;">
    
    <input type="text" id="message"name="message" value=""/>
    
    <inputtype="button"value="发送消息" onclick="send()" />
    
    <hr >
    
    <h1>客户端接收到服务端返回的消息</h1>
    
    <textarea name=""id="responseContext" cols="30"rows="10"></textarea>
    
    </form>
    
    </body>
    
    </html>

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpbGxhaW55MTM1Nzk_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpbGxhaW55MTM1Nzk_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191114114547783.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3ZpbGxhaW55MTM1Nzk=,size_16,color_FFFFFF,t_70