转载

使用netty实现websocket

1. 前言

年前有个大屏的项目是基于websocket实现的，今年又有后续的开发工作，最近就想着先把之前的代码优化一下。那么得先研究一下之前的后端实现框架 - spring-boot-starter-websocket ，这算是spring boot 官方的websocket实现框架了，但出乎意料的受到网上一致差评。有很多博主都推荐用netty来实现，正好我前几天在看nio的实现原理，就萌生了重构后端代码,用netty来实现websocket的想法。

2. nio的应用

本段是做一些技术知识点的铺垫，为了避免本末倒置，一些知识只是简单说明，详细内容你们可以找专门的书看看，每个知识点衍生出来的内容都很多。

nio模型是同步非阻塞io，它针对于传统io的优化主要体现在两点：channel中零拷贝的实现提升了资源传递效率；selector的模式提升了线程的利用率。nio模型提出的很早，现如今很多框架都完成了bio到nio的升级，例如，我们springboot开发都会用到的tomcat。

2.1. tomcat

tomcat 6 之前几乎都是bio，你可以在启动日志中看到每个执行线程名字都是以 “ http-bio-端口号-exec- ” 为前缀的。而后都是nio，对应线程名前缀是 “ http-nio-端口号-exec- ” 。后来又出现了apr，对于tomcat中 bio、nio、apr 三者之间的区别，有兴趣可以了解一下，这里就不细讲了。

因为spring boot是内嵌tomcat，而很多有意思的日志信息需要你手动来开启，可以加上如下的配置信息。

logging.level.org.apache.tomcat=debug
logging.level.org.apache.catalina=debug

2.2. netty

而基于nio模型，目前公认最有影响力的java框架就是netty，像目前大火的 dubbo、zookeeper等，都是通过netty来实现nio。建议大家还是相信大厂，在玩nio模型方面，目前还没几个玩的比netty好。

Netty中的Reactor模型主要由多路复用器(Acceptor)、事件分发器(Dispatcher)、事件处理器(Handler)组成，可以分为三种：

  1. 单线程模型：所有I/O操作都由一个线程完成，即多路复用、事件分发和处理都是在一个Reactor线程上完成的。适用于小容量场景。
  2. 多线程模型：为了解决单线程模型存在的一些问题，演化而来的Reactor线程模型。在绝大多数场景下，Reactor多线程模型都可以满足性能需求。
  3. 主从多线程模型：采用多个reactor，每个reactor都在自己单独的线程里执行。如果是多核，则可以同时响应多个客户端的请求，一旦链路建立成功就将链路注册到负责I/O读写的SubReactor线程池上。适用于大容量场景，比如海量连接处理。

Netty的线程模型并非固定不变，在启动辅助类中创建不同的EventLoopGroup实例并通过适当的参数配置，就可以支持上述三种Reactor线程模型。正是因为Netty对Reactor线程模型的支持提供了灵活的定制能力，所以可以满足不同业务场景的性能需求。

3. 技术选型

websocket只是一个协议，实现websocket协议的方式有很多种，在本次技术选型中，我尝试了三种方式：

  1. spring-boot-starter-websocket：基于tomcat容器实现websocket。
  2. netty-websocket-spring-boot-starter：基于netty实现websocket，封装成starter。
  3. netty-all：基于netty实现websocket，原生实现。

在实现websocket方式上，主要还是区别在tomcat和netty上，2和3的底层实现原理一样。

3.1. tomcat 和netty

我们来看看分别用tomcat和netty实现websocket之后，程序的线程运行情况。

1、（spring-boot-starter-websocket）线程监控

使用netty实现websocket

2、（netty）线程监控

使用netty实现websocket

在图一中，对于tomcat，在nio多路复用上是典型的单线程模型。由一个一直轮询的poller线程来连接请求（另一个是处理阻塞的），然后将封装后的任务交给worker线程池，即图中 http-nio-9001-exec- 。而且这时，tomcat处理的请求，既包括传统的http请求，还包括websocket。

在图二中，因为是在springboot项目中使用netty，因此即能看到tomcat的nio实现，也能看到netty的Reactor模型（nioEventLoopGroup-x-x）。我的设计是，专门用netty的nio来处理websocket，其他的交给tomcat。针对高并发的websocket场景，我还可以选用netty的主从多线程模型来应对。

