Netty网络聊天(一) 聊天室实战

简介: Netty网络聊天(一) 聊天室实战

之前做过一个IM的项目,里面涉及了基本的聊天功能,所以注意这系列的文章不是练习,不含基础和逐步学习的部分,直接开始实战和思想引导,基础部分需要额外的去补充,我有精力的话可以后续出一系列的文章。


为什么第一篇是聊天室,聊天室是最容易实现的部分。也是IM结构最简单的一部分,其次作单聊和群聊,业务逻辑层层递增,彻底的拿下聊天室的代码,进阶单聊和群聊就很简单了,后续我还会推出直播间的实现。


如果单纯想实现聊天室很简单,但是我尽量会把流程都走全,为了方便理解。


主要由两个功能类实现:初始化类+响应处理类


1、准备工作


添加pom.xml


<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.2.Final</version>
</dependency>


2、辅助接口实现


辅助接口让服务架构更加清晰,这里有两个接口,一个用来处理Http请求,一个处理Webocket请求。


MyHttpService.java


/**
 * 处理 http请求
 */
public interface MyHttpService {
    void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request);
}
MyWebSocketService.java
/**
 * 处理 WebSocket 请求中的frame
 */
public interface MyWebSocketService {
    void handleFrame(ChannelHandlerContext ctx, WebSocketFrame frame);
}


那么问题来了,谁来实现这两个类呢,谁来处理这两种请求的分发呢。


下面来看服务响应处理类:


WebSocketServerHandler.java


3、请求处理类


继承SimpleChannelInboundHandler类,实现channelRead0() handlerAdded() handlerRemoved() exceptionCaught()等方法,第一个是必选方法,其他方法供我们做一些标记和后续处理。


WebSocketServerHandler.java


