Netty实时聊天实践

原文请参考微信公众账号（包含图文）：https://mp.weixin.qq.com/s/o2AGrNd8rZHp9DbDppjI2Q 

Netty数据传输

Netty数据传输流程：客户端发送数据（出站）时，将消息流存放到缓冲区（ByteBuf），然后将ByteBuf数据传递给ChannelPipeline中的第一个ChannelOutBoundHandler，然后再传递给第二个ChannelOutBoundHandler，如果ChannelPipeline绑定有多个ChannelOutBoundHandler，则需要一层层传递下去，其中可以在ChannelOutBoundHandler中进行编码（Encoder）操作，接着会将ByteBuf数据从ChannelOutBoundHandler传递给ChannelInBoundHandler，如果ChannelPipeline绑定有多个ChannelInBoundHandler，则需一层层传递下去，可在ChannelInBoundHandler中进行解码（Decoder）操作,然后服务端收到消息（入站）后，将消息进行解码Decoder操作；两个BoundHandler之间的数据传递是通过ChannelHandlerContext来实现；

Netty实现聊天功能

ChatServerHandler

1. publicclassChatServerHandlerextendsSimpleChannelInboundHandler<String>{

2.  

3.    privatefinalChannelGroup group;

4.  

5.    publicChatServerHandler(ChannelGroup group){

6.        this.group = group;

7.    }

8.  

9.    /**

10.     * 将新加入的连接放入group中

11.     * @param ctx

12.     * @throws Exception

13.     */

14.    @Override

15.    publicvoid handlerAdded(ChannelHandlerContext ctx)throwsException{

16.        group.add(ctx.channel());

17.    }

18.  

19.    /**

20.     * 删除group中下线的连接

21.     * @param ctx

22.     * @throws Exception

23.     */

24.    @Override

25.    publicvoid handlerRemoved(ChannelHandlerContext ctx)throwsException{

26.        group.remove(ctx.channel());

27.    }

28.  

29.  

30.    @Override

31.    protectedvoid channelRead0(ChannelHandlerContext ctx,String msg)throwsException{

32.        if(StringUtils.isEmpty(msg)){

33.            return;

34.        }

35.        Channel incoming = ctx.channel();

36.        for(Channel channel : group){

37.            if(channel != incoming){

38.                channel.writeAndFlush(msg +"\n");

39.            }else{

40.                channel.writeAndFlush(msg +"\n");

41.            }

42.        }

43.    }

44. }

服务端引导类ChatServer

1. publicclassChatServer{

2.  

3.    //创建 DefaultChannelGroup 用来 保存所有连接的的 WebSocket channel

4.    privatefinalChannelGroup channelGroup =newDefaultChannelGroup(GlobalEventExecutor.INSTANCE);

5.  

6.    publicvoid start(int port){

7.        EventLoopGroup bossGroup =newNioEventLoopGroup();

8.        EventLoopGroup workerGroup =newNioEventLoopGroup();

9.  

10.        try{

11.            ServerBootstrap bootstrap =newServerBootstrap();

12.            bootstrap.group(bossGroup, workerGroup)

13.                    .channel(NioServerSocketChannel.class)

14.                    .childHandler(newChannelInitializer<SocketChannel>(){

15.                        @Override

16.                        protectedvoid initChannel(SocketChannel ch)throwsException{

17.                            ChannelPipeline pipeline = ch.pipeline();

18.                            pipeline.addLast("framer",newDelimiterBasedFrameDecoder(8192,Delimiters.lineDelimiter()));

19.                            pipeline.addLast("decoder",newStringDecoder());

20.                            pipeline.addLast("encoder",newStringEncoder());

21.                            pipeline.addLast(newChatServerHandler(channelGroup));

22.                        }

23.                    })

24.                    .option(ChannelOption.SO_BACKLOG,128)

25.                    .childOption(ChannelOption.SO_KEEPALIVE,true);

26.            ChannelFuture future = bootstrap.bind(port).sync();

27.            future.channel().closeFuture().sync();

28.        }catch(Exception e){

29.            e.printStackTrace();

30.        }finally{

31.            bossGroup.shutdownGracefully();

32.            workerGroup.shutdownGracefully();

33.        }

34.    }

35.  

36.    publicstaticvoid main(String[] args){

37.       newChatServer().start(8080);

38.    }

39.  

40. }

