分布式物联网（IOT）服务实战（二）－消息通讯处理服务集成Netty

曾经终败给现在 2023-01-13 01:44 140阅读 0赞

# Netty简介 #

官网的简介：

Netty是一个NIO客户端服务器框架，可以快速轻松地开发网络应用程序，例如协议服务器和客户端。它极大地简化和简化了诸如TCP和UDP套接字服务器之类的网络编程。  
“快速简便”并不意味着最终的应用程序将遭受可维护性或性能问题的困扰。 Netty经过精心设计，结合了许多协议（例如FTP，SMTP，HTTP以及各种基于二进制和基于文本的旧协议）的实施经验。结果，Netty成功地找到了一种无需妥协即可轻松实现开发，性能，稳定性和灵活性的方法。

百度的介绍：

Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。  
也就是说，Netty 是一个基于NIO的客户、服务器端的编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。  
“快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议（包括FTP、SMTP、HTTP等各种二进制文本协议）的实现经验，并经过相当精心设计的项目。最终，Netty 成功的找到了一种方式，在保证易于开发的同时还保证了其应用的性能，稳定性和伸缩性

# 为什么用Netty？ #

硬件的智能化肯定是以后的发展方向，智能化最本质的当然是智能，这就需要用到人工智能技术了，但如果只有人工智能，硬件能否成为智能设备呢？答案肯定是否定的！只要硬件还需要与其它设备连接或者需要与服务器通讯，就离不开物联网，万物互联是智能社会的发展趋势。

而实现物联网通讯，我采用集成Netty的方式，为什么选择Netty呢？

Netty 是业界最流行的 NIO 框架之一，它的**健壮性、功能、性能、可定制性和可扩展性**在同类框架中都是首屈一指的，它已经得到成百上千的商用项目验证，例如 Hadoop 的 RPC 框架 Avro 使用 Netty 作为通信框架。很多其它业界主流的 RPC 和分布式服务框架，也使用Netty 来构建高性能的异步通信能力。

此外，**Netty还有以下优点：**

*  API 使用简单，开发门槛低；
 *  功能强大，预置了多种编解码功能，支持多种主流协议；
 *  定制能力强，可以通过 ChannelHandler 对通信框架进行灵活的扩展；
 *  性能高，通过与其它业界主流的 NIO 框架对比，Netty 的综合性能最优；
 *  社区活跃，版本迭代周期短，发现的 BUG 可以被及时修复，同时，更多的新功能会被加入；
 *  经历了大规模的商业应用考验，质量得到验证。在互联网、大数据、网络游戏、企业应用、电信软件等众多行业得到成功商用，证明了它完全满足不同行业的商用标准。

# 集成Netty #

集成Netty需要引入netty包，代码如下：

<dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
    </dependency>

Netty服务核心启动类为NettyServer，代码如下：

