欢迎大家关注github.com/hsfxuebao，期望对大家有所协助，要是觉得能够的话费事给点一下Star哈
1. 群聊系统
实例要求: 编写一个 Netty 群聊系统，完成服务器端和客户端之间的数据简略通讯（非阻塞），完成多人群聊
服务器端：能够监测用户上线，离线，并完成音讯转发功能
客户端：经过channel 能够无阻塞发送音讯给其它一切用户，同时能够接受其它用户发送的音讯(有服务器转发得到)
意图：进一步理解Netty非阻塞网络编程机制
GroupChatServer端的代码：
public class GroupChatServer {
private int port; //监听端口
public GroupChatServer(int port) {
this.port = port;
}
//编写run办法，处理客户端的恳求
public void run() throws Exception{
//创立两个线程组
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 128)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//获取到pipeline
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 = b.bind(port).sync();
//监听封闭
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
new GroupChatServer(7000).run();
}
}
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
//public static List<Channel> channels = new ArrayList<Channel>();
//运用一个hashmap 管理
//public static Map<String, Channel> channels = new HashMap<String,Channel>();
//定义一个channle 组，管理一切的channel
//GlobalEventExecutor.INSTANCE) 是全局的事情执行器，是一个单例
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
public static final DateTimeFormatter DATE_KEY_FORMATTER = DateTimeFormatter.ofPattern("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() + " 参加谈天" + LocalDateTime.now().format(DATE_KEY_FORMATTER) + " \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 处于不活动状况, 提示 xx离线了
@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客户端代码：
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{
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap()
.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//得到pipeline
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 {
group.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
new GroupChatClient("127.0.0.1", 7000).run();
}
}
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(msg.trim());
}
}
GroupCharServer执行成果如下：
netty 服务器发动
/127.0.0.1:53907 上线了~
/127.0.0.1:53924 上线了~
/127.0.0.1:53924 离线了~
channelGroup size1
/127.0.0.1:53949 上线了~
/127.0.0.1:53957 上线了~
GroupCharClient 1 成果如下：
-------/127.0.0.1:53907--------
[客户端]/127.0.0.1:53924 参加谈天2022-05-07 17:05:41
[客户端]/127.0.0.1:53924 脱离了
[客户端]/127.0.0.1:53949 参加谈天2022-05-07 17:07:03
[客户端]/127.0.0.1:53957 参加谈天2022-05-07 17:07:12
[客户]/127.0.0.1:53949 发送了音讯我是 2
我是1
[自己]发送了音讯我是1
[客户]/127.0.0.1:53957 发送了音讯我是3
GroupCharClient 2 成果如下：
-------/127.0.0.1:53949--------
[客户端]/127.0.0.1:53957 参加谈天2022-05-07 17:07:12
我是 2
[自己]发送了音讯我是 2
[客户]/127.0.0.1:53907 发送了音讯我是1
[客户]/127.0.0.1:53957 发送了音讯我是3
GroupCharClient 3成果如下：
-------/127.0.0.1:53957--------
[客户]/127.0.0.1:53949 发送了音讯我是 2
[客户]/127.0.0.1:53907 发送了音讯我是1
我是3
[自己]发送了音讯我是3
2. 心跳检测机制案例
实例要求:
编写一个 Netty心跳检测机制案例, 当服务器超越3秒没有读时，就提示读闲暇。当服务器超越5秒没有写操作时，就提示写闲暇。完成当服务器超越7秒没有读或许写操作时，就提示读写闲暇
Server端代码：
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(10,13,20, TimeUnit.SECONDS));
//参加一个对闲暇检测进一步处理的handler(自定义)
pipeline.addLast(new MyServerHandler());
}
});
//发动服务器
ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
public class MyServerHandler extends ChannelInboundHandlerAdapter {
// 事情触发
@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("服务器做相应处理.." + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
//如果产生闲暇，咱们封闭通道
// ctx.channel().close();
}
}
}
客户端运用群聊 那个就能够，服务端执行成果如下：
/127.0.0.1:54273--超时时刻--读闲暇
服务器做相应处理..2022-05-07 17:22:53
/127.0.0.1:54273--超时时刻--写闲暇
服务器做相应处理..2022-05-07 17:22:56
/127.0.0.1:54273--超时时刻--读写闲暇
服务器做相应处理..2022-05-07 17:23:03
/127.0.0.1:54273--超时时刻--读闲暇
服务器做相应处理..2022-05-07 17:23:03
/127.0.0.1:54273--超时时刻--写闲暇
服务器做相应处理..2022-05-07 17:23:09
3. WebSocket编程完成长衔接
实例要求:
Http协议是无状况的, 浏览器和服务器间的恳求响应一次，下一次会从头创立衔接.
要求：完成基于webSocket的长衔接的全双工的交互
改变Http协议屡次恳求的束缚，完成长衔接了， 服务器能够发送音讯给浏览器
客户端浏览器和服务器端会相互感知，比方服务器封闭了，浏览器会感知，同样浏览器封闭了，服务器会感知
服务端：
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.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/hello 表明恳求的uri
4. WebSocketServerProtocolHandler 核心功能是将 http协议升级为 ws协议 , 坚持长衔接
5. 是经过一个 状况码 101
*/
pipeline.addLast(new WebSocketServerProtocolHandler("/hello2"));
//自定义的handler ，处理业务逻辑
pipeline.addLast(new MyTextWebSocketFrameHandler());
}
});
//发动服务器
ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
channelFuture.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
//这儿 TextWebSocketFrame 类型，表明一个文本帧(frame)
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()));
}
//当web客户端衔接后， 触发办法
@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(); //封闭衔接
}
}
参考文档
Netty学习和源码分析github地址
Netty从入门到精通视频教程(B站)
Netty威望攻略 第二版
