Netty学习-简易的通信

因为大创项目需要一个类似于QQ的聊天功能的需求，因此学习了netty与websocket。现在记录下第一次简易的实时聊天

只是引入了一个netty的包

    4.0.0

    org.example
    netty-learn
    1.0-SNAPSHOT

    
        
        
            io.netty
            netty-all
            4.1.79.Final
        
    
    
        11
        11
    

package org.example;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class WebSocketServer {

    public static void main(String[] args) {
        // 创建主线程池
        NioEventLoopGroup mainGroup = new NioEventLoopGroup();
        // 创建从线程池
        NioEventLoopGroup subGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap server = new ServerBootstrap();
            // 分组
            server.group(mainGroup, subGroup)
                    // 使用 NioServerSocketChannel 作为数据通道
                    .channel(NioServerSocketChannel.class)
                    // 添加 websocket 处理器
                    .childHandler(new WebSocketServerInitialzer());
            ChannelFuture future = server.bind(8088).sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }finally {
            // 关闭资源
            mainGroup.shutdownGracefully();
            subGroup.shutdownGracefully();
        }
    }
}

package org.example;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServerInitialzer extends ChannelInitializer {
    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        // 集成了很多的 handler 的流水线
        ChannelPipeline pipeline = channel.pipeline();

        // http 的编码与解码器
        pipeline.addLast(new HttpServerCodec());

        // 数据过大处理
        pipeline.addLast(new ChunkedWriteHandler());

        // 消息聚合
        pipeline.addLast(new HttpObjectAggregator(1024*64));

        // 协议 前端需要写这个一致
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));

        // 添加聊天处理器
        pipeline.addLast(new ChatHandler());
    }
}

package org.example;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.Date;



public class ChatHandler extends SimpleChannelInboundHandler {

    
    private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) throws Exception {
        String content = msg.text();
        System.out.println("content = " + content);

        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

        // 数据响应前端
        clients.writeAndFlush(
                new TextWebSocketFrame(
                        "用户" + channelHandlerContext.channel().id().asShortText() + "在" + dateFormat.format(new Date()) + "说" + content
                )
        );
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("用户上线" + ctx.channel().id().asShortText());
        clients.add(ctx.channel());
        ctx.channel().writeAndFlush(new TextWebSocketFrame(ctx.channel().id().asShortText()));
    }


    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户断开链接" + ctx.channel().id().asShortText());
        clients.remove(ctx.channel());
    }
}

一个简单的vue3脚手架

// 这是configs 下的index.js
const BASE_URL = '192.168.1.5'
const WS_PORT = '8088'
const WS_ADDRESS = `ws://${BASE_URL}:${WS_PORT}/ws` // 这个ws 就是后端写的那个
export { WS_ADDRESS }

// 这个是hooks 的 index.js
import userWebSocket from './websocket'

export { userWebSocket }

import { WS_ADDRESS } from '../configs/index.js'
function userWebSocket(handleMessage) {
  const ws = new WebSocket(WS_ADDRESS)

  const init = () => {
    bindEvent()
  }

  function bindEvent() {
    ws.addEventListener('open', handleOpen, false)
    ws.addEventListener('close', handleClose, false)
    ws.addEventListener('error', handleError, false)
    ws.addEventListener('message', handleMessage, false)
  }

  function handleOpen(e) {
    console.log('socket 连接成功', e)
  }
  function handleClose(e) {
    console.log('socket 连接关闭 close', e)
  }
  function handleError(e) {
    console.log('socket 连接错误', e)
  }

  init()

  return ws
}
export default userWebSocket

将组件挂到APP.vue 上就可以了最后运行

最后效果

现在仅是学习阶段，有点潦草，一起学习，勿喷