写点什么

小六六学 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
用户头像

自然

关注

还未添加个人签名 2020.03.01 加入

小六六,目前负责营收超百亿的支付中台

评论

发布
暂无评论
小六六学Netty系列之Netty群聊_Netty_自然_InfoQ写作社区