二者对比上，主要体现在netty对于nio的实现更好，不仅体现在图示中的线程模式上，还包括在底层的数据拷贝。而且如果程序中websocket的应用规模比较大的话，无论是tomcat还是netty，都建议独立出一个连接，单独映射一个端口。如果http短连接和websocket长连接混用一套线程模型，很多时候不好针对单独一方做性能调优。

3.2. netty-websocket-spring-boot-starter和netty-all

netty-websocket-spring-boot-starter的内部实现其实就是netty-all，只是一位程序员好心的将netty-all封装了一下。封装后，netty-websocket-spring-boot-starter的代码实现，基本上和spring-boot-starter-websocket就没什么区别了，让你基本看不到netty的影子。

这样很好吗？开发上可能更简单了，但是这样的封装实在太强行了。强行往官方的spring-boot-starter-websocket上靠，失去了自己的灵活性，有些时候还容易带来很多误解。我在使用netty-websocket-spring-boot-starter之后，就是被它的名词概念绕来绕去。最后实在忍不住，删掉代码，直接基于netty-all来重构。

4. 代码参考

按道理，我应该分享一份“干净”的demo代码，让大家可以直接运行的。但最近实在无心写代码了，所以直接将项目代码里面websocket的部分粘贴下来，核心内容都在。

4.1.pom.xml

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

4.2.ChannelGroupUtil.java

@Slf4j
@Service
public class ChannelGroupUtil {
    public static final String KEY_USER_ID = "userId";
    public static final AttributeKey<String> ATTRIBUTE_KEY_USER_ID = AttributeKey.valueOf(KEY_USER_ID);

    private static ChannelGroup gChannelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private static ConcurrentHashMap<String, Channel> gUserChannelMap = new ConcurrentHashMap<>();

    /**
     * 新增 Channel
     *
     * @param channel
     */
    public void addChannel(Channel channel) {
        gChannelGroup.add(channel);
    }

    /**
     * 关闭 Channel
     *
     * @param channel
     */
    public void removeChannel(Channel channel) {
        gChannelGroup.remove(channel);
        channel.close();
    }

    /**
     * 关闭 userId的所有 channel
     * @param userId
     */
    public void removeChannel(String userId) {
        if (userId != null) {
            gChannelGroup.stream()
                    .forEach((Channel channel) -> {
                        String channelUserId = channel.attr(ChannelGroupUtil.ATTRIBUTE_KEY_USER_ID).get();
                        if (userId.equals(channelUserId)) {
                            gChannelGroup.remove(channel);
                            channel.close();
                        }
                    });
        }
    }


    /**
     * 用户 注册
     * 1、设置 Channel 中 userId属性
     * 2、存放在 gUserChannelMap 中
     *
     * @param channel
     * @param userId
     */
    public void userRegister(Channel channel, String userId) {
        this.setChannelUserAttribute(channel, userId);
        gUserChannelMap.put(userId, channel);
        log.info(userId + " 已连接");
    }

    /**
     * 用户 注销
     * 1、已注册channel 关闭
     * 2、未注册channel 关闭
     *
     * @param channel
     */
    public void userRemove(Channel channel) {
        String userId = this.getUserIdByChannel(channel);
        if(userId!=null){
            gUserChannelMap.remove(userId);
            this.removeChannel(userId);
        }else {
            this.removeChannel(channel);
        }
        log.info(userId + " 已退出");
    }


    /**
     * 获取用户 对应的 Channel
     *
     * @param userId
     * @return
     */
    public Channel getChannelByUserId(String userId) {
        return gUserChannelMap.get(userId);
    }


    /**
     * 给 Channel 设置用户名 userId 的属性
     *
     * @param channel
     * @param userId
     */
    private void setChannelUserAttribute(Channel channel, String userId) {
        channel.attr(ATTRIBUTE_KEY_USER_ID).setIfAbsent(userId);
    }

    /**
     * 获取 channel 的userId
     *
     * @param channel
     * @return
     */
    public String getUserIdByChannel(Channel channel) {
        return channel.attr(ATTRIBUTE_KEY_USER_ID).get();
    }


    public ChannelGroup getChannelGroup() {
        return gChannelGroup;
    }

    public ConcurrentHashMap<String, Channel> getUserChannelMap() {
        return gUserChannelMap;
    }
}

4.3. ChannelHandlerBase.java