@Slf4j
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    private MyHttpService httpService;
    private MyWebSocketService webSocketService;
    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    public WebSocketServerHandler(MyHttpService httpService, MyWebSocketService webSocketService) {
        super();
        this.httpService = httpService;
        this.webSocketService = webSocketService;
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            httpService.handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            webSocketService.handleFrame(ctx, (WebSocketFrame) msg);
        }
    }
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        channels.add(ctx.channel());
        channels.writeAndFlush(new TextWebSocketFrame(ctx.channel() +"上线了"));
    }
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        channels.remove(ctx.channel());
        channels.writeAndFlush(new TextWebSocketFrame(ctx.channel() +"下线了"));
    }
    /**
     * 发生异常时处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        channels.remove(ctx.channel());
        ctx.close();
        log.info("异常信息:{}",cause.getMessage());
    }
}


  • 创建ChannelGroup,来保存每个已经建立连接的channel,在handlerAdded()方法中channels.add(ctx.channel());,相应的在handlerRemoved方法中remove。


  • 在channelRead0()方法中,实现了对请求的识别和分别处理。


  • exceptionCaught()方法为发生异常时处理。


4、初始化类实现


@Slf4j
public class WebSocketServer implements MyHttpService, MyWebSocketService {
    /**
     * 握手用的 变量
     */
    private static final AttributeKey<WebSocketServerHandshaker> ATTR_HAND_SHAKER = AttributeKey.newInstance("ATTR_KEY_CHANNEL_ID");
    private static final int MAX_CONTENT_LENGTH = 65536;
    /**
     * 请求类型常量
     */
    private static final String WEBSOCKET_UPGRADE = "websocket";
    private static final String WEBSOCKET_CONNECTION = "Upgrade";
    private static final String WEBSOCKET_URI_ROOT_PATTERN = "ws://%s:%d";
    /**
     * 用户字段
     */
    private String host;
    private int port;
    /**
     * 保存 所有的连接
     */
    private Map<ChannelId, Channel> channelMap = new HashMap<>();
    private final String WEBSOCKET_URI_ROOT;
    public WebSocketServer(String host, int port) {
        this.host = host;
        this.port = port;
        // 将 ip 和端口 按照格式 赋值给 uri
        WEBSOCKET_URI_ROOT = String.format(WEBSOCKET_URI_ROOT_PATTERN, host, port);
    }
    public void start(){
        // 实例化 nio监听事件池
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 实例化 nio工作线程池
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        // 启动器
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup,workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                ChannelPipeline pl = channel.pipeline();
                // 保存 该channel 到map中
                channelMap.put(channel.id(),channel);
                log.info("new channel {}",channel);
                channel.closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                    log.info("channel close future  {}",channelFuture);
                    //关闭后 从map中移除
                    channelMap.remove(channelFuture.channel().id());
                });
                //添加 http 编解码
                pl.addLast(new HttpServerCodec());
                // 聚合器
                pl.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH));
                // 支持大数据流
                pl.addLast(new ChunkedWriteHandler());
                // 设置 websocket 服务处理方式
                pl.addLast(new WebSocketServerHandler(WebSocketServer.this, WebSocketServer.this));
            }
        });
        /**
         * 实例化完毕后,需要完成端口绑定
         */
        try {
            ChannelFuture channelFuture = bootstrap.bind(host,port).addListener((ChannelFutureListener) channelFuture1 -> {
                if (channelFuture1.isSuccess()){
                    log.info("webSocket started");
                }
            }).sync();
            channelFuture.channel().closeFuture().addListener((ChannelFutureListener) channelFuture12 ->
                    log.info("server channel {} closed.", channelFuture12.channel())).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.out.println("绑定端口失败");
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
        log.info("webSocket shutdown");
    }
    @Override
    public void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        //判断是不是 socket 请求
        if (isWebSocketUpgrade(request)){
            //如果是webSocket请求
            log.info("请求是webSocket协议");
            // 获取子协议
            String subProtocols = request.headers().get(HttpHeaderNames.SEC_WEBSOCKET_PROTOCOL);
            //握手工厂 设置 uri+协议+不允许扩展
            WebSocketServerHandshakerFactory handshakerFactory = new WebSocketServerHandshakerFactory(WEBSOCKET_URI_ROOT,subProtocols,false);
            // 从工厂中实例化一个 握手请求
            WebSocketServerHandshaker handshaker = handshakerFactory.newHandshaker(request);
            if (handshaker == null){
                //握手失败:不支持的协议
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            }else {
                //响应请求:将 握手转交给 channel处理
                handshaker.handshake(ctx.channel(),request);
                //将 channel 与 handshaker 绑定
                ctx.channel().attr(ATTR_HAND_SHAKER).set(handshaker);
            }
            return;
        }else {
            // 不处理 HTTP 请求
            log.info("不处理 HTTP 请求");
        }
    }
    @Override
    public void handleFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        /**
         * text frame handler
         */
        if (frame instanceof TextWebSocketFrame){
            String text = ((TextWebSocketFrame) frame).text();
            TextWebSocketFrame textWebSocketFrame = new TextWebSocketFrame(text);
            log.info("receive textWebSocketFrame from channel: {} , 目前一共有{}个在线",ctx.channel(),channelMap.size());
            //发给其它的 channel  (群聊功能)
            for (Channel ch: channelMap.values()){
                if (ch.equals(ctx.channel())){
                    continue;
                }
                //将 text frame 写出
                ch.writeAndFlush(textWebSocketFrame);
                log.info("消息已发送给{}",ch);
                log.info("write text: {} to channel: {}",textWebSocketFrame,ctx.channel());
            }
            return;
        }
        /**
         * ping frame , 回复  pong frame
         */
        if (frame instanceof PingWebSocketFrame){
            log.info("receive pingWebSocket from channel: {}",ctx.channel());
            ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        /**
         * pong frame, do nothing
         */
        if (frame instanceof PongWebSocketFrame){
            log.info("receive pongWebSocket from channel: {}",ctx.channel());
            return;
        }
        /**
         * close frame, close
         */
        if (frame instanceof CloseWebSocketFrame){
            log.info("receive closeWebSocketFrame from channel: {}", ctx.channel());
            //获取到握手信息
            WebSocketServerHandshaker handshaker = ctx.channel().attr(ATTR_HAND_SHAKER).get();
            if (handshaker == null){
                log.error("channel: {} has no handShaker", ctx.channel());
                return;
            }
            handshaker.close(ctx.channel(),((CloseWebSocketFrame) frame).retain());
            return;
        }
        /**
         * 剩下的都是 二进制 frame ,忽略
         */
        log.warn("receive binary frame , ignore to handle");
    }
    /**
     * 判断是否是 webSocket 请求
     */
    private boolean isWebSocketUpgrade(FullHttpRequest req) {
        HttpHeaders headers = req.headers();
        return req.method().equals(HttpMethod.GET)
                && headers.get(HttpHeaderNames.UPGRADE).contains(WEBSOCKET_UPGRADE)
                && headers.get(HttpHeaderNames.CONNECTION).contains(WEBSOCKET_CONNECTION);
    }
}


  • l.addLast(new WebSocketServerHandler(WebSocketServer.this, WebSocketServer.this)); 添加自己的响应处理。WebSocketServerHandler是第二点实现的请求处理类.


  • private Map channelMap = new HashMap<>();来将ChannelId和CHannel对应保存。方便后来对应获取。


  • bootstrap.bind(host,port)也可以替换成仅bind端口。


