网络编程_IO_Netty

矫情吗；* 2021-10-06 04:00 295阅读 0赞

### 网络编程\_IO\_Netty ###

*  依赖pom.xml
 *  源码
 *   *  启动入口
     *  通道初始化器
     *  通道消息处理
     *  前段代码
 *  简单聊天应用
 *   *  启动netty
     *  初始化
     *  消息处理
     *  心跳处理
     *  存储所有用户通道
     *  消息实体类

# 依赖pom.xml #

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

# 源码 #

## 启动入口 ##

/**
     * Netty Server
     * 
     * @author vander
     *
     */
    public class WebSocketNettyServer {
        public static void main(String[] args) {
            // 创建两个线程池
            NioEventLoopGroup mainGrp = new NioEventLoopGroup(); // 主线程池
            NioEventLoopGroup subGrp = new NioEventLoopGroup(); // 从线程池
    
            try {
                // 创建Netty服务器启动对象
                ServerBootstrap serverBootstrap = new ServerBootstrap();
    
                // 初始化服务器启动对象
                serverBootstrap
                        // 指定使用上面创建的两个线程池
                        .group(mainGrp, subGrp)
                        // 指定Netty通道类型
                        .channel(NioServerSocketChannel.class)
                        // 指定通道初始化器用来加载当Channel收到事件消息后，
                        // 如何进行业务处理
                        .childHandler(new WebSocketChannelInitializer());
    
                // 绑定服务器端口，以同步的方式启动服务器
                ChannelFuture future = serverBootstrap.bind(9090).sync();
                // 等待服务器关闭
                future.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 优雅关闭服务器
                mainGrp.shutdownGracefully();
                subGrp.shutdownGracefully();
            }
    
        }
    }

## 通道初始化器 ##

/**
     * 通道初始化器
     * 用来加载通道处理器（ChannelHandler)
     */
    public class WebSocketChannelInitializer extends ChannelInitializer<SocketChannel> {
    
        // 初始化通道
        // 在这个方法中去加载对应的ChannelHandler
        protected void initChannel(SocketChannel ch) throws Exception {
            // 获取管道，将一个一个的ChannelHandler添加到管道中
            ChannelPipeline pipeline = ch.pipeline();
    
            // 添加一个http的编解码器
            pipeline.addLast(new HttpServerCodec());
            // 添加一个用于支持大数据流的支持
            pipeline.addLast(new ChunkedWriteHandler());
            // 添加一个聚合器，这个聚合器主要是将HttpMessage聚合成FullHttpRequest/Response
            pipeline.addLast(new HttpObjectAggregator(1024 * 64));
    
            // 需要指定接收请求的路由
            // 必须使用以ws后缀结尾的url才能访问
            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    
            // 添加自定义的Handler
            pipeline.addLast(new ChatHandler());
        }
    }

## 通道消息处理 ##

/**
     * 消息处理
     * 
     * @author vander
     *
     */
    public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    
        // 用来保存所有的客户端连接
        private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-dd hh:MM");
    
        // 当Channel中有新的事件消息会自动调用
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
            // 当接收到数据后会自动调用
    
            // 获取客户端发送过来的文本消息
            String text = msg.text();
            System.out.println("接收到消息数据为：" + text);
    
            for (Channel client : clients) {
                // 将消息发送到所有的客户端
                client.writeAndFlush(new TextWebSocketFrame(sdf.format(new Date()) + ":" + text));
            }
        }
    
        // 当有新的客户端连接服务器之后，会自动调用这个方法
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            // 将新的通道加入到clients
            clients.add(ctx.channel());
        }
    }

## 前段代码 ##

<!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>在线聊天室</title>
    </head>
    <body>
        <input type="text" id="message">
        <input type="button" value="发送消息" onclick="sendMsg()">
    
        接收到的消息:
        <p id="server_message" style="background-color: #AAAAAA"></p>
    
        <script>
            var websocket = null;
            // 判断当前浏览器是否支持websocket
            if(window.WebSocket) {
                websocket = new WebSocket("ws://127.0.0.1:9090/ws");
                websocket.onopen = function() {
                    console.log("建立连接.");
                }
                websocket.onclose = function() {
                    console.log("断开连接");
                }
                websocket.onmessage = function(e) {
                    console.log("接收到服务器消息:" + e.data);
                    var server_message = document.getElementById("server_message");
                    server_message.innerHTML += e.data + "<br/>";
                }
            }
            else {
                alert("当前浏览器不支持web socket");
            }
            function sendMsg() {
                var message = document.getElementById("message");
                websocket.send(message.value);
            }
        </script>
    </body>
    </html>

