Netty学习05--Netty5客户端

曾经终败给现在 2023-02-17 12:28 48阅读 0赞

目录结构：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70][]

1.Client

package com.cxb.netty5.client;
    
    import java.io.BufferedReader;
    import java.io.InputStreamReader;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    /**
     * netty5的客户端
     */
    public class Client {
    
        public static void main(String[] args) {
            //服务类
            Bootstrap bootstrap = new Bootstrap();
    
            //worker
            EventLoopGroup worker = new NioEventLoopGroup();
    
            try {
                //设置线程池
                bootstrap.group(worker);
                //设置socket工厂、
                bootstrap.channel(NioSocketChannel.class);
                //设置管道
                bootstrap.handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new StringEncoder());
                        ch.pipeline().addLast(new ClientHandler());
                    }
                });
    
                ChannelFuture connect = bootstrap.connect("127.0.0.1", 8006);
    
                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
                while (true) {
                    System.out.println("请输入：");
                    String msg = bufferedReader.readLine();
                    connect.channel().writeAndFlush(msg);
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                worker.shutdownGracefully();
            }
        }
    }

2.ClientHandler

package com.cxb.netty5.client;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    /**
     * 客户端消息处理
     */
    public class ClientHandler extends SimpleChannelInboundHandler<String> {
    
    	@Override
    	protected void messageReceived(ChannelHandlerContext ctx, String msg) throws Exception {
    		System.out.println("客户端收到消息:"+msg);
    	}
    
    }

3.MultClient

package com.cxb.netty5.client;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.atomic.AtomicInteger;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    /**
     * 多连接客户端
     */
    public class MultClient {
    
        /**
         * 服务类
         */
        private Bootstrap bootstrap = new Bootstrap();
    
        /**
         * 会话
         */
        private List<Channel> channels = new ArrayList<>();
    
        /**
         * 引用计数
         */
        private final AtomicInteger index = new AtomicInteger();
    
        /**
         * 初始化
         *
         * @param count
         */
        public void init(int count) {
    
            //worker
            EventLoopGroup worker = new NioEventLoopGroup();
    
            //设置线程池
            bootstrap.group(worker);
            //设置socket工厂、
            bootstrap.channel(NioSocketChannel.class);
            //设置管道
            bootstrap.handler(new ChannelInitializer<Channel>() {
    
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new StringDecoder());
                    ch.pipeline().addLast(new StringEncoder());
                    ch.pipeline().addLast(new ClientHandler());
                }
            });
    
            for (int i = 1; i <= count; i++) {
                ChannelFuture future = bootstrap.connect("192.168.3.112", 8006);
                channels.add(future.channel());
            }
        }
    
        /**
         * 获取会话
         *
         * @return
         */
        public Channel nextChannel() {
            return getFirstActiveChannel(0);
        }
    
    
        private Channel getFirstActiveChannel(int count) {
            Channel channel = channels.get(Math.abs(index.getAndIncrement() % channels.size()));
            if (!channel.isActive()) {
                //重连
                reconnect(channel);
                if (count >= channels.size()) {
                    throw new RuntimeException("no can use channel");
                }
                return getFirstActiveChannel(count + 1);
            }
            return channel;
        }
    
        /**
         * 重连
         *
         * @param channel
         */
        private void reconnect(Channel channel) {
            synchronized (channel) {
                if (channels.indexOf(channel) == -1) {
                    return;
                }
    
                Channel newChannel = bootstrap.connect("192.168.3.112", 8006).channel();
                channels.set(channels.indexOf(channel), newChannel);
            }
        }
    
    }

4.Start

package com.cxb.netty5.client;
    
    import java.io.BufferedReader;
    import java.io.InputStreamReader;
    /**
     * 启动类
     */
    public class Start {
    
    	public static void main(String[] args) {
    
    		MultClient client = new MultClient();
    		client.init(5);
    		
    		BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in));
    		while(true){
    			try {
    				System.out.println("请输入:");
    				String msg = bufferedReader.readLine();
    				client.nextChannel().writeAndFlush(msg);
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    }

测试，启动服务端，和客户端。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 2][]

客户端输入任意字符，服务端收到消息之后，返回hi给客户端。

[代码下载][Link 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200614093644214.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200614094539713.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200614094530987.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMzcxNzY2,size_16,color_FFFFFF,t_70
[Link 1]: https://github.com/ChenXbFrank/nettyLearnCode