package com.kenson.netty.server;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class NettyServer {/*** 端口*/private int port;public NettyServer(int port) {this.port = port;}public void run() {//EventLoopGroup是用来处理IO操作的多线程事件循环器//负责接收客户端连接线程EventLoopGroup bossGroup = new NioEventLoopGroup();//负责处理客户端i/o事件、task任务、监听任务组EventLoopGroup workerGroup = new NioEventLoopGroup();//启动 NIO 服务的辅助启动类ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup);//配置 Channelbootstrap.channel(NioServerSocketChannel.class);bootstrap.childHandler(new ServerIniterHandler());//BACKLOG用于构造服务端套接字ServerSocket对象，// 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度bootstrap.option(ChannelOption.SO_BACKLOG, 1024);//是否启用心跳保活机制bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);try {//绑定服务端口监听Channel channel = bootstrap.bind(port).sync().channel();System.out.println("server run in port " + port);//服务器关闭监听/*channel.closeFuture().sync()实际是如何工作:channel.closeFuture()不做任何操作，只是简单的返回channel对象中的closeFuture对象，对于每个Channel对象，都会有唯一的一个CloseFuture，用来表示关闭的Future，所有执行channel.closeFuture().sync()就是执行的CloseFuturn的sync方法，从上面的解释可以知道，这步是会将当前线程阻塞在CloseFuture上*/channel.closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {//关闭事件流组
            bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {new NettyServer(8899).run();}
}

package com.kenson.netty.server;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;public class ServerHandler extends SimpleChannelInboundHandler<String> {/*** 所有的活动用户*/public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 读取消息通道** @param context* @param s* @throws Exception*/@Overrideprotected void channelRead0(ChannelHandlerContext context, String s)throws Exception {Channel channel = context.channel();//当有用户发送消息的时候，对其他的用户发送消息for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[you]: " + s + "\n");} else {ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");}}System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");}/*** 处理新加的消息通道** @param ctx* @throws Exception*/@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");}}group.add(channel);}/*** 处理退出消息通道** @param ctx* @throws Exception*/@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();for (Channel ch : group) {if (ch == channel) {ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");}}group.remove(channel);}/*** 在建立连接时发送消息** @param ctx* @throws Exception*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();boolean active = channel.isActive();if (active) {System.out.println("[" + channel.remoteAddress() + "] is online");} else {System.out.println("[" + channel.remoteAddress() + "] is offline");}ctx.writeAndFlush("[server]: welcome");}/*** 退出时发送消息** @param ctx* @throws Exception*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {Channel channel = ctx.channel();if (!channel.isActive()) {System.out.println("[" + channel.remoteAddress() + "] is offline");} else {System.out.println("[" + channel.remoteAddress() + "] is online");}}/*** 异常捕获** @param ctx* @param e* @throws Exception*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {Channel channel = ctx.channel();System.out.println("[" + channel.remoteAddress() + "] leave the room");ctx.close().sync();}}

package com.kenson.netty.server;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//管道注册handlerChannelPipeline pipeline = socketChannel.pipeline();//编码通道处理pipeline.addLast("decode", new StringDecoder());//转码通道处理pipeline.addLast("encode", new StringEncoder());//聊天服务通道处理pipeline.addLast("chat", new ServerHandler());}
}

package com.kenson.netty.client;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;public class NettyClient {private String ip;private int port;private boolean stop = false;public NettyClient(String ip, int port) {this.ip = ip;this.port = port;}public void run() throws IOException {//设置一个多线程循环器EventLoopGroup workerGroup = new NioEventLoopGroup();//启动附注类Bootstrap bootstrap = new Bootstrap();bootstrap.group(workerGroup);//指定所使用的NIO传输channelbootstrap.channel(NioSocketChannel.class);//指定客户端初始化处理bootstrap.handler(new ClientIniterHandler());try {//连接服务Channel channel = bootstrap.connect(ip, port).sync().channel();while (true) {//向服务端发送内容BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));String content = reader.readLine();if (StringUtils.isNotEmpty(content)) {if (StringUtils.equalsIgnoreCase(content, "q")) {System.exit(1);}channel.writeAndFlush(content);}}} catch (InterruptedException e) {e.printStackTrace();System.exit(1);} finally {workerGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {new NettyClient("127.0.0.1", 8899).run();}
}

package com.kenson.netty.client;import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;public class ClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {//打印服务端的发送数据
        System.out.println(s);}
}

package com.kenson.netty.client;import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//注册管道ChannelPipeline pipeline = socketChannel.pipeline();pipeline.addLast("decoder", new StringDecoder());pipeline.addLast("encoder", new StringEncoder());pipeline.addLast("http", new HttpClientCodec());pipeline.addLast("chat", new ClientHandler());}
}