# 简单聊天应用 #

适用单节点，分布式部署可通过消息队列实现；

## 启动netty ##

/**
     * spring监听，启动netty服务器
     * 
     * @author vander
     *
     */
    @Component
    public class NettyListener implements ApplicationListener<ContextRefreshedEvent> {
    
        @Autowired
        private WebSocketServer websocketServer;
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent event) {
            if(event.getApplicationContext().getParent() == null) {
                try {
                    websocketServer.start();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
    
    @Component
    public class WebSocketServer {
    
        private EventLoopGroup bossGroup;       // 主线程池
        private EventLoopGroup workerGroup;     // 工作线程池
        private ServerBootstrap server;         // 服务器
        private ChannelFuture future;           // 回调
    
        public void start() {
            future = server.bind(9001);
            System.out.println("netty server - 启动成功");
        }
    
        public WebSocketServer() {
            bossGroup = new NioEventLoopGroup();
            workerGroup = new NioEventLoopGroup();
    
            server = new ServerBootstrap();
            server.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new WebsocketInitializer());
        }
    }

## 初始化 ##

public class WebsocketInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
    
            // ------------------
            // 用于支持Http协议
            // ------------------
    
            // websocket基于http协议，需要有http的编解码器
            pipeline.addLast(new HttpServerCodec());
            // 对写大数据流的支持
            pipeline.addLast(new ChunkedWriteHandler());
            // 添加对HTTP请求和响应的聚合器:只要使用Netty进行Http编程都需要使用
            // 对HttpMessage进行聚合，聚合成FullHttpRequest或者FullHttpResponse
            // 在netty编程中都会使用到Handler
            pipeline.addLast(new HttpObjectAggregator(1024 * 64));
    
            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    
            // 添加Netty空闲超时检查的支持
            // 1. 读空闲超时（超过一定的时间会发送对应的事件消息） 4s
            // 2. 写空闲超时  8s
            // 3. 读写空闲超时   12s
            pipeline.addLast(new IdleStateHandler(4, 8, 12));
            pipeline.addLast(new HearBeatHandler());
            
            // 添加自定义的handler
            pipeline.addLast(new ChatHandler());
    
        }
    }

## 消息处理 ##

/**
     * 处理消息的handler
     * TextWebSocketFrame: 在netty中，是用于为websocket专门处理文本的对象，frame是消息的载体
     */
    public class ChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    
        // 用来保存所有的客户端连接
        private static ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-dd hh:MM");
    
        // 当Channel中有新的事件消息会自动调用
        protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
            // 当接收到数据后会自动调用
    
            // 获取客户端发送过来的文本消息
            String text = msg.text();
            System.out.println("接收到消息数据为：" + text);
    
            Message message = JSON.parseObject(text, Message.class);
    
            // 通过SpringUtil工具类获取Spring上下文容器
            ChatRecordService chatRecordService = SpringUtil.getBean(ChatRecordService.class);
    
            switch (message.getType()) {
                // 处理客户端连接的消息
                case 0:
                    // 建立用户与通道的关联
                    String userid = message.getChatRecord().getUserid();
                    UserChannelMap.put(userid, ctx.channel());
                    System.out.println("建立用户:" + userid + "与通道" + ctx.channel().id() + "的关联");
                    UserChannelMap.print();
                    break;
                // 处理客户端发送好友消息
                case 1:
                    System.out.println("接收到用户消息");
                    // 将聊天消息保存到数据库
                    TbChatRecord chatRecord = message.getChatRecord();
                    chatRecordService.insert(chatRecord);
    
                    // 如果发送消息好友在线，可以直接将消息发送给好友
                    Channel channel = UserChannelMap.get(chatRecord.getFriendid());
                    if(channel != null) {
                        channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(message)));
                    }
                    else {
                        // 如果不在线，暂时不发送
                        System.out.println("用户" + chatRecord.getFriendid() + "不在线");
                    }
                    break;
                // 处理客户端的签收消息
                case 2:
                    // 将消息记录设置为已读
                    chatRecordService.updateStatusHasRead(message.getChatRecord().getId());
                    break;
                case 3:
                    // 接收心跳消息
                    System.out.println("接收到心跳消息:" + JSON.toJSONString(message));
                    break;
            }
    
        }
    
        // 当有新的客户端连接服务器之后，会自动调用这个方法
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            // 将新的通道加入到clients
            clients.add(ctx.channel());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            UserChannelMap.removeByChannelId(ctx.channel().id().asLongText());
            ctx.channel().close();
        }
    
        @Override
        public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
            System.out.println("关闭通道");
            UserChannelMap.removeByChannelId(ctx.channel().id().asLongText());
            UserChannelMap.print();
        }
    }