public ChannelFuture bind(String inetHost, int inetPort) {
        return bind(new InetSocketAddress(inetHost, inetPort));
    }
    public synchronized InetAddress anyLocalAddress() {
        if (anyLocalAddress == null) {
            anyLocalAddress = new Inet4Address(); // {0x00,0x00,0x00,0x00}
            anyLocalAddress.holder().hostName = "0.0.0.0";
        }
        return anyLocalAddress;
    }


它默认会给0.0.0.0端口开放服务。


  • handleHttpRequest和handleFrame是MyWebSocketService类的一个实现。


  • 各个细节都有注释,仔细看注释。


5、启动服务


public class Main {
    public static void main(String[] args) {
        new WebSocketServer("192.168.1.33",9999).start();
    }
}


局域网内如何测试?


我用的是npm 的一个serve 服务来搞局域网。


官网介绍:

https://www.npmjs.com/package/serve

我的文章:

React打包注意事项及静态文件服务搭建


image.png


这下保证你的手机和电脑都在局域网内,就可以访问你自己的群聊了。


6、前端页面


要送就送一套


<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>Document</title>
    <style type="text/css">
        .talk_con{
            width:600px;
            height:500px;
            border:1px solid #666;
            margin:50px auto 0;
            background:#f9f9f9;
        }
        .talk_show{
            width:580px;
            height:420px;
            border:1px solid #666;
            background:#fff;
            margin:10px auto 0;
            overflow:auto;
        }
        .talk_input{
            width:580px;
            margin:10px auto 0;
        }
        .whotalk{
            width:80px;
            height:30px;
            float:left;
            outline:none;
        }
        .talk_word{
            width:420px;
            height:26px;
            padding:0px;
            float:left;
            margin-left:10px;
            outline:none;
            text-indent:10px;
        }
        .talk_sub{
            width:56px;
            height:30px;
            float:left;
            margin-left:10px;
        }
        .atalk{
            margin:10px;
        }
        .atalk span{
            display:inline-block;
            background:#0181cc;
            border-radius:10px;
            color:#fff;
            padding:5px 10px;
        }
        .btalk{
            margin:10px;
            text-align:right;
        }
        .btalk span{
            display:inline-block;
            background:#ef8201;
            border-radius:10px;
            color:#fff;
            padding:5px 10px;
        }
    </style>
    <script type="text/javascript">
        //
        document.onkeydown = function (ev) {
            if (ev && ev.keyCode == 13){
                send();
                clear();
            }
        }
        var socket;
        if (window.WebSocket) {
            socket = new WebSocket("ws://192.168.1.33:9999");
            // socket = new WebSocket("ws://127.0.0.1:9999");
            // socket = new WebSocket("ws://192.168.43.186:9999");
            socket.onmessage = function (ev) {
                atalkAppendIn("接收:"+socket.channel + ":" + ev.data)
            };
            socket.onopen = function () {
                btalkAppendIn("连接已建立");
            }
            socket.onclose = function () {
                btalkAppendIn("连接关闭");
            };
        }else {
            alert("浏览器不支持");
        }
        function send(){
            var message = document.getElementById("talkwords");
            if (!window.WebSocket){
                return
            }
            if (socket.readyState === WebSocket.OPEN){
                socket.send(message.value);
                btalkAppendIn("发送:"+ message.value);
                clear();
            } else {
                alert("WebSocket 建立失败");
            }
        }
        function atalkAppendIn(text) {
            var append = document.getElementById("words");
            append.innerHTML+= '<div class="atalk"><span>'+ text +'</span></div>';
        }
        function btalkAppendIn(text) {
            var append = document.getElementById("words");
            append.innerHTML+= '<div class="btalk"><span>'+ text +'</span></div>';
        }
        function clear () {
            var elementById = document.getElementById("talkwords");
            elementById.value = "";
        }
    </script>
