springboot基于netty加websocket实现高性能IM系统案例（1）

IM通讯系统

即时通讯（IM）系统是一种允许用户实时地进行通信和交流的软件系统。构建一个IM系统涉及到多个技术挑战，包括但不限于消息的实时传递、系统的可伸缩性、安全性、消息顺序性以及用户体验，整个实现起来相对繁琐。要考虑多个方面，随便举几个例子，例如：

消息有序性

保证消息在推送到客户端之后，消息是按照规定的顺序排列的。可参考方案如下：

发送时间作为排序的标准，但是客户端的时间是可以自己去修改的，这也就导致了不确定性
雪花算法：用生成的key去做排序，生成的key是趋势递增的，不是绝对递增的，在一定的场景下，他还是可能导致消息的乱序，例如时间回拨，跨数据中心
服务端可以用一些手段生成绝对递增的序列号，比如使用redis，但是比较依赖redis的可用性

如果对这方面要求极高，同时并发量大到一定程度的系统推荐选择方案3，一般场景下方案2能解决大部分问题。

消息实时性

保证消息发送后快速推送到指定客户端，延迟小。解决方案一般有：

利用多线程解决消息串行的问题，提高处理效率
异步解耦存储逻辑，利用消息队列延迟存储消息
增加前置过滤操作，避免进一步浪费资源，比如是否是好友，是否禁用中，如果是结束

消息幂等性

保证消息不会被重复消费或者因网络问题重复发送，解决方案为：

im服务端搞点文章，比如第一次处理该消息的时候，可以将它缓存到redis中（设置过期时间），当第二次处理的时候，可以从redis寻找这个消息，如果找到了就说明处理过了，所以就不二次持久化了，只用去同步消息即可
我们也可以在客户端做一些改造，比如说重传的消息都会是同一个messageId（可以当做上面那个查询redis的依据），客户端收到多条messageId的消息，可以过滤掉重复的，只显示一条消息即可
如果说一条消息，重传了一定的时间段后，还没有收到ack的话，就可以将它放弃了（就像微信没有网络，最后出现一个红色的感叹号），当我们再手动点击红色的感叹号，sdk就会生成一个新的id和旧的消息体，再次去发送

这里说得都是单人聊天，群聊同理，只不过要在逻辑变一些东西

另外这只是一部分，完整的系统还要考虑很多，例如身份校验，消费已读未读，用户状态的维护，多端的数据同步等，这里就不详细说了；

websocket

长连接和短连接

HTTP的长连接和短连接本质上是TCP长连接和短连接。HTTP属于应用层协议，在传输层使用TCP协议，在网络层使用IP协议。IP协议主要解决网络路由和寻址问题，TCP协议主要解决如何在IP层之上可靠的传递数据包，使在网络上的另一端收到发端发出的所有包，并且顺序与发出顺序一致。TCP有可靠，面向连接的特点。

长连接用通俗易懂的话来说，就是客户端不停的向服务器发送请求以获取最新的数据信息。这里的“不停”其实是有停止的，只是我们人眼无法分辨是否停止，它只是一种快速的停下然后又立即开始连接而已

什么时候要用长连接

长连接多用于操作频繁，点对点的通讯，而且连接数不能太多情况，。每个TCP连接都需要三步握手，这需要时间，如果每个操作都是先连接，再操作的话那么处理速度会降低很多，所以每个操作完后都不断开，次处理时直接发送数据包就OK了，不用建立TCP连接。例如：数据库的连接用长连接，如果用短连接频繁的通信会造成socket错误，而且频繁的socket 创建也是对资源的浪费。

　　而像WEB网站的http服务一般都用短连接，因为长连接对于服务端来说会耗费一定的资源，而像WEB网站这么频繁的成千上万甚至上亿客户端的连接用短连接会更省一些资源，如果用长连接，而且同时有成千上万的用户，如果每个用户都占用一个连接的话，那可想而知吧。所以并发量大，但每个用户无需频繁操作情况下需用短连好。

websocket协议

Websocket是html5提出的一个协议规范，websocket约定了一个通信的规范，通过一个握手的机制，客户端（浏览器）和服务器（webserver）之间能建立一个类似tcp的连接，从而方便c－s之间的通信。在websocket出现之前，web交互一般是基于http协议的短连接或者长连接。

WebSocket是为解决客户端与服务端实时通信而产生的技术。websocket协议本质上是一个基于tcp的协议，是先通过HTTP/HTTPS协议发起一条特殊的http请求进行握手后创建一个用于交换数据的TCP连接，此后服务端与客户端通过此TCP连接进行实时通信。

