Netty应⽤实例-聊系统，⼼跳检测机制案例，WebSocket编程
实现服务器和客户端长连接
实例要求:
1) 编写⼀个 Netty 聊系统，实现服务器端和客户端之间的数据简单通讯（⾮阻塞）
2) 实现多⼈聊
3) 服务器端：可以监测⽤户上线，离线，并实现消息转发功能
4) 客户端：通过 channel 可以⽆阻塞发送消息给其它所有⽤户，同时可以接受其它⽤户发送的消息(有服务器转发得到)
5) ⽬的：进⼀步理解 Netty
代码：
GroupChatServer
import ioty.bootstrap.ServerBootstrap;
import ioty.channel.*;
import ioty.channel.nio.NioEventLoopGroup;
import ioty.channel.socket.SocketChannel;
import ioty.channel.socket.nio.NioServerSocketChannel;
import dec.string.StringDecoder;
import dec.string.StringEncoder;
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();
.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());
bootstrap检验方法}
});
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();
}
}
View Code
GroupChatServerHandler
import ioty.channel.Channel;
import ioty.channel.ChannelHandlerContext;
import ioty.channel.SimpleChannelInboundHandler;
import up.ChannelGroup;
import up.DefaultChannelGroup;
import urrent.GlobalEventExecutor;
SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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);
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("[客户端]" + Address() + " 加⼊聊天" + 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("[客户端]" + Address() + " 离开了\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("[客户]" + Address() + " 发送了消息" + msg + "\n"); }else {//回显⾃⼰发送的消息给⾃⼰
ch.writeAndFlush("[⾃⼰]发送了消息" + msg + "\n");
}
});
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭通道
ctx.close();
}
}
View Code
GroupChatClient
import ioty.bootstrap.Bootstrap;
import ioty.channel.*;
import ioty.channel.nio.NioEventLoopGroup;
import ioty.channel.socket.SocketChannel;
import ioty.channel.socket.nio.NioSocketChannel;
import dec.string.StringDecoder;
import dec.string.StringEncoder;
import java.util.Scanner;
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 = t(host, port).sync();
//得到channel
Channel channel = channelFuture.channel();
System.out.println("-------" + channel.localAddress()+ "--------");
//客户端需要输⼊信息，创建⼀个扫描器
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
String msg = Line();
//通过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();
}
}
View Code
GroupChatClientHandler
import ioty.channel.ChannelHandlerContext;
import ioty.channel.SimpleChannelInboundHandler;
public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { System.out.im());
}
}
View Code
User
public class User {
private int id;
private String pwd;
}
View Code
Netty ⼼跳检测机制案例
实例要求:
1) 编写⼀个 Netty ⼼跳检测机制案例, 当服务器超过 3 秒没有读时，就提⽰读空闲
2) 当服务器超过 5 秒没有写操作时，就提⽰写空闲
3) 实现当服务器超过 7 秒没有读或者写操作时，就提⽰读写空闲
4) 代码如下：
MyServer
import ioty.bootstrap.ServerBootstrap;
import ioty.channel.ChannelFuture;
import ioty.channel.ChannelInitializer;
import ioty.channel.ChannelPipeline;
import ioty.channel.EventLoopGroup;
import ioty.channel.nio.NioEventLoopGroup;
import ioty.channel.socket.SocketChannel;
import ioty.channel.socket.nio.NioServerSocketChannel;
import ioty.handler.logging.LogLevel;
import ioty.handler.logging.LoggingHandler;
import ioty.handler.timeout.IdleStateHandler;
import urrent.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.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();
}
}
}
View Code
MyServerHandler
import ioty.channel.ChannelHandlerContext;
import ioty.channel.ChannelInboundHandlerAdapter;
import ioty.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();
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。
发表评论