springboot整合Netty应用

Posted Javachichi

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了springboot整合Netty应用相关的知识,希望对你有一定的参考价值。

什么是Netty

Netty是一个异步事件驱动的网络应用程序框架,用于快速开发可维护的高性能协议服务器和客户端,Netty是一个NIO客户端服务器框架,可以快速轻松地开发网络应用程序,例如协议服务器和客户端。它极大地简化和简化了诸如TCP和UDP套接字服务器之类的网络编程。

“快速简便”并不意味着最终的应用程序将遭受可维护性或性能问题的困扰。Netty经过精心设计,结合了许多协议(例如FTP,SMTP,HTTP以及各种基于二进制和文本的旧式协议)的实施经验。结果,Netty成功地找到了一种无需妥协即可轻松实现开发,性能,稳定性和灵活性的方法。

特性

  • 高性能 事件驱动
  • 异步非堵塞 基于NIO的客户端,服务器端编程框架
  • 稳定性和伸缩性
  • 适用于各种传输类型的统一API-阻塞和非阻塞套接字
  • 基于灵活且可扩展的事件模型,可将关注点明确分离
  • 高度可定制的线程模型-单线程,一个活多个线程池 ,例如SEDA
  • 真正的无连接数据报套接字支持 (从3.1开始)

表现

  • 更高的吞吐量
  • 减少资源消耗
  • 减少不必要的内存复制

使用
在pom.xml中添加依赖

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

一丶Springboot应用启动加载Netty应用

暴露Netty端口,随springboot应用一并启动

/**
 * @dete: 2021/4/21 9:08 上午
 * @author: 徐子木
 */
@SpringBootApplication
public class PmSocketApplication implements CommandLineRunner {
    
// yml中指定netty端口号
    @Value("${netty.port}")
    private int nettyServerPort;

    @Autowired
    private NettyWebSocketServer nettyServer;

    public static void main(String[] args) {
        SpringApplication.run(PmSocketApplication.class, args);
    }


    @Override
    public void run(String... args) throws Exception {
        //netty 服务端启动的端口不可和Springboot启动类的端口号重复
        nettyServer.start(nettyServerPort);

        //关闭服务器的时候同时关闭Netty服务
         Runtime.getRuntime().addShutdownHook(new Thread(() -> nettyServer.destroy()));
    }
}

二丶Netty整合WebSocket服务端

启动或销毁netty服务端的过程实现

/**
 * Netty整合websocket 服务端
 * 运行流程:
 * 1.创建一个ServerBootstrap的实例引导和绑定服务器
 * 2.创建并分配一个NioEventLoopGroup实例以进行事件的处理,比如接收连接和读写数据
 * 3.指定服务器绑定的本地的InetSocketAddress
 * 4.使用一个NettyServerHandler的实例初始化每一个新的Channel
 * 5.调用ServerBootstrap.bind()方法以绑定服务器
 *
 * @description
 * @author: 徐子木
 * @create: 2020-06-02 14:23
 **/
@Component
@Slf4j
public class NettyWebSocketServer {


    /**
     * EventLoop接口
     * NioEventLoop中维护了一个线程和任务队列,支持异步提交任务,线程启动时会调用NioEventLoop的run方法,执行I/O任务和非I/O任务
     * I/O任务即selectionKey中的ready的事件,如accept,connect,read,write等,由processSelectedKeys方法触发
     * 非I/O任务添加到taskQueue中的任务,如register0,bind0等任务,由runAllTasks方法触发
     * 两种任务的执行时间比由变量ioRatio控制,默认为50,则表示允许非IO任务执行的事件与IO任务的执行时间相等
     */
    private final EventLoopGroup boosGroup = new NioEventLoopGroup();

    private final EventLoopGroup workGroup = new NioEventLoopGroup();

