前后端分离的Netty + WebSocket实现聊天室-EW帮帮网

1. 前端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket Chat</title>
</head>
<body>


<script type="text/javascript">

    var socket;
    if (!window.WebSocket) {
        window.WebSocket = window.MozWebSocket;
    }

    if (window.WebSocket) {
        socket = new WebSocket("ws://localhost:8088/ws");
        socket.onmessage = function(event) {
            console.log("receive message:" + event.data)
            var ele = document.getElementById('respText');
            ele.value = ele.value + "\n" + event.data;
        };

        socket.onopen = function(event) {
            var ele = document.getElementById('respText');
            ele.value = "连接开启";
        }

        socket.onclose = function(event) {
            var ele = document.getElementById('respText');
            ele.value = ele.value + "连接被关闭";
        }

    } else {
        alert("Your browser does not support WebSocket!");
    }



    function sendMessage(message) {
        if (!window.WebSocket) {
            alert("Your browser does not support WebSocket!")
            return;
        }
        if (socket.readyState == WebSocket.OPEN) {
            socket.send(message);
            console.log("send message:" + message)
        } else {
            alert("The connection is not open.");
        }
    }


</script>


<form onsubmit="return false">
    <h3>WebSocket Chat</h3>

    <textarea name="" id= "respText" style="width: 500px;height: 300px"></textarea>
    <input type="text" name="message" style="width: 300px" value="Welcome to chat.marding.cn">
    <input type="button" value="发送" onclick="sendMessage(this.form.message.value)">
    </input>
</form>
</body>
</html>

2. 后端

@Component
public class WebSocketChatServer {

    private int port;

    public WebSocketChatServer(@Value("${websocket.port:8088}") int port) {
        this.port = port;
    }

    @PostConstruct
    public void init() {
        try {
            this.run();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        //new WebSocketChatServer(8088).run();
    }

    public void run() throws Exception {

        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup)
            		// 非阻塞式IO
                    .channel(NioServerSocketChannel.class)
                    // 在每次有新的客户端连接时被调用，用来初始化这个新连接的 ChannelPipeline
                    .childHandler(new WebSocketChatServerInitializer())
                    // 服务器端点的最大排队连接数
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childOption(ChannelOption.SO_KEEPALIVE,true);

            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            System.out.println("bind port success");
            // 在Netty中用于阻塞当前线程，直到与通道（Channel）关联的连接被关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
            System.out.println("server stop");
        }


    }

}

public class WebSocketChatServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {

