public class TestServer {
public static void main(String[] args) {
//创建bossGroup和workGroup两个工作组
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
try {
//创建bootstrap来组装参数
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup,workGroup) //设置线程组
.channel(NioServerSocketChannel.class) //设置通道类型
.childHandler(new TestServerInitializer()); //设置处理器
//绑定端口 同时开启同步
ChannelFuture future = bootstrap.bind(8888).sync();
//对发生闭关通道的时候 做监听处理
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//获取管道
ChannelPipeline pipeline = ch.pipeline();
//加入一个 netty 提供的 httpServerCodec codec =>[coder - decoder]
// HttpServerCodec 说明
// 1. HttpServerCodec 是 netty 提供的处理 http 的 编-解码器
pipeline.addLast("MyHttpServerCodec",new HttpServerCodec());
//2. 管道中添加业务处理handle
pipeline.addLast(new TestHttpServerHandler());
}
}
/*
说明1. SimpleChannelInboundHandler 是 ChannelInboundHandlerAdapter
2. HttpObject 客户端和服务器端相互通讯的数据被封装成 HttpObject
*/
public class TestHttpServerHandler extends SimpleChannelInboundHandler<HttpObject> {
//channelRead0 读取客户端数据
@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
//判断ms是不是 httpRequest请求
if (msg instanceof HttpRequest) {
System.out.println("pipeline hashcode" + ctx.pipeline().hashCode() + " TestHttpServerHandler hash=" + this.hashCode());
System.out.println("msg 类型=" + msg.getClass());
System.out.println("客户端地址" + ctx.channel().remoteAddress());
HttpRequest httpRequest = (HttpRequest) msg;
//获取 uri, 过滤指定的资源
URI uri = new URI(httpRequest.uri());
ByteBuf content = Unpooled.copiedBuffer("hello, 我是服务器", CharsetUtil.UTF_8);
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
//将构建好 response 返回
ctx.writeAndFlush(response);
}
}
}
实例要求:
public class GroupChatServer {
//监听端口
private int port;
public GroupChatServer(int port) {
this.port = port;
}
//编写run方法，处理客户端请求
public void run() {
//创建两个线程组一个处理连接，一个处理业务
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
try {
//创建启动器连接参数
ServerBootstrap bootstrap = new ServerBootstrap();
//开始组装参数
bootstrap.group(bossGroup, workGroup)
.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 {
//获取上下文来处理数据 在管道中添加业务处理handle
ChannelPipeline pipeline = ch.pipeline();
//pipeline加入解码器
pipeline.addLast("decoder", new StringDecoder());
//pipeline加入编码器
pipeline.addLast("encoder", new StringEncoder());
//加入业务的handle
pipeline.addLast(new GroupChatServerHandler());
}
});
System.out.println("netty 服务器启动");
ChannelFuture channelFuture = bootstrap.bind(port).sync();
//监听关闭
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭线程组
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
public static void main(String[] args) {
new GroupChatServer(7000).run();
}
}
public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
//通道集合
private static List<Channel> channelList = new ArrayList<Channel>();
//使用一个hashMap管理
public static Map<String, Channel> map = new HashMap<>();
//GlobalEventExecutor.INSTANCE) 是全局的事件执行器，是一个单例
private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//处理业务 handleAdded表示连接建立，一但连接，第一个被执行
@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 处于不活动状态, 提示 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.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();
}
}
public class GroupChatClient {
//属性
private String host;
private int port;
public GroupChatClient(String host, int port) {
this.host = host;
this.port = port;
}
public void run() {
NioEventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
//获取上下文来处理数据 在管道中添加业务处理handle
ChannelPipeline pipeline = ch.pipeline();
//pipeline加入解码器
pipeline.addLast("decoder", new StringDecoder());
//pipeline加入编码器
pipeline.addLast("encoder", new StringEncoder());
//加入业务的handle
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");
}
} catch (Exception e) {
e.printStackTrace();
} 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());
}
}
编写一个 Netty 心跳检测机制案例, 当服务器超过 3 秒没有读时，就提示读空闲
当服务器超过 5 秒没有写操作时，就提示写空闲
实现当服务器超过 7 秒没有读或者写操作时，就提示读写空闲
public class MyServer {
public static void main(String[] args) {
//创建两个线程组 一个负责连接 一个复杂处理业务
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
NioEventLoopGroup workGroup = new NioEventLoopGroup();
try {
//创建连接器
ServerBootstrap bootstrap = new ServerBootstrap();
//开始组装业务链
bootstrap.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.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(13,5,2, TimeUnit.SECONDS));
//加入一个对空闲检测进一步处理的 handler(自定义)
pipeline.addLast(new MyServerHandler());
}
});
ChannelFuture channelFuture = bootstrap.bind(7000).sync();
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
//关闭当前线程
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
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) {
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();
}
}
}
因篇幅问题不能全部显示，请点此查看更多更全内容