package com.iot.cloud.process.service;
    
    import com.iot.cloud.process.ProcessNettyProperties;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    import java.net.InetSocketAddress;
    
    /** * @author Hill Fu * netty server */
    @Component
    public class NettyServer { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(NettyServer.class);
    
        @Autowired
        private NettyServerHandlerInitializer nettyServerHandlerInitializer;
    
        @Autowired
        private ProcessNettyProperties properties;
    
        /** * boss thread group, Used to accept client connections */
        private EventLoopGroup bossGroup = new NioEventLoopGroup();
        /** * worker thread group, Used to accept data read and write from the client */
        private EventLoopGroup workerGroup = new NioEventLoopGroup();
    
        /** * Netty Server Channel */
        private Channel channel;
    
        /** * start * @throws InterruptedException */
        @PostConstruct
        public void start() throws InterruptedException { 
            //creat ServerBootstrap
            ServerBootstrap bootstrap = new ServerBootstrap();
            //set ServerBootstrap's various attributes
            bootstrap.group(bossGroup, workerGroup) //set two EventLoopGroup
                    .channel(NioServerSocketChannel.class)  //set Channel is NioServerSocketChannel
                    .localAddress(new InetSocketAddress(properties.getPort())) //set Netty Server port
                    .option(ChannelOption.SO_BACKLOG, 1024) // accept queue size
                    .childOption(ChannelOption.SO_KEEPALIVE, true) //TCP Keepalive mechanism to realize the TCP-level heartbeat keepalive function
                    .childOption(ChannelOption.TCP_NODELAY, true) //Allow smaller data packets to be sent, reducing delay
                    .childHandler(nettyServerHandlerInitializer);
            try { 
                // bind
                ChannelFuture future = bootstrap.bind().sync();
                if (future.isSuccess()) { 
                    channel = future.channel();
                    LOGGER.info("Netty Server start, port is {} ]", properties.getPort());
                }
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        /** * stop */
        @PreDestroy
        public void stop() { 
            if (channel != null) { 
                channel.close();
            }
            if (bossGroup != null) { 
                bossGroup.shutdownGracefully();
            }
            if (workerGroup != null) { 
                workerGroup.shutdownGracefully();
            }
        }
    
    }

处理程序初始化类NettyServerHandlerInitializer，代码如下：

package com.iot.cloud.process.service;
    
    import com.iot.cloud.process.ProcessNettyProperties;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.SocketChannel;
    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.timeout.IdleStateHandler;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.TimeUnit;
    
    /** * @author Hill Fu * init handler */
    @Component
    public class NettyServerHandlerInitializer extends ChannelInitializer<SocketChannel> { 
    
        @Autowired
        private ProcessNettyProperties properties;
    
        @Autowired
        private NettyServerHandler nettyServerHandler;
    
        @Override
        protected void initChannel(SocketChannel ch) { 
            // 编解码http请求
            ch.pipeline().addLast(new HttpServerCodec())
                    //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
                    //保证接收的Http请求的完整性
                    .addLast(new HttpObjectAggregator(64 * 1024))
                    // 处理其他的WebSocketFrame
                    .addLast(new WebSocketServerProtocolHandler(properties.getPath()))
                    //heartbeat handler
                    .addLast(new IdleStateHandler(properties.getTimeout(),0L,0L, TimeUnit.SECONDS))
                    // 处理TextWebSocketFrame
                    .addLast(nettyServerHandler);
        }
    
    }

编码解码程序，如果要实现具体的物联网功能，自定义，此时暂时不做变动。

关于Netty的一些配置，在ProcessNettyProperties中，代码如下：

package com.iot.cloud.process;
    
    import lombok.Data;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    
    /** * @author Hill Fu * properties */
    @ConfigurationProperties(prefix = "process.netty")
    @Data
    public class ProcessNettyProperties { 
    
        /** * netty socket port */
        private int port = 90000;
    
        /** * netty socket path */
        private String path = "/ws";
    
        /** * heartbeat timeout value */
        private long timeout = 180L;
    
    }

其中，process.netty是yml中的配置，代码如下：

process:
      netty:
        port: 41000
        path: /ws
        timeout: 180

另外，需要在启动类ProcessApplication加上@EnableConfigurationProperties(ProcessNettyProperties.class)注解，代码如下：

package com.iot.cloud.process;
    
    import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.boot.context.properties.EnableConfigurationProperties;
    import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
    import org.springframework.cloud.openfeign.EnableFeignClients;
    
    @SpringBootApplication
    @EnableEurekaClient
    @EnableFeignClients
    @EnableConfigurationProperties(ProcessNettyProperties.class)
    public class ProcessApplication { 
    
        public static void main(String[] args) { 
            SpringApplication.run(ProcessApplication.class, args);
        }
    
    }

最后，是自定义的处理类NettyServerHandler，代码如下：

package com.iot.cloud.process.service;
    
    import com.alibaba.fastjson.JSONObject;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    
    /** * @author Hill Fu * netty sever handler */
    @Component
    @ChannelHandler.Sharable
    public class NettyServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { 
    
        private static final Logger LOGGER = LoggerFactory.getLogger(NettyServer.class);
    
        /** * connection active * @param ctx * @throws Exception */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            LOGGER.info("connection active : {}", ctx.channel().id().asLongText());
            super.channelActive(ctx);
        }
    
        /** * connection inactive * @param ctx * @throws Exception */
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
            LOGGER.info("connection inactive : {}", ctx.channel().id().asLongText());
            super.channelInactive(ctx);
        }
    
        /** * connection exception * @param ctx * @param cause * @throws Exception */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            LOGGER.info("connection exception : {}", ctx.channel().id().asLongText());
            super.exceptionCaught(ctx, cause);
        }
    
        /** * message handle */
        @Override
        public void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { 
            LOGGER.info("received message : {} {}", ctx.channel().id().asLongText(), msg.text());
            RequestMsg requestMsg = JSONObject.parseObject(msg.text(), RequestMsg.class);
            ctx.channel().writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(requestMsg)));
        }
    
    }