    /**
     * Channel
     * Channel类似Socket,它代表一个实体(如一个硬件设备,一个网络套接字) 的开放连接,如读写操作.通俗的讲,Channel字面意思就是通道,每一个客户端与服务端之间进行通信的一个双向通道.
     * Channel主要工作:
     * 1.当前网络连接的通道的状态(例如是否打开?是否已连接?)
     * 2.网络连接的配置参数(例如接收缓冲区的大小)
     * 3.提供异步的网络I/O操作(如建立连接,读写,绑定端口),异步调用意味着任何I/O调用都将立即返回,并且不保证在调用结束时锁清秋的I/O操作已完成.
     * 调用立即放回一个ChannelFuture实例,通过注册监听器到ChannelFuture上,可以I/O操作成功,失败或取消时回调通知调用方.
     * 4.支持关联I/O操作与对应的处理程序.
     * 不同协议,不同的阻塞类型的连接都有不同的Channel类型与之对应,下面是一些常用的Channel类型
     * NiosocketChannel,异步的客户端 TCP Socket连接
     * NioServerSocketChannel,异步的服务端 TCP Socket 连接
     * NioDatagramChannel,异步的UDP连接
     * NioSctpChannel,异步的客户端Sctp连接
     * NioSctoServerChannel,异步的Sctp服务端连接
     * 这些通道涵盖了UDP 和TCP网络IO以及文件IO
     */
    private Channel channel;


    /**
     * 启动服务
     *
     * @param port
     */
    public void start(int port) {
        log.info("=================Netty 端口启动:{}==================",port);

        /**
         * Future
         * Future提供了另外一种在操作完成时通知应用程序的方式,这个对象可以看做一个异步操作的结果占位符.
         * 通俗的讲,它相当于一位指挥官,发送了一个请求建立完连接,通信完毕了,你通知一声它回来关闭各种IO通道,整个过程,它是不阻塞的,异步的.
         * 在Netty中所有的IO操作都是异步的,不能理科的值消息是否被正确处理,但是可以过一会儿等他执行完成或者直接注册一个监听,具体的实现就是通过Future和ChannelFutures.
         * 他们可以注册一个监听,当操作执行成功成功或者失败时监听会自动触发注册的监听事件
         */
        try {
            /**
             * Bootstrap
             * Bootstrap是引导的意思,一个Netty应用通常由一个Bootstrap开始
             * 主要作用是配置整个Netty程序,串联各个组件
             * Netty中Bootstrap类是服务端启动引导类
             */
            ServerBootstrap server = new ServerBootstrap();
            server.group(boosGroup, workGroup)
                    //非阻塞异步服务端TCP Socket 连接
                    .channel(NioServerSocketChannel.class)
                    //设置为前端WebSocket可以连接
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // HttpServerCodec: 将请求和映带消息节吗为HTTP消息
                            pipeline.addLast("http-codec", new HttpServerCodec());
                            // 讲HTTP消息的多个部分合成一条完整的HTTP消息
                            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
                            // 向客户端发送html5文件
                            socketChannel.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
                            // 进行设置心跳检测
                            socketChannel.pipeline().addLast(new IdleStateHandler(60, 30, 60 * 30, TimeUnit.SECONDS));
                            // 配置通道处理 来进行业务处理
                            pipeline.addLast("handler", new WebSocketServerHandler());
                        }

                    });
            channel = server.bind(port).sync().channel();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @PreDestroy
    public void destroy() {
        log.info("=================Netty服务关闭==================");
        if (channel != null) {
            channel.close();
        }
        boosGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }


}

三丶Socket处理注册连接类

接收客户端连接信息,心跳检测,存储于销毁等的处理

其余部分业务类(messageService),均为自行处理连接存储为业务所用,请忽略

/**
 * @description
 * @author: 徐子木
 * @create: 2020-06-02 14:57
 **/