</head>
<body>
<div class="talk_con">
    <div class="talk_show" id="words">
    </div>
    <div class="talk_input">
        <!--<select class="whotalk" id="who">-->
            <!--<option value="0">A说:</option>-->
            <!--<option value="1">B说:</option>-->
        <!--</select>-->
        <input type="text" class="talk_word" id="talkwords">
        <input type="button" onclick="send()" value="发送" class="talk_sub" id="talksub">
    </div>
</div>
</body>
</html>


  • socket = new WebSocket("ws://192.168.1.33:9999");
    注意这里ip和port与服务一一对应。


  • socket.onmessage()是获取socket信息。socket.onopen是创建连接。socket.onclose是关闭连接。socket.send(message.value);是发送socket信息。


image.png


image.png


控制台输出:


15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - receive textWebSocketFrame from channel: [id: 0x0d08c657, L:/192.168.1.33:9999 - R:/192.168.1.33:50440] , 目前一共有2个在线 
15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - 消息已发送给[id: 0xacd5c1ad, L:/192.168.1.33:9999 - R:/192.168.1.33:50438] 
15:12:42.444 [nioEventLoopGroup-3-5] DEBUG io.netty.handler.codec.http.websocketx.WebSocket08FrameEncoder - Encoding WebSocket Frame opCode=1 length=5 
15:12:42.443 [nioEventLoopGroup-3-6] INFO com.fantj.springbootjpa.netty.WebSocketServer - wri


