小六六学 Netty 系列之 Netty 群聊
作者：自然
- 2022 年 9 月 05 日 广东
本文字数：9991 字
阅读完需：约 33 分钟
前言
文本已收录至我的 GitHub 仓库，欢迎 Star：https://github.com/bin392328206/six-finger
种一棵树最好的时间是十年前，其次是现在
我知道很多人不玩 qq 了,但是怀旧一下,欢迎加入六脉神剑 Java 菜鸟学习群，群聊号码：549684836 鼓励大家在技术的路上写博客
絮叨
为了学习 Netty，我们前面铺垫了那么多，NIO Java 的零拷贝，UNIX 的 I/O 模型等等。下面是前面系列的链接
昨天我们看了一个 Netty 的基本介绍，那么今天我们继续学习 Netty
其实到目前为止，小六六都是看山是山，其实这一遍的学习，小六六只是让自己先熟悉熟悉，反正从小学习，我感觉很多东西都得反复学习，但是有些同学他们学一遍就会了，而我得 2 遍 3 遍，哎 只能说比不得。不过，坚持吧，相信勤能补拙。
Netty 网络编程应用实例-群聊系统
实例要求: 编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）实现多人群聊服务器端：可以监测用户上线，离线，并实现消息转发功能客户端：通过channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息(有服务器转发得到)目的：进一步理解Netty非阻塞网络编程机制
复制代码
GroupChatServer
package com.liuliu.nio.netty.groupchat;
import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;
/** * @author 小六六 * @version 1.0 * @date 2020/8/19 11:54 */public class GroupChatServer {
private int port; //监听的端口
public GroupChatServer( int port){ this.port=port; }
public void run() throws Exception{ NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
ServerBootstrap serverBootstrap = new ServerBootstrap();
try { serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG,128) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //向pipeline加入解码器 pipeline.addLast("decoder", new StringDecoder()); //向pipeline加入编码器 pipeline.addLast("encoder", new StringEncoder());
//加入自己的业务处理handler pipeline.addLast(new GroupChatServerHandler()); } }); System.out.println("netty 服务器启动"); ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
//监听关闭 channelFuture.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }
}
public static void main(String[] args) throws Exception {
new GroupChatServer(7000).run(); }}
复制代码
GroupChatServerHandler
package com.liuliu.nio.netty.groupchat;
import io.netty.channel.Channel;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.channel.group.ChannelGroup;import io.netty.channel.group.DefaultChannelGroup;import io.netty.util.concurrent.GlobalEventExecutor;
import java.text.SimpleDateFormat;
/** * @author 小六六 * @version 1.0 * @date 2020/8/19 16:15 */public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//handlerAdded 表示连接建立，一旦连接，第一个被执行 //将当前channel 加入到 channelGroup @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); //将该客户加入聊天的信息推送给其它在线的客户端 /* 该方法会将 channelGroup 中所有的channel 遍历，并发送 消息， 我们不需要自己遍历 */ channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天" + sdf.format(new java.util.Date()) + " \n"); channelGroup.add(channel); }
//断开连接, 将xx客户离开信息推送给当前在线的客户 @Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { Channel channel = ctx.channel(); channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了\n"); System.out.println("channelGroup size" + channelGroup.size()); }
//表示channel 处于活动状态, 提示 xx上线 @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress()+"上线了"); }
//表示channel处于不活跃状态 @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println(ctx.channel().remoteAddress() + " 离线了~"); } //读取数据 @Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { //获取到当前channel Channel channel = ctx.channel(); //这时我们遍历channelGroup, 根据不同的情况，回送不同的消息 channelGroup.forEach(ch -> { if(channel != ch) { //不是当前的channel,转发消息 ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + "\n"); }else {//回显自己发送的消息给自己 ch.writeAndFlush("[自己]发送了消息" + msg + "\n"); } }); }
//关闭 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭通道 ctx.close(); }}
复制代码
GroupChatClient
package com.liuliu.nio.netty.groupchat;
import io.netty.bootstrap.Bootstrap;import io.netty.channel.Channel;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;
import java.util.Scanner;
/** * @author 小六六 * @version 1.0 * @date 2020/8/19 16:26 */public class GroupChatClient {
private final String host; private final int port;
public GroupChatClient(String host, int port) { this.host = host; this.port = port; }
public void run() throws Exception{ NioEventLoopGroup eventExecutors = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(eventExecutors).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入相关handler pipeline.addLast("decoder", new StringDecoder()); pipeline.addLast("encoder", new StringEncoder()); //加入自定义的handler pipeline.addLast(new GroupChatClientHandler()); } }); ChannelFuture channelFuture = bootstrap.connect(host, port).sync(); //得到channel Channel channel = channelFuture.channel(); System.out.println("-------" + channel.localAddress()+ "--------"); Scanner scanner = new Scanner(System.in); while (scanner.hasNextLine()) { String msg = scanner.nextLine(); //通过channel 发送到服务器端 channel.writeAndFlush(msg + "\r\n"); }
} finally { eventExecutors.shutdownGracefully(); }
} public static void main(String[] args) throws Exception { new GroupChatClient("127.0.0.1", 7000).run(); }}
复制代码
GroupChatClientHandler
package com.liuliu.nio.netty.groupchat;
import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;
/** * @author 小六六 * @version 1.0 * @date 2020/8/19 16:33 */public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.println(msg.trim()); }}
复制代码
结果
Netty 心跳检测机制案例
编写一个 Netty心跳检测机制案例, 当服务器超过3秒没有读时，就提示读空闲当服务器超过5秒没有写操作时，就提示写空闲实现当服务器超过7秒没有读或者写操作时，就提示读写空闲
复制代码
MyServer
package com.atguigu.netty.heartbeat;
import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;
public class MyServer { public static void main(String[] args) throws Exception{
//创建两个线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup); serverBootstrap.channel(NioServerSocketChannel.class); serverBootstrap.handler(new LoggingHandler(LogLevel.INFO)); serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入一个netty 提供 IdleStateHandler /* 说明 1. IdleStateHandler 是netty 提供的处理空闲状态的处理器 2. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接 3. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接 4. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接
5. 文档说明 triggers an {@link IdleStateEvent} when a {@link Channel} has not performed * read, write, or both operation for a while. * 6. 当 IdleStateEvent 触发后 , 就会传递给管道 的下一个handler去处理 * 通过调用(触发)下一个handler 的 userEventTiggered , 在该方法中去处理 IdleStateEvent(读空闲，写空闲，读写空闲) */ pipeline.addLast(new IdleStateHandler(7000,7000,10, TimeUnit.SECONDS)); //加入一个对空闲检测进一步处理的handler(自定义) pipeline.addLast(new MyServerHandler());
} });
//启动服务器 ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync();
}finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }}
复制代码
MyServerHandler
package com.atguigu.netty.heartbeat;
import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.handler.timeout.IdleStateEvent;
public class MyServerHandler extends ChannelInboundHandlerAdapter {
/** * * @param ctx 上下文 * @param evt 事件 * @throws Exception */ @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if(evt instanceof IdleStateEvent) {
//将 evt 向下转型 IdleStateEvent IdleStateEvent event = (IdleStateEvent) evt; String eventType = null; switch (event.state()) { case READER_IDLE: eventType = "读空闲"; break; case WRITER_IDLE: eventType = "写空闲"; break; case ALL_IDLE: eventType = "读写空闲"; break; } System.out.println(ctx.channel().remoteAddress() + "--超时时间--" + eventType); System.out.println("服务器做相应处理..");
//如果发生空闲，我们关闭通道 // ctx.channel().close(); } }}
复制代码
结果
Netty 通过 WebSocket 编程实现服务器和客户端长连接
实例要求: Http协议是无状态的, 浏览器和服务器间的请求响应一次，下一次会重新创建连接.
要求：实现基于webSocket的长连接的全双工的交互改变Http协议多次请求的约束，实现长连接了， 服务器可以发送消息给浏览器客户端浏览器和服务器端会相互感知，比如服务器关闭了，浏览器会感知，同样浏览器关闭了，服务器会感知
复制代码
MyServer
package com.liuliu.nio.netty.websocket;
import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;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.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;import io.netty.handler.stream.ChunkedWriteHandler;
import java.net.ServerSocket;
/** * @author 小六六 * @version 1.0 * @date 2020/8/20 10:57 */public class MyServer {
public static void main(String[] args) throws Exception {
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1); NioEventLoopGroup workerGroup = new NioEventLoopGroup(8); try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {
@Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //因为基于http协议，使用http的编码和解码器 pipeline.addLast(new HttpServerCodec()); //是以块方式写，添加ChunkedWriteHandler处理器 pipeline.addLast(new ChunkedWriteHandler()); /* 说明 1. http数据在传输过程中是分段, HttpObjectAggregator ，就是可以将多个段聚合 2. 这就就是为什么，当浏览器发送大量数据时，就会发出多次http请求 */
pipeline.addLast(new HttpObjectAggregator(8192)); /* 说明 1. 对应websocket ，它的数据是以 帧(frame) 形式传递 2. 可以看到WebSocketFrame 下面有六个子类 3. 浏览器请求时 ws://localhost:7000/xiaoliuliu 表示请求的uri 4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 保持长连接 5. 是通过一个 状态码 101 */ pipeline.addLast(new WebSocketServerProtocolHandler("/xiaoliuliu")); //自定义的handler ，处理业务逻辑 pipeline.addLast(new MyTextWebSocketFrameHandler()); } });
//启动服务器 ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync();
} finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }
}}
复制代码
MyTextWebSocketFrameHandler
package com.liuliu.nio.netty.websocket;
import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import java.time.LocalDateTime;
/** * @author 小六六 * @version 1.0 * @date 2020/8/20 11:07 */public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
@Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception { System.out.println("服务器收到消息"+msg.text()); //回复消息 ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " " + msg.text())); }
//客户端连接的时候触发 @Override public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //id 表示唯一的值，LongText 是唯一的 ShortText 不是唯一 System.out.println("handlerAdded 被调用" + ctx.channel().id().asLongText()); System.out.println("handlerAdded 被调用" + ctx.channel().id().asShortText()); }
@Override public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
System.out.println("handlerRemoved 被调用" + ctx.channel().id().asLongText()); }
@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { System.out.println("异常发生 " + cause.getMessage()); ctx.close(); //关闭连接 }}
复制代码
xiaoliuliu.html
<!DOCTYPE html><html lang="en"><head> <meta charset="UTF-8"> <title>Title</title></head><body>
<script>
var socket; if (window.WebSocket){ // go socket = new WebSocket("ws://localhost:7000/xiaoliuliu"); //相当于channelReado, ev 收到服务器端回送的消息 socket.onmessage = function (ev) { var rt = document.getElementById("responseText"); rt.value = rt.value + "\n" + ev.data; } //相当于连接开启(感知到连接开启) socket.onopen = function (ev) { var rt = document.getElementById("responseText"); rt.value = "连接开启了.." }
//相当于连接关闭(感知到连接关闭) socket.onclose = function (ev) { var rt = document.getElementById("responseText"); rt.value = rt.value + "\n" + "连接关闭了.." } } else { alert("当前浏览器不支持websocket") }
//发送消息到服务器 function send(message) { if(!window.socket) { //先判断socket是否创建好 return; } if(socket.readyState == WebSocket.OPEN) { //通过socket 发送消息 socket.send(message) } else { alert("连接没有开启"); } }</script><form onsubmit="return false"> <textarea name="message" style="height: 300px; width: 300px"></textarea> <input type="button" value="发生消息" onclick="send(this.form.message.value)"> <textarea id="responseText" style="height: 300px; width: 300px"></textarea> <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''"></form></body></html>
复制代码
结果
结尾
马上就结束了入门 Netty 系列了，其实这个系列也就是一个科普而已，并不能让你可以马上开发 Netty 程序，但是对于以后的使用肯定是有帮助的
日常求赞
好了各位，以上就是这篇文章的全部内容了，能看到这里的人呀，都是真粉。
创作不易，各位的支持和认可，就是我创作的最大动力，我们下篇文章见
六脉神剑 | 文 【原创】如果本篇博客有任何错误，请批评指教，不胜感激 ！
划线
评论
复制
发布于: 刚刚阅读数: 4
版权声明: 本文为 InfoQ 作者【自然】的原创文章。
原文链接:【http://xie.infoq.cn/article/0b734023417b2af4bf747e006】。文章转载请联系作者。
自然
关注
还未添加个人签名 2020.03.01 加入
小六六，目前负责营收超百亿的支付中台
评论