## 心跳处理 ##

检测 Channel的心跳 Handler，客户端连接空闲检测和处理

public class HearBeatHandler extends ChannelInboundHandlerAdapter {
    	//客户端一段时间没有动作会触发这个事件
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        	
            if(evt instanceof IdleStateEvent) {
                IdleStateEvent idleStateEvent = (IdleStateEvent)evt;
                if(idleStateEvent.state() == IdleState.READER_IDLE) {
                    System.out.println("读空闲事件触发...");
                }
                else if(idleStateEvent.state() == IdleState.WRITER_IDLE) {
                    System.out.println("写空闲事件触发...");
                }
                else if(idleStateEvent.state() == IdleState.ALL_IDLE) {
                    System.out.println("---------------");
                    System.out.println("读写空闲事件触发");
                    System.out.println("关闭通道资源");
                    ctx.channel().close();
                }
            }
        }
    }

## 存储所有用户通道 ##

/**
     * 建立用户ID与通道的关联
     */
    public class UserChannelMap {
        // 用户保存用户id与通道的Map对象
        private static Map<String, Channel> userChannelMap;
    
        static {
            userChannelMap = new HashMap<String, Channel>();
        }
    
        /**
         * 添加用户id与channel的关联
         * @param userid
         * @param channel
         */
        public static void put(String userid, Channel channel) {
            userChannelMap.put(userid, channel);
        }
    
        /**
         * 根据用户id移除用户id与channel的关联
         * @param userid
         */
        public static void remove(String userid) {
            userChannelMap.remove(userid);
        }
    
        /**
         * 根据通道id移除用户与channel的关联
         * @param channelId 通道的id
         */
        public static void removeByChannelId(String channelId) {
            if(!StringUtils.isNotBlank(channelId)) {
                return;
            }
    
            for (String s : userChannelMap.keySet()) {
                Channel channel = userChannelMap.get(s);
                if(channelId.equals(channel.id().asLongText())) {
                    System.out.println("客户端连接断开,取消用户" + s + "与通道" + channelId + "的关联");
                    userChannelMap.remove(s);
                    break;
                }
            }
        }
    
    
        // 打印所有的用户与通道的关联数据
        public static void print() {
            for (String s : userChannelMap.keySet()) {
                System.out.println("用户id:" + s + " 通道:" + userChannelMap.get(s).id());
            }
        }
    
        /**
         * 根据好友id获取对应的通道
         * @param friendid 好友id
         * @return Netty通道
         */
        public static Channel get(String friendid) {
            return userChannelMap.get(friendid);
        }
    }

## 消息实体类 ##

/**
     * 消息记录
     * 
     * @author vander
     *
     */
    public class TbChatRecord {
        private String id;
    
        private String userid;
    
        private String friendid;
    
        private Integer hasRead;
    
        private Date createtime;
    
        private Integer hasDelete;
    
        private String message;
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id == null ? null : id.trim();
        }
    
        public String getUserid() {
            return userid;
        }
    
        public void setUserid(String userid) {
            this.userid = userid == null ? null : userid.trim();
        }
    
        public String getFriendid() {
            return friendid;
        }
    
        public void setFriendid(String friendid) {
            this.friendid = friendid == null ? null : friendid.trim();
        }
    
        public Integer getHasRead() {
            return hasRead;
        }
    
        public void setHasRead(Integer hasRead) {
            this.hasRead = hasRead;
        }
    
        public Date getCreatetime() {
            return createtime;
        }
    
        public void setCreatetime(Date createtime) {
            this.createtime = createtime;
        }
    
        public Integer getHasDelete() {
            return hasDelete;
        }
    
        public void setHasDelete(Integer hasDelete) {
            this.hasDelete = hasDelete;
        }
    
        public String getMessage() {
            return message;
        }
    
        public void setMessage(String message) {
            this.message = message == null ? null : message.trim();
        }
    }
    
    
    public class Message {
        private Integer type; // 消息类型
        private TbChatRecord chatRecord;    // 聊天消息
        private Object ext;  // 扩展消息字段
    
        public Integer getType() {
            return type;
        }
    
        public void setType(Integer type) {
            this.type = type;
        }
    
        public TbChatRecord getChatRecord() {
            return chatRecord;
        }
    
        public void setChatRecord(TbChatRecord chatRecord) {
            this.chatRecord = chatRecord;
        }
    
        public Object getExt() {
            return ext;
        }
    
        public void setExt(Object ext) {
            this.ext = ext;
        }
    }