赞
踩
Netty 是基于 Java NIO 的异步事件驱动的网络应用框架,使用 Netty 可以快速开发网络应用,Netty 提供了高层次的抽象来简化 TCP 和 UDP 服务器的编程,但是你仍然可以使用底层的 API。
Netty 的内部实现是很复杂的,但是 Netty 提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty 是完全基于 NIO 实现的,所以整个 Netty 都是异步的。
Netty 是最流行的 NIO 框架,它已经得到成百上千的商业、商用项目验证,许多框架和开源组件的底层 rpc 都是使用的 Netty,如Dubbo、Elasticsearch 等等。
优点:
- API使用简单,学习成本低。
- 高度可定制的线程模型——单线程、一个或多个线程池。
- 功能强大,内置了多种解码编码器,支持多种协议。
- 社区活跃,发现BUG会及时修复,迭代版本周期短,不断加入新的功能。
- Dubbo、Elasticsearch都采用了Netty,质量得到验证。
- 更好的吞吐量,更低的等待延迟,更少的资源消耗
各部分介绍:
BossGroup 和 WorkerGroup:
- bossGroup 和 workerGroup 是两个线程池, 它们默认线程数为 CPU 核心数乘以 2
- bossGroup 用于接收客户端传过来的请求,接收到请求后将后续操作交由 workerGroup 处理
Selector(选择器):
- 检测多个通道上是否有事件的发生
TaskQueue(任务队列):
- 上面的任务都是在当前的 NioEventLoop ( 反应器 Reactor 线程 ) 中的任务队列中排队执行 , 在其它线程中也可以调度本线程的 Channel 通道与该线程对应的客户端进行数据读写
Channel:
- Channel 是框架自己定义的一个通道接口,
- Netty 实现的客户端 NIO 套接字通道是 NioSocketChannel
- 提供的服务器端 NIO 套接字通道是 NioServerSocketChannel
- 当服务端和客户端建立一个新的连接时, 一个新的 Channel 将被创建,同时它会被自动地分配到它专属的 ChannelPipeline
ChannelPipeline:
- 是一个拦截流经 Channel 的入站和出站事件的 ChannelHandler 实例链,并定义了用于在该链上传播入站和出站事件流的 API
ChannelHandler:
- 分为 ChannelInBoundHandler 和 ChannelOutboundHandler 两种
- 如果一个入站 IO 事件被触发,这个事件会从第一个开始依次通过 ChannelPipeline中的 ChannelInBoundHandler,先添加的先执行。
- 若是一个出站 I/O 事件,则会从最后一个开始依次通过 ChannelPipeline 中的 ChannelOutboundHandler,后添加的先执行,然后通过调用在 ChannelHandlerContext 中定义的事件传播方法传递给最近的 ChannelHandler。
- 在 ChannelPipeline 传播事件时,它会测试 ChannelPipeline 中的下一个 ChannelHandler 的类型是否和事件的运动方向相匹配。
- 如果某个ChannelHandler不能处理则会跳过,并将事件传递到下一个ChannelHandler,直到它找到和该事件所期望的方向相匹配的为止。
1、pom文件
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-devtools</artifactId> <scope>runtime</scope> <optional>true</optional> </dependency> <!--整合模板引擎 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-thymeleaf</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>io.netty</groupId> <artifactId>netty-all</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> </dependency>
2、yml配置
server: port: 8080 netty: port: 8081 path: /chat resources: static-locations: - classpath:/static/ spring: thymeleaf: cache: false checktemplatelocation: true enabled: true encoding: UTF-8 mode: HTML5 prefix: classpath:/templates/ suffix: .html
3、netty实体配置类
import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.stereotype.Component; @Component @ConfigurationProperties(prefix = "netty") @Data public class Netty { /** * netty监听的端口 */ private int port; /** * websocket访问路径 */ private String path; }
4、启动类
5、服务器端
import com.netty.socket.config.WebSocketChannelConfig; import com.netty.socket.entity.Netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import javax.annotation.PreDestroy; /** * Netty服务器 * @author 流星 */ @Component @Slf4j public class NettyWebSocketServer implements Runnable { @Autowired private Netty netty; @Autowired private WebSocketChannelConfig webSocketChannelConfig; /** * boss线程组,用于处理连接 */ private EventLoopGroup bossGroup = new NioEventLoopGroup(1); /** * work线程组,用于处理消息 */ private EventLoopGroup workerGroup = new NioEventLoopGroup(); /** * 资源关闭——在容器销毁时关闭 */ @PreDestroy public void close() { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } @Override public void run() { try { //创建服务端启动助手 ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup); serverBootstrap.channel(NioServerSocketChannel.class) .handler(new LoggingHandler(LogLevel.DEBUG)) .childHandler(webSocketChannelConfig); //启动 ChannelFuture channelFuture = serverBootstrap.bind(netty.getPort()).sync(); log.info("【-----Netty服务端启动成功-----】"); channelFuture.channel().closeFuture().sync(); } catch (Exception e) { e.printStackTrace(); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }
bootstrap:
- boostrap 用来为 Netty 程序的启动组装配置一些必须要组件,例如上面的创建的两个线程组
channel方法:
- 用于指定服务器端监听套接字通道NioServerSocketChannel,其内部管理了一个 Java NIO 中的ServerSocketChannel实例
channelHandler 方法:
- 用于设置业务职责链,责任链是我们下面要编写的,责任链具体是什么,它其实就是由一个个的 ChannelHandler 串联而成,形成的链式结构
bind 方法:
- 将服务绑定到端口上,bind 方法内部会执行端口绑定等一系列操,使得前面的配置都各就各位各司其职
sync 方法:
- 用于阻塞当前 Thread,一直到端口绑定操作完成。
6、 Channel配置类
import com.netty.socket.entity.Netty; import com.netty.socket.handler.WebSocketHandler; import io.netty.channel.Channel; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; 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 org.springframework.stereotype.Component; import javax.annotation.Resource; /** * 通道初始化对象 * @author 流星 */ @Component public class WebSocketChannelConfig extends ChannelInitializer<Channel> { @Resource private Netty netty; @Resource private WebSocketHandler webSocketHandler; @Override protected void initChannel(Channel channel) throws Exception { ChannelPipeline pipeline = channel.pipeline(); // 对http协议的支持. pipeline.addLast(new HttpServerCodec()); // 对大数据流的支持 pipeline.addLast(new ChunkedWriteHandler()); // HttpObjectAggregator将多个信息转化成单一的request或者response对象 pipeline.addLast(new HttpObjectAggregator(8000)); // 将http协议升级为ws协议. 对websocket的支持 pipeline.addLast(new WebSocketServerProtocolHandler(netty.getPath())); // 自定义处理handler pipeline.addLast(webSocketHandler); } }
addLast 方法:
- 将一个一个的 ChannelHandler 添加到责任链上并给它们取个名称(不取也可以,Netty 会给它个默认名称),这样就形成了链式结构。在请求进来或者响应出去时都会经过链上这些 ChannelHandler 的处理。
7、自定义处理类
import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.util.ArrayList; import java.util.List; /** * 自定义处理类 * TextWebSocketFrame: websocket数据是帧的形式处理 * * @author 流星 */ @Component @Slf4j /** * 设置通道共享 */ @ChannelHandler.Sharable public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> { public static List<Channel> channelList = new ArrayList<>(); /** * 通道就绪事件 * * @param ctx * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //当有新的客户端连接的时候, 将通道放入集合 channelList.add(channel); log.info("有新的连接加入。。。"); } /** * 通道未就绪事件 * * @param ctx * @throws Exception */ @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //当有客户端断开连接的时候,就移除对应的通道 channelList.remove(channel); log.info("有连接已经断开。。。"); } /** * 读就绪事件 * * @param ctx * @param textWebSocketFrame * @throws Exception */ @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame) throws Exception { String msg = textWebSocketFrame.text(); log.info("msg:{}", msg); //当前发送消息的通道, 当前发送的客户端连接 Channel channel = ctx.channel(); for (Channel channel1 : channelList) { //排除自身通道 if (channel != channel1) { channel1.writeAndFlush(new TextWebSocketFrame(msg)); } } } /** * 异常处理事件 * * @param ctx * @param cause * @throws Exception */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); Channel channel = ctx.channel(); //移除集合 channelList.remove(channel); } }
- 对于自定义的 ChannelHandler, 一般会继承 Netty 提供的SimpleChannelInboundHandler类,并且对于 Http 请求我们可以给它设置泛型参数为 HttpOjbect 类,然后覆写 channelRead0 方法。
- channelRead0 方法中编写我们的业务逻辑代码,此方法会在接收到服务器数据后被系统调用
- channelActive 方法当连接建立时,此方法会被调用,我们在方法中构建了一个 Channel对象,并且通过 writeAndFlush 方法将请求发送出去
- channelRead0 方法用于处理服务端返回给我们的响应,打印服务端返回给客户端的信息
8、html
<!DOCTYPE html> <html lang="en" xmlns:th="http://www.thymeleaf.org"> <head> <meta charset="UTF-8"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>聊天室</title> <link rel="stylesheet" href="/css/chat.css"> <script src="https://code.jquery.com/jquery-3.5.1.min.js"></script> <!-- <script src="/js/chat.js"></script>--> </head> <body> <div id="chat"> <div class="sidebar"> <div class="m-card"> <header> <span class="name" >姓名:</span> <span class="name" id="username"></span> </header> </div> <div class="m-list"> <ul id="user_list"> </ul> </div> </div> <div class="main"> <div class="m-message"> <ul id="msg_list"> </ul> </div> <div class="m-text"> <textarea placeholder="按 Enter 发送" id="my_test"></textarea> <!-- <div class="pager_btn">--> <!-- <button id="send">发送</button>--> <!-- </div>--> <!-- <div class="arrow_box">--> <!-- 发送内容不能为空--> <!-- <div class="arrow"></div>--> <!-- </div>--> </div> </div> </div> <script> $(function () { //这里需要注意的是,prompt有两个参数,前面是提示的话,后面是当对话框出来后,在对话框里的默认值 var username = ""; while (true) { //弹出一个输入框,输入一段文字,可以提交 username = prompt("请输入您的名字", ""); //将输入的内容赋给变量 name , if (username.trim() === "")//如果返回的有内容 { alert("名称不能输入空") } else { $("#username").text(username); break; } } var ws = new WebSocket("ws://localhost:8081/chat"); ws.onopen = function () { console.log("连接成功.") } ws.onmessage = function (evt) { showMessage(evt.data); } ws.onclose = function (){ console.log("连接关闭") } ws.onerror = function (){ console.log("连接异常") } function showMessage(message) { // 张三:你好 var str = message.split(":"); $("#msg_list").append(`<li class="active"}> <div class="main-others"> <img class="avatar" width="35" height="35" src="/img/user.jpg"> <div class="main-text"> <div class="user_name">${str[0]}</div> <div class="text">${str[1]}</div> </div> </div> </li>`); // 置底 setBottom(); } $('#my_test').bind({ focus: function (event) { event.stopPropagation() $('#my_test').val(''); $('.arrow_box').hide() }, keydown: function (event) { event.stopPropagation() if (event.keyCode === 13) { if ($('#my_test').val().trim() === '') { this.blur() $('.arrow_box').show() setTimeout(() => { this.focus() }, 1000) } else { $('.arrow_box').hide() //发送消息 sendMsg(); this.blur() setTimeout(() => { this.focus() }) } } } }); $('#send').on('click', function (event) { event.stopPropagation() if ($('#my_test').val().trim() === '') { $('.arrow_box').show() } else { sendMsg(); } }) function sendMsg() { var message = $("#my_test").val(); $("#msg_list").append(`<li class="active"}> <div class="main-self"> <span class="main-text">` + message + `</span> </div> </li>`); $("#my_test").val(''); //发送消息 message = username + ":" + message; ws.send(message); // 置底 setBottom(); } // 置底 function setBottom() { // 发送消息后滚动到底部 const container = $('.m-message') const scroll = $('#msg_list') container.animate({ scrollTop: scroll[0].scrollHeight - container[0].clientHeight + container.scrollTop() + 100 }); } var textarea = document.querySelector('textarea'); textarea.addEventListener('input', (e) => { textarea.style.height = '100px'; textarea.style.height = e.target.scrollHeight + 'px'; }); }); </script> </body> </html>
模拟三个用户:
聊天测试:
后台输出:
1、 超详细Netty入门,看这篇就够了!
2、SpringBoot+Netty+WebSocket实现在线聊天
3、Netty实战入门详解——让你彻底记住什么是Netty(看不懂你来找我)
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。