1 public class MyChatServer {
 2 
 3     public static void main(String[] args) throws InterruptedException {
 4 
 5 
 6         EventLoopGroup bossGroup = new NioEventLoopGroup();
 7         EventLoopGroup workerGroup = new NioEventLoopGroup();
 8 
 9         try{
10             ServerBootstrap serverBootstrap = new ServerBootstrap();
11             serverBootstrap.group(bossGroup,workerGroup).
12                     channel(NioServerSocketChannel.class)
13                     .childHandler(new MyChatServerInitializer());
14 
15             ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
16             channelFuture.channel().closeFuture().sync();
17         }finally {
18             bossGroup.shutdownGracefully();
19             workerGroup.shutdownGracefully();
20         }
21 
22 
23     }
24 
25 }

 1 public class MyChatServerInitializer  extends ChannelInitializer<SocketChannel> {
 2     @Override
 3     protected void initChannel(SocketChannel ch) throws Exception {
 4 
 5         ChannelPipeline channelPipeline = ch.pipeline();
 6 
 7         channelPipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
 8         channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
 9         channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
10         channelPipeline.addLast(new MyChatServerHandler());
11 
12     }
13 }

 1 public class MyChatServerHandler extends SimpleChannelInboundHandler<String> {
 2 
 3     //连接实例集合
 4     private  static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 5 
 6 
 7     @Override
 8     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
 9 
10         Channel channel = ctx.channel();
11 
12         channelGroup.forEach(ch ->{
13             if(channel != ch){
14                 ch.writeAndFlush(channel.remoteAddress()+"发送消息" + msg + "\n");
15             }else{
16                 ch.writeAndFlush("[自己]"+"\n");
17             }
18         });
19     }
20 
21 
22     @Override
23     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
24 
25         Channel channel = ctx.channel();
26         channelGroup.writeAndFlush("[服务器] - " + channel.remoteAddress() + "加入\n");
27         channelGroup.add(channel);
28     }
29 
30     @Override
31     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
32 
33         Channel channel = ctx.channel();
34         channelGroup.writeAndFlush("[服务器 - ]"+ channel.remoteAddress() + "离开\n");
35         System.out.println("channelGroup size"+ channelGroup.size());
36     }
37 
38     @Override
39     public void channelActive(ChannelHandlerContext ctx) throws Exception {
40         Channel channel = ctx.channel();
41         System.out.println(channel.remoteAddress()+"上线\n");
42     }
43 
44     @Override
45     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
46         Channel channel = ctx.channel();
47         System.out.println(channel.remoteAddress()+"下线\n");
48     }
49 
50     @Override
51     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
52         cause.printStackTrace();
53         ctx.close();
54     }
55 }

 1 public class MyChatClient {
 2 
 3     public static void main(String[] args) throws InterruptedException, IOException {
 4 
 5         EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
 6 
 7         try{
 8             Bootstrap bootstrap = new Bootstrap();
 9             bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
10                     .handler(new MyChatClientInitializer());
11 
12             Channel channel = bootstrap.connect("localhost",8899).sync().channel();
13 
14             BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
15             for(;;){
16                 channel.writeAndFlush(br.readLine() + "\r\n");
17             }
18 
19         }finally {
20             eventLoopGroup.shutdownGracefully();
21         }
22 
23     }
24 
25 }

 1 public class MyChatClientInitializer extends ChannelInitializer<SocketChannel> {
 2     @Override
 3     protected void initChannel(SocketChannel ch) throws Exception {
 4 
 5         ChannelPipeline channelPipeline = ch.pipeline();
 6 
 7         channelPipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
 8         channelPipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
 9         channelPipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
10 
11         channelPipeline.addLast(new MyChatClientHandler());  //自己的处理器
12 
13     }
14 }

1 public class MyChatClientHandler extends SimpleChannelInboundHandler<String> {
2     @Override
3     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
4         System.out.println(msg);
5     }
6 }