@Slf4j
@Component
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    public static final byte PING_MSG = 1;
    public static final byte PONG_MSG = 2;
    public static final byte CUSTOM_MSG = 3;
    private int heartbeatCount = 0;

  // 配置客户端是否为https的控制
    @Value("${netty.ssl-enabled:false}")
    private Boolean useSsl;

    /**
     * 这里可以引入自己业务类来处理进行的客户端连接
     */
    @Autowired
    private MessageService messageService;
    
    public static WebSocketServerHandler webSocketServerHandler;

    /**
     * 解决启动加载不到自己业务类
     */
    @PostConstruct
    public void init() {
        webSocketServerHandler = this;
    }


    private WebSocketServerHandshaker handshaker;


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        //http请求和tcp请求分开处理
        if (msg instanceof HttpRequest) {
            handlerHttpRequest(ctx, (HttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            //踩坑: simpleChannelInboundHandler 他会进行一次释放(引用计数器减一),参考源码,而我们释放的时候就变为了0,所以必须手动进行引用计数器加1
            WebSocketFrame frame = (WebSocketFrame) msg;
            frame.retain();
            handlerWebSocketFrame(ctx, frame);
        }

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * WebSocket 消息处理
     *
     * @param ctx
     * @param frame
     */
    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        //判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            log.info("【" + ctx.channel().remoteAddress() + "】已关闭(服务器端)");
            //移除channel
            NioSocketChannel channel = (NioSocketChannel) ctx.channel();
            webSocketServerHandler.messageService.removeConnection(channel);
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame);
            return;
        }
        //判断是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            log.info("【ping】");
            return;
        }
        //判断实时是pong消息
        if (frame instanceof PongWebSocketFrame) {
            log.info("【pong】");
            return;
        }
        //本例子只支持文本,不支持二进制
        if (!(frame instanceof TextWebSocketFrame)) {
            log.info("【不支持二进制】");
            throw new UnsupportedOperationException("不支持二进制");
        }

        // 传送的消息 ,接收客户端指定格式(自己与客户端约定json格式)的消息,并进行处理
        MessageObject messageObject = JSONObject.parseObject(((TextWebSocketFrame) frame).text().toString(), MessageObject.class);
        webSocketServerHandler.messageService.sendMessage(messageObject, ctx);
    }

    /**
     * websocket第一次连接握手
     *
     * @param ctx
     */
    @SuppressWarnings("deprecation")
    private void handlerHttpRequest(ChannelHandlerContext ctx, HttpRequest req) {
// 这里接收客户端附加连接参数,根据自己业务与客户端指定需要哪些参数来辨别连接唯一性
        String userUid = null;
        String sectionId = null;
        if ("GET".equalsIgnoreCase(req.getMethod().toString())) {
            String uri = req.getUri();
            userUid = req.getUri().substring(uri.indexOf("/", 2) + 1, uri.lastIndexOf("/"));
            sectionId = req.getUri().substring(uri.lastIndexOf("/") + 1);
            //对用户信息进行存储
            NioSocketChannel channel = (NioSocketChannel) ctx.channel();

            webSocketServerHandler.messageService.putConnection(userUid, sectionId, channel);
        }

        // http 解码失败
        if (!req.getDecoderResult().isSuccess() || (!"websocket".equalsIgnoreCase(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, (FullHttpRequest) req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
        }
        //可以通过url获取其他参数
        WebSocketServerHandshakerFactory factory;
// 这里主要用于 客户端为wss连接的处理
        if (useSsl != null && useSsl) {
            factory = new WebSocketServerHandshakerFactory(
                    "wss://" + req.headers().get("Host") + "/" + req.getUri() + "", null, false
            );
        } else {
            factory = new WebSocketServerHandshakerFactory(
                    "ws://" + req.headers().get("Host") + "/" + req.getUri() + "", null, false
            );
        }
        handshaker = factory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
        } else {
            //进行连接
            handshaker.handshake(ctx.channel(), (FullHttpRequest) req);
        }
    }

    @SuppressWarnings("deprecation")
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
        // 返回应答给客户端
        if (res.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);

            // buf.release();
        }
        // 如果是非Keep-Alive,关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }


    /**
     * 这里是保持服务器与客户端长连接  进行心跳检测 避免连接断开
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent stateEvent = (IdleStateEvent) evt;
            PingWebSocketFrame ping = new PingWebSocketFrame();
            switch (stateEvent.state()) {
                //读空闲(服务器端)
                case READER_IDLE:
                    //log.info("【" + ctx.channel().remoteAddress() + "】读空闲(服务器端)");
                    ctx.writeAndFlush(ping);
                    break;
                //写空闲(客户端)
                case WRITER_IDLE:
                    //log.info("【" + ctx.channel().remoteAddress() + "】写空闲(客户端)");
                    ctx.writeAndFlush(ping);
                    break;
                case ALL_IDLE:
                    //log.info("【" + ctx.channel().remoteAddress() + "】读写空闲");
                    break;
                default:
                    break;
            }
        }
    }


    /**
     * 出现异常时
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        //移除channel
        webSocketServerHandler.messageService.removeConnection((NioSocketChannel) ctx.channel());
        ctx.close();
        log.info("【" + ctx.channel().remoteAddress() + "】已关闭(服务器端)");
    }


}

四丶与业务整合并存储客户连接

以下均为自己业务与socket连接对象的存储,发送,踢出等处理,可根据自己业务自行参考

当时公司业务需要将连接存储两级维度,可根据自己的业务给客户端分组,或不分组都可以

/**
 * @description
 * @author: 徐子木
 * @create: 2020-09-30 15:06
 **/