websocket的一大特点就是双向通信，这意味着客户端和服务器可以同时独立地向对方发送数据，而不需要等待对方的响应。这与传统的 HTTP 请求/响应模式不同，在传统模式中，客户端必须等待服务器的响应后才能发送下一个请求。

实战

package com.imooc.bilibili.service.websocket;
import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.domain.Danmu;
import com.imooc.bilibili.domain.constant.UserMomentsConstant;
import com.imooc.bilibili.service.DanmuService;
import com.imooc.bilibili.service.util.RocketMQUtil;
import com.imooc.bilibili.service.util.TokenUtil;
import io.netty.util.internal.StringUtil;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
@Component
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {

    private final Logger logger =  LoggerFactory.getLogger(this.getClass());

    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    private Session session;

    private String sessionId;

    private Long userId;

    private static ApplicationContext APPLICATION_CONTEXT;

    public static void setApplicationContext(ApplicationContext applicationContext){
        WebSocketService.APPLICATION_CONTEXT = applicationContext;
    }

    @OnOpen
    //从前端获取token获取发送人并保存
    public void openConnection(Session session, @PathParam("token") String token){
        try{
            this.userId = TokenUtil.verifyToken(token);
        }catch (Exception ignored){}
        this.sessionId = session.getId();
        this.session = session;
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        }else{
            WEBSOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement();
        }
        logger.info("用户连接成功：" + sessionId + "，当前在线人数为：" + ONLINE_COUNT.get());
        try{
            this.sendMessage("0");
        }catch (Exception e){
            logger.error("连接异常");
        }
    }

    @OnClose
    public void closeConnection(){
        if(WEBSOCKET_MAP.containsKey(sessionId)){
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出：" + sessionId + "当前在线人数为：" + ONLINE_COUNT.get());
    }

    @OnMessage
    public void onMessage(String message){
        logger.info("用户信息：" + sessionId + "，报文：" + message);
        if(!StringUtil.isNullOrEmpty(message)){
            try{
                //群发消息
                for(Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()){
                    WebSocketService webSocketService = entry.getValue();
                    DefaultMQProducer danmusProducer = (DefaultMQProducer)APPLICATION_CONTEXT.getBean("danmusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("message", message);
                    jsonObject.put("sessionId", webSocketService.getSessionId());
                    Message msg = new Message(UserMomentsConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RocketMQUtil.asyncSendMsg(danmusProducer, msg);
                }
                if(this.userId != null){
                    //保存弹幕到数据库
                    Danmu danmu = JSONObject.parseObject(message, Danmu.class);
                    danmu.setUserId(userId);
                    danmu.setCreateTime(new Date());
                    DanmuService danmuService = (DanmuService)APPLICATION_CONTEXT.getBean("danmuService");
                    danmuService.asyncAddDanmu(danmu);
                    //保存弹幕到redis
                    danmuService.addDanmusToRedis(danmu);
                }
            }catch (Exception e){
                logger.error("弹幕接收出现问题");
                e.printStackTrace();
            }
        }
    }

    @OnError
    public void onError(Throwable error){
    }

    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    //或直接指定时间间隔，例如：5秒
    @Scheduled(fixedRate=5000)
    private void noticeOnlineCount() throws IOException {
        for(Map.Entry<String, WebSocketService> entry : WebSocketService.WEBSOCKET_MAP.entrySet()){
            WebSocketService webSocketService = entry.getValue();
            if(webSocketService.session.isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onlineCount", ONLINE_COUNT.get());
                jsonObject.put("msg", "当前在线人数为" + ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }
    //私有属性,需要生成get方法
    public Session getSession() {
        return session;
    }

    public String getSessionId() {
        return sessionId;
    }
}

以上为一个接受弹幕的websocket服务，当用户建立连接后，触发OnOpen方法，执行初始化操作，这里是记录实时人数并将session存储到Map里面管理，因为这属于并发场景，要用原子类，因为websocket服务属于多例模式，所以要对每一个服务类单独管理，执行操作；进入当用户端进行消息发送后，服务端执行OnMessage方法，进行消息的异步推送和存储，这里使用mq进行异步，推送逻辑为找到当前所有在线的用户，根据sessionId进行退送，这里的sessionId可以理解为每个用户的唯一标识，关联着会话；实时推送消息调用本类中的sendMessage方法，注意客户端往服务端发送消息要转化为json格式，当用户断开连接时，执行人数减操作；@Scheduled(fixedRate=5000)为websocket内的定时任务。这是一个纯websocket案例，在业务比较简单的情况下，可以采用这种模式，但当业务和消息包都比较多且复杂时，这种写法就不是很推荐。

netty

Netty 是一个 NIO 客户端服务器框架，可快速轻松地开发网络应用程序，例如协议服务器和客户端。它极大地简化和简化了网络编程，例如 TCP 和 UDP 套接字服务器。

阻塞和非阻塞

阻塞就是把当前线程挂起，直到结果返回才会恢复运行。

非阻塞就是得到结果之前，可以进行其它操作而不会阻塞当前线程

常见的IO模型

常见的IO模型有BIO、NIO、AIO、IO复用。

BIO就是常规的IO操作，同步阻塞，适合少请求的场景。

NIO就是单线程能处理多个请求，当有事件发生会通知去处理，同步非阻塞，适合大量并发请求的场景。

AIO就是NIO的升级版，通过操作系统的回调通知去让线程处理事件，异步非阻塞，适合超大并发请求的场景。

IO复用就是基于select、poll、epoll机制去实现，适合大量并发连接的场景select基于数组实现，每次调用都进行遍历，最大连接数有上限。

poll通过链表实现，每次调用都进行遍历，最大连接数无上限。

epoll通过哈希表实现，通过事件通知，每当有IO事件就绪，系统注册的回调函数就会被调用，最大连接数无上限。

在选择三者上需要根据实际场景进行选择，表面上看epol性能最好，但是如果在连接数少并且活跃度较低时，select和poll的性能最好，低效的原因主要在轮询遍历上，所以也得视情况看

为什么Netty选择NIO而非AIO

在Linux系统上，AIO的底层实现仍然使用的是epoll，没有很好的实现API，在性能上没有明显的优势。同时Linux上AIO不是很成熟，处理回调结果的速度跟不上处理需求。
Netty整体架构采用的是Reactor模型，而AIO采用的是Proactor模型，整合相对较乱
AIO接收数据时采用的是预分配缓存，而NIO采用的是需要接收时才分配缓存。AIO在连接数相对较大，而实际流量小的情况下会浪费内存

netty核心组件

Channel

Channel是 Java NIO 的一个基本构造。可以看作是传入或传出数据的载体。因此，它可以被打开或关闭，连接或者断开连接。

channelcloseFuture()用来处理channel的关闭

sync方法作用是同步等待channel关闭

而addListener方法是异步等待channel关闭

pipeline()方法添加处理器

write()方法将数据写入

writeAndFlush(）方法将数据写入并刷出

EventLoop 与 EventLoopGroup

EventLoop 定义了Netty的核心抽象，用来处理连接的生命周期中所发生的事件，在内部，将会为每个Channel分配一个EventLoop。

EventLoopGroup 是一个 EventLoop 池，包含很多的 EventLoop，可理解为一个线程池

Netty 为每个 Channel 分配了一个 EventLoop，用于处理用户连接请求、对用户请求的处理等所有事件。EventLoop 本身只是一个线程驱动，在其生命周期内只会绑定一个线程，让该线程处理一个 Channel 的所有 IO 事件。

一个 Channel 一旦与一个 EventLoop 相绑定，那么在 Channel 的整个生命周期内是不能改变的。一个 EventLoop 可以与多个 Channel 绑定。即 Channel 与 EventLoop 的关系是 n:1，而 EventLoop 与线程的关系是 1:1。

一般会定义两个EventLoop一个负责 accept事件一个负责read&write事件

ServerBootstrap 与 Bootstrap

Bootstarp 和 ServerBootstrap 被称为引导类，指对应用程序进行配置，并使他运行起来的过程。Netty处理引导的方式是使你的应用程序和网络层相隔离。

Bootstrap 是客户端的引导类，Bootstrap 在调用 bind()（连接UDP）和 connect()（连接TCP）方法时，会新创建一个 Channel，仅创建一个单独的、没有父 Channel 的 Channel 来实现所有的网络交换。

ServerBootstrap 是服务端的引导类，ServerBootstarp 在调用 bind() 方法时会创建一个 ServerChannel 来接受来自客户端的连接，并且该 ServerChannel 管理了多个子 Channel 用于同客户端之间的通信。

ChannelHandler 与 ChannelPipeline

ChannelHandler 是对 Channel 中数据的处理器，这些处理器可以是系统本身定义好的编解码器，也可以是用户自定义的。这些处理器会被统一添加到一个 ChannelPipeline 的对象中，然后按照添加的顺序对 Channel 中的数据进行依次处理。

ChannelFuture

Netty 中所有的 I/O 操作都是异步的，即操作不会立即得到返回结果，所以 Netty 中定义了一个 ChannelFuture 对象作为这个异步操作的“代言人”，表示异步操作本身。如果想获取到该异步操作的返回值，可以通过该异步操作对象的addListener() 方法为该异步操作添加监 NIO 网络编程框架 Netty 听器，为其注册回调：当结果出来后马上调用执行。

Netty 的异步编程模型都是建立在 Future 与回调概念之上的。

netty结合websocket实现IM服务示例

项目基于DDD脚手架搭建

netty启动类

package cn.bugstack.trigger;
import cn.bugstack.domain.chat.service.encoder.WebsocketEncoder;
import cn.bugstack.domain.chat.service.handler.WebSocketHandler;
import cn.bugstack.domain.chat.service.handler.WsSharkHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
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;
import lombok.Data;
import org.springframework.context.ConfigurableApplicationContext;

/**
 * @author: Larry
 * @Date: 2024 /04 /29 / 16:30
 * @Description:
 */
@Data
public class NettyServer {
    //指定监听的端口
    private int port;

    public NettyServer(int port) {
        this.port = port;
    }
    public static void setApplicationContext(ConfigurableApplicationContext configurableApplicationContext) {

    }

    public  void start(int port) {
        //配置服务端NIO线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(); //NioEventLoopGroup extends MultithreadEventLoopGroup Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            System.out.println("netty 初始化");
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)    //非阻塞模式
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //保活开关2h没有数据服务端会发送心跳包
                    .option(ChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    //打印日志，方便观察
                    //因为基于http协议 使用http的编码和解码器
                    ch.pipeline().addLast(new HttpServerCodec());
                    //是以块方式写 添加处理器
                    ch.pipeline().addLast(new ChunkedWriteHandler());
                    //http数据在传输过程中是分段 就是可以将多个段聚合 这就是为什么当浏览器发生大量数据时 就会发生多次http请求
                    ch.pipeline().addLast(new HttpObjectAggregator(8192));
                    ch.pipeline().addLast(new WebsocketEncoder());
                    ch.pipeline().addLast(new WsSharkHandler());
                    ch.pipeline().addLast(new WebSocketHandler());
//                    ch.pipeline().addLast(
//                            new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS),new NettyClientHandler()
//                    );
                    ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
                }
            });
            //基于钩子函数优雅关闭
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }));
            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

其中，

WebSocketEncoder

是一个用于将 Java 对象编码为

WebSocket

帧的

ChannelOutboundHandler

。它实现了

MessageToMessageEncoder

接口，允许开发者将自定义的业务对象直接写入到

WebSocket

连接中，而不需要关心底层的

WebSocket

协议细节。即在向服务端发送消息时，他为自动帮你完成格式转换

WsSharkHandler是一个基于 Netty 的 WebSocket 握手处理器，它继承自

ChannelInboundHandlerAdapter

。这个处理器主要进行一些前置操作，如将用户channel存入缓存，解析URL，过滤不合法连接，类似于上文中的OnOpen;在刚建立连接后调用

package cn.bugstack.domain.chat.service.handler;

import cn.bugstack.domain.chat.cache.ChannelHandlerContextCache;
import cn.bugstack.domain.chat.model.entity.ImMsgEntity;
import cn.bugstack.domain.chat.service.handler.impl.ImHandlerFactoryImpl;
import cn.bugstack.domain.chat.service.handler.impl.UserMessageHandler;
import cn.bugstack.domain.chat.util.ImContextUtils;
import com.alibaba.fastjson.JSON;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: Larry
 * @Date: 2024 /05 /03 / 0:11
 * @Description:
 */
@Component
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Autowired
    private ImHandlerFactory imHandlerFactory;
    private static WebSocketHandler webSocketHandler;

    public void setImHandlerFactory(ImHandlerFactory imHandlerFactory) {
         this.imHandlerFactory = imHandlerFactory;
    }
    @PostConstruct     //关键二   通过@PostConstruct 和 @PreDestroy 方法 实现初始化和销毁bean之前进行的操作
    public void init() {
        webSocketHandler = this;
        webSocketHandler.imHandlerFactory = imHandlerFactory;   // 初使化时将已静态化的testService实例化
    }
    private Bootstrap bootstrap;
    private EventLoop eventLoop;
    private final Logger logger = LoggerFactory.getLogger(WebSocketHandler.class);
    /**
     * 存储已经登录用户的channel对象
     */
    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============================start=================");
        logger.info("与客户端建立连接，通道开启！");
        channelGroup.add(ctx.channel());
        ctx.channel().id();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("与客户端断开连接，通道关闭！");
        //添加到channelGroup 通道组
        channelGroup.remove(ctx.channel());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof HttpRequest) {
            HttpRequest httpRequest = (HttpRequest) msg;
            // 处理 HTTP 请求
            System.out.println("Received HTTP request: " + httpRequest.method() + " " + httpRequest.uri());
            // 如果需要，可以在这里处理 WebSocket 握手逻辑
        } else if (msg instanceof WebSocketFrame) {
            WebSocketFrame webSocketFrame = (WebSocketFrame) msg;
            // 处理 WebSocket 消息
            if (webSocketFrame instanceof TextWebSocketFrame) {
                TextWebSocketFrame frame = (TextWebSocketFrame) webSocketFrame;
                System.out.println(frame.text());
                ImMsgEntity imMsgEntity = JSON.parseObject(frame.text(), ImMsgEntity.class);
                webSocketHandler.imHandlerFactory.doMsgHandler(ctx,imMsgEntity);
            } else if (webSocketFrame instanceof BinaryWebSocketFrame) {
                BinaryWebSocketFrame binaryWebSocketFrame = (BinaryWebSocketFrame) webSocketFrame;
                ByteBuf byteBuf = binaryWebSocketFrame.content();
                // 处理二进制 WebSocket 消息
                // 注意：在处理二进制消息时，需要根据实际情况解码消息内容
            }
        }

        super.channelRead(ctx, msg);
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception {

    }

    private static Integer getUrlParams(String url) {
        if (!url.contains("=")) {
            return null;
        }
        String userId = url.substring(url.indexOf("=") + 1);
        return Integer.parseInt(userId);
    }
}