# 通讯消息测试 #

可以用静态html完成socket连接，发送消息进行测试，代码如下：

<!DOCTYPE html>
    <html>
    	<head>
    		<meta charset="utf-8" />
    		<title>Netty iot案例</title>
    	</head>
    	<body>
    		<div id="">发送消息：</div><br>
    		<input type="text" name="messageContent" id="messageContent"/>
    		<input type="button" name="" id="" value="发送" onclick="CHAT.chat()"/>
    		
    		<hr>
    		
    		<div id="">接收消息：</div><br>
    		<div id="receiveNsg" style="background-color: gainsboro;"></div>
    		
    		
    		<script type="text/javascript"> window.CHAT = { socket: null, //初始化 init: function(){ //首先判断浏览器是否支持WebSocket if (window.WebSocket){ CHAT.socket = new WebSocket("ws://localhost:41000/ws"); CHAT.socket.onopen = function(){ console.log("客户端与服务端建立连接成功"); }, CHAT.socket.onmessage = function(e){ console.log("接收到消息："+e.data); var receiveNsg = window.document.getElementById("receiveNsg"); var html = receiveNsg.innerHTML; var source = JSON.parse(e.data); receiveNsg.innerHTML = html + "<br>" + source.msg; }, CHAT.socket.onerror = function(){ console.log("发生错误"); }, CHAT.socket.onclose = function(){ console.log("客户端与服务端关闭连接成功"); } }else{ alert("8102年都过了，升级下浏览器吧"); } }, chat: function(){ var mc = window.document.getElementById("messageContent"); // var param = {msg:mc.value}; var data = { "msg" : mc.value }; CHAT.socket.send(JSON.stringify(data)); } } CHAT.init(); </script>
    		
    	</body>
    </html>

打开后界面如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNzgxODE_size_16_color_FFFFFF_t_70_pic_center]

文件打开后，连接建立后，后台日志，如图：

![在这里插入图片描述][2021041511281087.png_pic_center]

发送消息，展示如图：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNzgxODE_size_16_color_FFFFFF_t_70_pic_center 1]

后端日志，如图：

![在这里插入图片描述][20210415112836219.png_pic_center]

# 总结 #

本节只完成了Netty的集成，并且对消息通讯进行了简单的测试，后续我们会处理关于编码、解码、消息处理、权限校验等问题的处理。

敬请期待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNzgxODE_size_16_color_FFFFFF_t_70_pic_center]: /images/20221022/68dec7d0f3e6481cb8f727807ecb89cd.png
[2021041511281087.png_pic_center]: /images/20221022/8e9cb937db474e74b5993f0ffdc07e84.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQzNzgxODE_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221022/21faf986d45d4646a5d3c8e8d1f281d1.png
[20210415112836219.png_pic_center]: /images/20221022/708596850d074956a83925c7f5307425.png