目录
相关文章
|
2月前
|
机器学习/深度学习 PyTorch 算法框架/工具
目标检测实战(一):CIFAR10结合神经网络加载、训练、测试完整步骤
这篇文章介绍了如何使用PyTorch框架,结合CIFAR-10数据集,通过定义神经网络、损失函数和优化器,进行模型的训练和测试。
116 2
目标检测实战(一):CIFAR10结合神经网络加载、训练、测试完整步骤
|
28天前
|
数据采集 存储 JSON
Python网络爬虫:Scrapy框架的实战应用与技巧分享
【10月更文挑战第27天】本文介绍了Python网络爬虫Scrapy框架的实战应用与技巧。首先讲解了如何创建Scrapy项目、定义爬虫、处理JSON响应、设置User-Agent和代理,以及存储爬取的数据。通过具体示例,帮助读者掌握Scrapy的核心功能和使用方法,提升数据采集效率。
79 6
|
2月前
|
机器学习/深度学习 数据可视化 测试技术
YOLO11实战:新颖的多尺度卷积注意力(MSCA)加在网络不同位置的涨点情况 | 创新点如何在自己数据集上高效涨点,解决不涨点掉点等问题
本文探讨了创新点在自定义数据集上表现不稳定的问题,分析了不同数据集和网络位置对创新效果的影响。通过在YOLO11的不同位置引入MSCAAttention模块,展示了三种不同的改进方案及其效果。实验结果显示,改进方案在mAP50指标上分别提升了至0.788、0.792和0.775。建议多尝试不同配置,找到最适合特定数据集的解决方案。
461 0
|
18天前
|
消息中间件 编解码 网络协议
Netty从入门到精通:高性能网络编程的进阶之路
【11月更文挑战第17天】Netty是一个基于Java NIO(Non-blocking I/O)的高性能、异步事件驱动的网络应用框架。使用Netty,开发者可以快速、高效地开发可扩展的网络服务器和客户端程序。本文将带您从Netty的背景、业务场景、功能点、解决问题的关键、底层原理实现,到编写一个详细的Java示例,全面了解Netty,帮助您从入门到精通。
66 0
|
29天前
|
数据采集 前端开发 中间件
Python网络爬虫:Scrapy框架的实战应用与技巧分享
【10月更文挑战第26天】Python是一种强大的编程语言,在数据抓取和网络爬虫领域应用广泛。Scrapy作为高效灵活的爬虫框架,为开发者提供了强大的工具集。本文通过实战案例,详细解析Scrapy框架的应用与技巧,并附上示例代码。文章介绍了Scrapy的基本概念、创建项目、编写简单爬虫、高级特性和技巧等内容。
57 4
|
29天前
|
网络协议 物联网 API
Python网络编程:Twisted框架的异步IO处理与实战
【10月更文挑战第26天】Python 是一门功能强大且易于学习的编程语言,Twisted 框架以其事件驱动和异步IO处理能力,在网络编程领域独树一帜。本文深入探讨 Twisted 的异步IO机制,并通过实战示例展示其强大功能。示例包括创建简单HTTP服务器,展示如何高效处理大量并发连接。
45 1
|
1月前
|
网络协议 安全 NoSQL
网络空间安全之一个WH的超前沿全栈技术深入学习之路(8-2):scapy 定制 ARP 协议 、使用 nmap 进行僵尸扫描-实战演练、就怕你学成黑客啦!
scapy 定制 ARP 协议 、使用 nmap 进行僵尸扫描-实战演练等具体操作详解步骤;精典图示举例说明、注意点及常见报错问题所对应的解决方法IKUN和I原们你这要是学不会我直接退出江湖;好吧!!!
网络空间安全之一个WH的超前沿全栈技术深入学习之路(8-2):scapy 定制 ARP 协议 、使用 nmap 进行僵尸扫描-实战演练、就怕你学成黑客啦!
|
1月前
|
网络协议 安全 算法
网络空间安全之一个WH的超前沿全栈技术深入学习之路(9):WireShark 简介和抓包原理及实战过程一条龙全线分析——就怕你学成黑客啦!
实战:WireShark 抓包及快速定位数据包技巧、使用 WireShark 对常用协议抓包并分析原理 、WireShark 抓包解决服务器被黑上不了网等具体操作详解步骤;精典图示举例说明、注意点及常见报错问题所对应的解决方法IKUN和I原们你这要是学不会我直接退出江湖;好吧!!!
网络空间安全之一个WH的超前沿全栈技术深入学习之路(9):WireShark 简介和抓包原理及实战过程一条龙全线分析——就怕你学成黑客啦!
|
2月前
|
XML Java Nacos
网络协议与Netty
网络协议与Netty
|
28天前
|
网络协议 调度 开发者
Python网络编程:Twisted框架的异步IO处理与实战
【10月更文挑战第27天】本文介绍了Python网络编程中的Twisted框架,重点讲解了其异步IO处理机制。通过反应器模式,Twisted能够在单线程中高效处理多个网络连接。文章提供了两个实战示例:一个简单的Echo服务器和一个HTTP服务器,展示了Twisted的强大功能和灵活性。
37 0