客户端ChatClientHandler

1. publicclassChatClientHandlerextendsSimpleChannelInboundHandler<String>{

2.    /** 控制台不打印没有的信息 */

3.    @Override

4.    protectedvoid channelRead0(ChannelHandlerContext ctx,String msg)throwsException{

5.        if(!msg.contains("HTTP/1.1")&&!msg.contains("Host:")&&!msg.contains("Proxy-Connection: keep-alive")&&!msg.contains("User-Agent:")){

6.            System.out.println(msg);

7.        }

8.    }

9. }

客户端引导类ChatClientA

1. publicclassChatClientA{

2.  

3.    publicstaticvoid main(String[] args){

4.        newChatClientA().start();

5.    }

6.  

7.  

8.    publicvoid start(){

9.        EventLoopGroup group =newNioEventLoopGroup();

10.        try{

11.            Bootstrap bootstrap =newBootstrap();

12.            bootstrap.group(group)

13.                    .channel(NioSocketChannel.class)

14.                    .handler(newChannelInitializer<SocketChannel>(){

15.                @Override

16.                protectedvoid initChannel(SocketChannel ch)throwsException{

17.                    ch.pipeline().addLast("framer",newDelimiterBasedFrameDecoder(8192,Delimiters.lineDelimiter()));

18.                    ch.pipeline().addLast("decoder",newStringDecoder());

19.                    ch.pipeline().addLast("encoder",newStringEncoder());

20.                    ch.pipeline().addLast("handler",newChatClientHandler());

21.  

22.                }

23.            });

24.            ChannelFuture future = bootstrap.connect("localhost",8080).sync();

25.            Channel channel = future.channel();

26.            BufferedReader in =newBufferedReader(newInputStreamReader(System.in));

27.            while(true){

28.                channel.writeAndFlush("ChatClientA说："+ in.readLine()+"\r\n");

29.            }

30.        }catch(Exception e){

31.            e.printStackTrace();

32.        }finally{

33.            group.shutdownGracefully();

34.        }

35.  

36.    }

37.  

38. }

客户端引导类ChatClientB

1. publicclassChatClientB{

2.  

3.    publicstaticvoid main(String[] args){

4.        newChatClientB().start();

5.    }

6.  

7.    publicvoid start(){

8.        EventLoopGroup group =newNioEventLoopGroup();

9.        try{

10.            Bootstrap bootstrap =newBootstrap();

11.            bootstrap.group(group)

12.                    .channel(NioSocketChannel.class)

13.                    .handler(newChannelInitializer<SocketChannel>(){

14.                @Override

15.                protectedvoid initChannel(SocketChannel ch)throwsException{

16.                    ch.pipeline().addLast("framer",newDelimiterBasedFrameDecoder(8192,Delimiters.lineDelimiter()));

17.                    ch.pipeline().addLast("decoder",newStringDecoder());

18.                    ch.pipeline().addLast("encoder",newStringEncoder());

19.                    ch.pipeline().addLast("handler",newChatClientHandler());

20.  

21.                }

22.            });

23.            ChannelFuture future = bootstrap.connect("localhost",8080).sync();

24.            Channel channel = future.channel();

25.            BufferedReader in =newBufferedReader(newInputStreamReader(System.in));

26.            while(true){

27.                channel.writeAndFlush("ChatClient说："+ in.readLine()+"\r\n");

28.            }

29.        }catch(Exception e){

30.            e.printStackTrace();

31.        }finally{

32.            group.shutdownGracefully();

33.        }

34.  

35.    }

36.  

37. }

ChatClientA与ChatClientB代码一模一样，因为需要实现聊天功能，所以需要启用至少2个客户端，故Copy一样的代码，分别启动ChatClientA与ChatClientB；

先启动ChatServer.main()，然后分别启动ChatClientA与ChatClientB的main()方法

分别在ChatClientA和ChatClientB的控制台输入聊天内容，聊天结果如下：

ClientA：

ClientB：

 

 

 

微信扫一扫
关注该公众号