        ch.pipeline()
                .addLast(new HttpServerCodec())
                .addLast(new HttpObjectAggregator(64 * 1024))
                .addLast(new ChunkedWriteHandler())
                .addLast(new HttpRequestHandler("/ws"))
                .addLast(new WebSocketServerProtocolHandler("/ws"))
                .addLast(new TextWebSocketFrameHandler());
    }
}

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channels.add(channel);
        // 广播
        channels.writeAndFlush(new TextWebSocketFrame(channel.remoteAddress() + "加入"));
        System.out.println(channel.remoteAddress() + "加入");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        channels.writeAndFlush(new TextWebSocketFrame(channel.remoteAddress() + "离开"));
        System.out.println(channel.remoteAddress() + "离开");
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "在线");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + "掉线");
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + " 异常");
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        String text = msg.text();
        System.out.println("收到消息：" + text);
        Channel channel = ctx.channel();

        text = WordsFilter.filter(text);
        for (Channel ch : channels){
            if (channel != ch){
                ch.writeAndFlush(new TextWebSocketFrame(channel.remoteAddress() + ":" + text));
            }else {
                ch.writeAndFlush(new TextWebSocketFrame("[you]" + text));
            }
        }
    }

}

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
 
    private final String wsUri;

    private static final File PAGE_FILE;


    static {
        URL location = HttpRequestHandler.class.getProtectionDomain()
                .getCodeSource().getLocation();
        String path = null;
        try {
            path = location.toURI() + "wsChatClient.html";
            path = path.contains("file:") ? path.substring(5):path;
            PAGE_FILE = new File("\"C:\\Jlearning\\wsChatClient.html\"");
        } catch (URISyntaxException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * see {@link #SimpleChannelInboundHandler(boolean)} with {@code true} as boolean parameter.
     */
    public HttpRequestHandler(String wsUri) {
        this.wsUri = wsUri;
    }

    /**
     * Is called for each message of type {@link I}.
     *
     * @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler}
     *            belongs to
     * @param msg the message to handle
     * @throws Exception is thrown if an error occurred
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        if (wsUri.equalsIgnoreCase(request.uri())) {
            // 交给下一个InboudHandler 处理
            ctx.fireChannelRead(request.retain());
            return;
        }

        if (HttpUtil.is100ContinueExpected(request)) {
            // 发送 100 continue
            send100Continue(ctx);
        }
        // 读取文件内容到字节数组
        byte[] fileContent;
        try (RandomAccessFile file = new RandomAccessFile(PAGE_FILE, "r")) {
            fileContent = new byte[(int) file.length()];
            file.readFully(fileContent);
        } catch (IOException e) {
            // 处理文件读取错误
            DefaultFullHttpResponse errorResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR);
            ctx.writeAndFlush(errorResponse);
            return;
        }

        DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
        resp.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");

        boolean keepAlive = HttpUtil.isKeepAlive(request);
        if (keepAlive) {
            resp.headers().set(HttpHeaderNames.CONTENT_LENGTH, fileContent.length);
            resp.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }

        // 将文件内容设置为响应内容
        resp.content().writeBytes(fileContent);

        ChannelFuture channelFuture = ctx.writeAndFlush(resp);

        if (!keepAlive) {
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                System.out.println("response empty last content success !");
            }
        });
    }

    private void send100Continue(ChannelHandlerContext ctx) {
        DefaultFullHttpResponse resp = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
        ctx.writeAndFlush(resp);
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("client " + channel.remoteAddress() + "异常");
        cause.printStackTrace();
        ctx.close();
    }
}

public class WordsFilter {
    public static Map<String, Object> dictionaryMap = new HashMap<>();


    /**
     * 生成关键词字典库
     * @param words
     * @return
     */
    public static void initMap(Collection<String> words) {
        if (words == null) {
            System.out.println("敏感词列表不能为空");
            return ;
        }

        // map初始长度words.size()，整个字典库的入口字数(小于words.size()，因为不同的词可能会有相同的首字)
        Map<String, Object> map = new HashMap<>(words.size());
        // 遍历过程中当前层次的数据
        Map<String, Object> curMap = null;
        Iterator<String> iterator = words.iterator();

        while (iterator.hasNext()) {
            String word = iterator.next();
            curMap = map;
            int len = word.length();
            for (int i =0; i < len; i++) {
                // 遍历每个词的字
                String key = String.valueOf(word.charAt(i));
                // 当前字在当前层是否存在, 不存在则新建, 当前层数据指向下一个节点, 继续判断是否存在数据
                Map<String, Object> wordMap = (Map<String, Object>) curMap.get(key);
                if (wordMap == null) {
                    // 每个节点存在两个数据: 下一个节点和isEnd(是否结束标志)
                    wordMap = new HashMap<>(2);
                    wordMap.put("isEnd", "0");
                    curMap.put(key, wordMap);
                }
                curMap = wordMap;
                // 如果当前字是词的最后一个字，则将isEnd标志置1
                if (i == len -1) {
                    curMap.put("isEnd", "1");
                }
            }
        }

        dictionaryMap = map;
    }

    static {
        List<String> list = new ArrayList<>();
        list.add("坏蛋");
        list.add("笨蛋");
        initMap(list);
    }

    /**
     * 搜索文本中某个文字是否匹配关键词
     * @param text
     * @param beginIndex
     * @return
     */
    private static int checkWord(String text, int beginIndex) {
        if (dictionaryMap == null) {
            throw new RuntimeException("字典不能为空");
        }
        boolean isEnd = false;
        int wordLength = 0;
        Map<String, Object> curMap = dictionaryMap;
        int len = text.length();
        // 从文本的第beginIndex开始匹配
        for (int i = beginIndex; i < len; i++) {
            String key = String.valueOf(text.charAt(i));
            // 获取当前key的下一个节点
            curMap = (Map<String, Object>) curMap.get(key);
            if (curMap == null) {
                break;
            } else {
                wordLength ++;
                if ("1".equals(curMap.get("isEnd"))) {
                    isEnd = true;
                }
            }
        }
        if (!isEnd) {
            wordLength = 0;
        }
        return wordLength;
    }

    public static String filter(String text) {
        StringBuilder result = new StringBuilder();
        int length = text.length();
        int position = 0;

        while (position < length) {
            int matchLength = checkWord(text, position);
            if (matchLength > 0) {
                // 发现敏感词，用**替换
                result.append("**");
                position += matchLength;
            } else {
                result.append(text.charAt(position));
                position++;
            }
        }
        return result.toString();
    }

    /**
     * 获取匹配的关键词和命中次数
     * @param text
     * @return
     */
    public static Map<String, Integer> matchWords(String text) {
        Map<String, Integer> wordMap = new HashMap<>();
        int len = text.length();
        for (int i = 0; i < len; i++) {
            int wordLength = checkWord(text, i);
            if (wordLength > 0) {
                String word = text.substring(i, i + wordLength);
                // 添加关键词匹配次数
                if (wordMap.containsKey(word)) {
                    wordMap.put(word, wordMap.get(word) + 1);
                } else {
                    wordMap.put(word, 1);
                }

                i += wordLength - 1;
            }
        }
        return wordMap;
    }
}

前后端分离的Netty + WebSocket实现聊天室

1. 前端

2. 后端

网站公告

今日签到

热门文章

最新发布