@Slf4j
@Service(value = "messageService")
public class MessageServiceImpl implements MessageService {

// 是消息通信dubbo类,请忽略
    @Autowired
    private ChatMsgService chatMsgService;
// 是db存储类 ,请忽略
    @Autowired
    private BidPresentDao bidPresentDao;

// 这里是netty可为指定的唯一key去与连接进行分组处理并存储
    private HashedWheelTimer hashedWheelTimer = new HashedWheelTimer();

    private final AttributeKey<String> userKey = AttributeKey.valueOf("user");

    private final AttributeKey<String> sectionKey = AttributeKey.valueOf("section");

    /**
     * 装载标段与对应在线的用户
     */
    private static final Map<String, ChannelGroup> SECTION_GROUPS = new ConcurrentHashMap<>();


    /**
     * 维护某标段中的socket连接
     *
     * @param sectionId
     * @param channel
     */
    @Override
    public void putConnection(String userId, String sectionId, NioSocketChannel channel) {

        channel.attr(userKey).set(userId);
        channel.attr(sectionKey).set(sectionId);

        bidPresentDao.comeOnlineByUserId(userId, sectionId);

        //存储用户标段对应连接
        ChannelGroup channelGroup = SECTION_GROUPS.get(sectionId);
        if (null == channelGroup) {
            //保存全局的,连接上的服务器的客户
            channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
            channelGroup.add(channel);
            SECTION_GROUPS.put(sectionId, channelGroup);
        } else {
            channelGroup.add(channel);
        }
    }

    /**
     * 判断一个通道是否有用户在使用
     *
     * @param channel
     * @return
     */
    private boolean hasUser(Channel channel) {
        return ((channel.hasAttr(userKey) || channel.hasAttr(sectionKey)));
    }

    /**
     * 获取连接对应用户
     *
     * @param channel
     * @return
     */
    @Override
    public String getBindUserId(NioSocketChannel channel) {
        if (hasUser(channel)) {
            return channel.attr(userKey).get();
        }
        return null;
    }

    /**
     * 获取连接对应标段Id
     *
     * @param channel
     * @return
     */
    @Override
    public String getBindSectionId(NioSocketChannel channel) {
        if (hasUser(channel)) {
            return channel.attr(sectionKey).get();
        }
        return null;
    }


    /**
     * 用户退出标段在线连接
     *
     * @param channel
     */
    @Override
    public void removeConnection(NioSocketChannel channel) {
        String userId = netty玩转irving聊天室(android整合netty客户端+springboot整合netty服务端),附源码

springboot和netty如何整合呢 有没有实例

# SpringBoot 集成 Netty

# SpringBoot 集成 Netty

实用水文篇--SpringBoot整合Netty实现消息推送服务器

三分钟构建高性能 WebSocket 服务:超优雅的 Springboot 整合 Netty 方案