public class ChannelHandlerBase extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Autowired
    private ChannelGroupUtil channelGroupUtil;
    @Autowired
    private MessageSender messageSender;

    /**
     * 连接开启
     * @param channelHandlerContext
     */
    @Override
    public void handlerAdded(ChannelHandlerContext channelHandlerContext) {
        channelGroupUtil.addChannel(channelHandlerContext.channel());
    }

    /**
     * 接受消息
     *  由子类实现
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg)  {
    }

    /**
     * 连接关闭
     * @param channelHandlerContext
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext channelHandlerContext) {
        channelGroupUtil.userRemove(channelHandlerContext.channel());
    }

    /**
     * 异常处理
     * @param channelHandlerContext
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable cause) {
        channelGroupUtil.userRemove(channelHandlerContext.channel());
    }

    /**
     * 心跳检查处理 HEART_CHECK
     * @param channel
     */
    protected void heartCheckHandler(Channel channel){
        messageSender.sendMessage(channel,
                WebSocketDTO.data(ModuleEnum.HEART_CHECK, "回复心跳检查"));
    }

    /**
     * 注册用户 USER_REGISTER
     * @param channel
     * @param request
     * @param webSocketEnum
     */
    protected void registerHandler(Channel channel, WebSocketDTO request, WebSocketEnum webSocketEnum){
        UserRegisterDataDTO userRegisterDataDTO =JSON.parseObject(JSON.toJSONString(request.getData()),UserRegisterDataDTO.class);
        String userId= userRegisterDataDTO.getUsername()+"_"+webSocketEnum.name()+"_"+ userRegisterDataDTO.getDevice().name();
        //关闭 userId 的其他channel
        channelGroupUtil.removeChannel(userId);
        //将该 channel绑定userId，存入UserChannelMap
        channelGroupUtil.userRegister(channel,userId);
        messageSender.sendMessage(channel,
                WebSocketDTO.data(ModuleEnum.USER_REGISTER, userId+"用户已注册"));
    }
}

4.4. DataChannelHandler.java

@Component
@ChannelHandler.Sharable
public class DataChannelHandler extends ChannelHandlerBase {


    @Override
    public void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame msg) {
        Channel channel = channelHandlerContext.channel();
        WebSocketDTO request = JSON.parseObject(msg.text(), WebSocketDTO.class);
        switch (request.getModule()) {
            case USER_REGISTER:
                super.registerHandler(channel, request, WebSocketEnum.DATA);
                break;
            case HEART_CHECK:
                super.heartCheckHandler(channel);
                break;
            default:
                break;
        }
    }

}

4.5. MessageSender.java

@Service
public class MessageSender {
    @Autowired
    private ChannelGroupUtil channelGroupUtil;

    /**
     * 发送给 具体用户
     *
     * @param userId
     * @param webSocketDTO
     */
    public void sendMessage(String userId, WebSocketDTO webSocketDTO) {
        Channel channel = channelGroupUtil.getChannelByUserId(userId);
        this.sendMessage(channel,webSocketDTO);
    }

    public void sendMessage(String username, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum, WebSocketDTO webSocketDTO) {
        String userId = username + "_" + webSocketEnum.name() + "_" + deviceEnum.name();
        this.sendMessage(userId, webSocketDTO);
    }

    public void sendMessage(Channel channel, WebSocketDTO webSocketDTO) {
        if(channel!=null) {
            channel.writeAndFlush(new TextWebSocketFrame(JSON.toJSONString(webSocketDTO)));
        }
    }

    /**
     * 发送给 该用户下的所有设备
     *
     * @param username
     * @param webSocketEnum
     * @param webSocketDTO
     */
    public void sendMessageToAllDevices(String username, WebSocketEnum webSocketEnum, WebSocketDTO webSocketDTO) {
        for (DeviceEnum deviceEnum : DeviceEnum.values()) {
            String userId = username + "_" + webSocketEnum.name() + "_" + deviceEnum.name();
            this.sendMessage(userId, webSocketDTO);
        }
    }


    /**
     * 转换 userId
     *
     * @param userId
     * @param webSocketEnum
     * @param deviceEnum
     * @return
     */
    public String convertUserId(String userId, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum) {
        if (userId != null) {
            if (webSocketEnum != null) {
                for (WebSocketEnum we : WebSocketEnum.values()) {
                    if (!we.equals(webSocketEnum)) {
                        userId= userId.replace("_" + we.name(), "_" + webSocketEnum.name());
                    }
                }
            }

            if (deviceEnum != null) {
                for (DeviceEnum de : DeviceEnum.values()) {
                    if (!de.equals(deviceEnum)) {
                        userId=userId.replace("_" + de.name(), "_" + deviceEnum.name());
                    }
                }
            }

        }
        return userId;
    }

