Netty 多客戶端連接與通信

本文轉載自查看原文 2019-08-03 22:22 1410 S. 並發與Netty

實現場景：聊天

服務端，客戶端A，客戶端B，客戶端C。當客戶端發送消息給服務端后，服務端在將這條消息廣播個所有客戶端戶端A，客戶端B，客戶端C。

需求1：客戶端上線后，會通知所有客戶端上線。

如客戶端A先建立連接，不需要通知。

當客戶端B與服務端建立連接，服務端告訴A，客戶端B上線。

A和B建立連接后，客戶端C和服務端建立連接。服務端廣播一條信息給A和B。

需求2： A、B、C都已經建立連接，當A發送一條給服務端，服務端廣播這條消息給所有客戶端，客戶端A會提示這條消息是自己發送的。

一、服務端程序的編寫

1、MyChartServer 類

public class MyChartServer {
    public static void main(String[] args) throws  Exception{
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try{

            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new MyChatServerInitializer());

            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

2、MyChatServerInitializer 類

public class MyChatServerInitializer extends ChannelInitializer<SocketChannel>{

    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new MyChartServerHandle());
    }
}

3、MyChartServerHandle 類

public class MyChartServerHandle extends SimpleChannelInboundHandler<String>{


    //用於保存所有Channel對象
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //channel為當前客戶端發給服務端的channel
        Channel channel = ctx.channel();
        channelGroup.forEach(ch -> {

            if(channel != ch){
                ch.writeAndFlush(channel.remoteAddress() + " 發送的消息:" + msg + "\n");
            } else{
                ch.writeAndFlush("[自己]" + msg + " \n");
            }
        });

    }

    //表示連接建立
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
       //chanel可以理解成Connection
        Channel channel = ctx.channel();
        //廣播消息給所有的客戶端
        channelGroup.writeAndFlush("[服務器] - " + channel.remoteAddress() + " 加入\n");
        channelGroup.add(channel);
    }

    //表示連接斷掉了
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //廣播消息給所有的客戶端
        channelGroup.writeAndFlush("[服務器] - " + channel.remoteAddress() + " 離開\n");
        //下面這行代碼Netty會自動調用
        //channelGroup.remove(channel);
    }

    //表示連接時活動狀態
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //廣播消息給所有的客戶端
        CommonUtil.println( channel.remoteAddress() + " 上線 \n");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        //廣播消息給所有的客戶端
        CommonUtil.println( channel.remoteAddress() + " 下線 \n");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();;
        ctx.close();
    }
}

　　二、客戶端程序編寫

1、MyChatClient類

public class MyChatClient {
    public static void main(String[] args) throws  Exception{
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new MyChatClientInitializer());

            //channel表示與服務端的Connection
            Channel channel = bootstrap.connect("localhost",8899).sync().channel();
            //不斷讀取客戶端輸入的內容
            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));

            for(;;){
                channel.writeAndFlush(br.readLine() + "\r\n");
            }

        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

　　2、MyChatClientInitializer 類

public class MyChatClientInitializer  extends ChannelInitializer<SocketChannel> {



    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new MyChartClientHandle());
    }
}

3、MyChartClientHandle 類

public class MyChartClientHandle extends SimpleChannelInboundHandler<String> {

    // 對於客戶端來說，輸入來自控制台
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //僅僅打印來自服務端的信息
        CommonUtil.println(msg);

    }


}

三、測試

1、啟動MyChartServer，然后啟動MyChatClient

MyChartServer打印 59786 上線