WebsocketHandler为websocket处理器，当完成握手之后进入，主要是处理业务的各种逻辑，是业务的核心类,其中的imHandlerFactory工厂为自定义工厂类，用来处理不同心跳包的逻辑；

WebSocketServerProtocolHandler为对websocket一些配置的处理，

Runtime.getRuntime().addShutdownHook(new Thread(() -> { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }));

这里通过调用

Runtime.getRuntime().addShutdownHook()

方法添加了一个 JVM 关闭钩子。当 JVM 正常关闭时，会执行这个钩子中的代码。钩子中的 lambda 表达式创建了一个新的线程，在该线程中调用了两个

EventLoopGroup

（

bossGroup

和

workerGroup

）的

shutdownGracefully()

方法。这是 Netty 优雅关闭服务器的推荐做法，它会使得所有正在进行的 I/O 操作有机会在关闭之前完成。

绑定端口并启动服务器：ChannelFuture f = b.bind(port).sync();这里 b.bind(port) 启动了服务器，绑定到指定的端口上。sync() 方法调用使得当前线程阻塞，直到绑定操作完成。ChannelFuture 对象 f 用于表示这个异步操作的结果。
等待服务器端口关闭：f.channel().closeFuture().sync();``````closeFuture().sync() 调用使得当前线程阻塞，直到服务器的端口关闭。这是必要的，因为如果不等待关闭完成，应用程序可能会提前退出，导致服务器不能正确地关闭。
异常处理：catch (InterruptedException e) { e.printStackTrace(); }如果当前线程在 sync() 调用中等待时被中断，会抛出 InterruptedException。这里通过 catch 块捕获了这个异常，并打印了堆栈跟踪。
finally 块：finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }``````finally 块确保了无论发生什么异常，bossGroup 和 workerGroup 都能被优雅地关闭。这是非常重要的，因为它可以释放资源，如线程、打开的文件等。

这段代码确保了 Netty 服务器可以在 JVM 关闭时或出现异常时，安全地关闭其资源和正在处理的连接。

结语

此篇文章主要解决im系统的设计、注意点和相关技术栈，和该项目的基础配置，下一章讲一下项目实现的思路和具体流程。部分设计上的点后续会提到。

标签： websocket 网络协议网络

本文转载自: https://blog.csdn.net/zy11517/article/details/138906265
版权归原作者 不负朝阳 所有，如有侵权，请联系我们删除。