    public String convertUserId(Channel channel, WebSocketEnum webSocketEnum, DeviceEnum deviceEnum) {

        String userId = channelGroupUtil.getUserIdByChannel(channel);

        return convertUserId(userId, webSocketEnum, deviceEnum);
    }
}

4.6. DataNettyServer.java

@Slf4j
@Component
public class DataNettyServer {
    @Autowired
    private DataChannelHandler dataChannelHandler;

    private static final String WEB_SOCKET_PROTOCOL = "WebSocket";
    private static final String THREAD_NAME_PREFIX="nettyStart-";
    @Value("${ws.data.port}")
    private int port;
    @Value("${ws.data.webSocketPath}")
    private String webSocketPath;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workGroup;



    /**
     * 启动
     *
     * @throws InterruptedException
     */
    private void start() throws InterruptedException {
        this.bossGroup= new NioEventLoopGroup();
        this.workGroup= new NioEventLoopGroup(16);
        ServerBootstrap bootstrap = new ServerBootstrap();
        // bossGroup辅助客户端的tcp连接请求, workGroup负责与客户端之前的读写操作
        bootstrap.group(this.bossGroup, this.workGroup);
        // 设置NIO类型的channel
        bootstrap.channel(NioServerSocketChannel.class);
        // 设置监听端口
        bootstrap.localAddress(new InetSocketAddress(this.port));
        // 连接到达时会创建一个通道
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch)  {
                // 流水线管理通道中的处理程序（Handler），用来处理业务
                // webSocket协议本身是基于http协议的，所以这边也要使用http编解码器
                ch.pipeline().addLast(new HttpServerCodec());
                ch.pipeline().addLast(new ObjectEncoder());
                // 以块的方式来写的处理器
                ch.pipeline().addLast(new ChunkedWriteHandler());
                //http数据在传输过程中是分段的，HttpObjectAggregator可以将多个段聚合
                ch.pipeline().addLast(new HttpObjectAggregator(8192));
                ch.pipeline().addLast(new WebSocketServerProtocolHandler(webSocketPath, WEB_SOCKET_PROTOCOL, true, 65536 * 10));
                // 自定义的handler，处理业务逻辑
                ch.pipeline().addLast(dataChannelHandler);

            }
        });
        // 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
        ChannelFuture channelFuture = bootstrap.bind().sync();
        // 对关闭通道进行监听
        channelFuture.channel().closeFuture().sync();
    }

    @PreDestroy
    public void destroy() throws InterruptedException {
        if (this.bossGroup != null) {
            this.bossGroup.shutdownGracefully().sync();
        }
        if (this.workGroup != null) {
            this.workGroup.shutdownGracefully().sync();
        }
    }

    @PostConstruct
    public void init() {
        //需要开启一个新的线程来执行netty server 服务器
        Thread initThread = new Thread(() -> {
            try {
                start();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        initThread.setName(THREAD_NAME_PREFIX+ WebSocketEnum.DATA.name());
        initThread.start();
    }
}
原文  https://segmentfault.com/a/1190000022610086
正文到此结束
所属分类： 编程技术 Java

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 springcloud-demo Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 如何将Spring Cloud Task发布Data Flow上执行？
  2. 2 Spring Cloud Data Flow快速入门Demo
  3. 3 Spring Cloud Consul实现选举机制
  4. 4 Spring Boot集成ShedLock实现分布式定时任务
  5. 5 利用oss进行数据库和网站图片备份
  6. 6 Spring Cloud Stream实现数据流处理
  7. 7 Python3访问MySQL数据库快速入门Demo
  8. 8 Github开源项目作者可以免费申请 JetBrains 全家桶
  9. 9 Spring Cloud Vault快速入门Demo
  10. 10 Spring Cloud Gateway快速入门Demo
网站信息
  • 文章总数：82,716 篇
  • 文件总数：284,297 个
  • 标签总数：2,396 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：656 人
  • 运行天数：4,410天
  • 最后更新：2024年11月23日17点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG