微言Netty：手写分布式服务框架

<span class="code-snippet_outer"><span class="code-snippet__keyword">public</span> <span class="code-snippet__keyword">class</span> <span class="code-snippet__title">NettyClient</span> {    <span class="code-snippet__comment">/**     * 日志记录     */</span>    <span class="code-snippet__keyword">private</span> <span class="code-snippet__keyword">static</span> final Logger logger = LoggerFactory.getLogger(NettyClient.class);    <span class="code-snippet__comment">/**     * 客户端请求Future列表     */</span>    <span class="code-snippet__keyword">private</span> Map<String, TinyWhaleFuture> clientFutures = <span class="code-snippet__keyword">new</span> ConcurrentHashMap<>();    <span class="code-snippet__comment">/**     * 客户端业务处理handler     */</span>    <span class="code-snippet__keyword">private</span> ClientHandler clientHandler = <span class="code-snippet__keyword">new</span> ClientHandler(clientFutures);    <span class="code-snippet__comment">/**     * 事件池     */</span>    <span class="code-snippet__keyword">private</span> EventLoopGroup <span class="code-snippet__keyword">group</span> = <span class="code-snippet__keyword">new</span> NioEventLoopGroup();    <span class="code-snippet__comment">/**     * 启动器     */</span>    <span class="code-snippet__keyword">private</span> Bootstrap bootstrap = <span class="code-snippet__keyword">new</span> Bootstrap();    <span class="code-snippet__comment">/**     * 客户端通道     */</span>    <span class="code-snippet__keyword">private</span> Channel clientChannel;    <span class="code-snippet__comment">/**     * 客户端连接     * @param host     * @param port     * @throws InterruptedException     */</span>    <span class="code-snippet__function"><span class="code-snippet__keyword">public</span> <span class="code-snippet__title">NettyClient</span>(<span class="code-snippet__params">String host, <span class="code-snippet__keyword">int</span> port</span>) throws InterruptedException</span> {        bootstrap.<span class="code-snippet__keyword">group</span>(<span class="code-snippet__keyword">group</span>)                .channel(NioSocketChannel.class)                .option(ChannelOption.TCP_NODELAY, <span class="code-snippet__literal">true</span>)                .handler(<span class="code-snippet__keyword">new</span> ChannelInitializer<SocketChannel>() {                    @<span class="code-snippet__function">Override                    <span class="code-snippet__keyword">protected</span> <span class="code-snippet__keyword">void</span> <span class="code-snippet__title">initChannel</span>(<span class="code-snippet__params">SocketChannel channel</span>) throws Exception</span> {                        <span class="code-snippet__comment">//通道空闲检测                        channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(45, 45, 120));                        //解码器                        channel.pipeline().addLast("nettyMessageDecoder", new NettyMessageDecoder(1024 * 1024, 4, 4));                        //编码器                        channel.pipeline().addLast("nettyMessageEncoder", new NettyMessageEncoder());                        //心跳处理                        channel.pipeline().addLast("heartBeatHandler", new HeartBeatRequestHandler());                        //业务处理                        channel.pipeline().addLast("clientHandler", clientHandler);                        //鉴权处理                        channel.pipeline().addLast("loginAuthHandler", new LoginAuthRequestHandler());                    }                });        //发起同步连接操作        ChannelFuture channelFuture = bootstrap.connect(host, port);        //注册连接事件        channelFuture.addListener((ChannelFutureListener)future -> {            //如果连接成功            if (future.isSuccess()) {                logger.info("客户端[" + channelFuture.channel().localAddress().toString() + "]已连接...");                clientChannel = channelFuture.channel();            }            //如果连接失败，尝试重新连接            else{                logger.info("客户端[" + channelFuture.channel().localAddress().toString() + "]连接失败，重新连接中...");                future.channel().close();                bootstrap.connect(host, port);            }        });        //注册关闭事件        channelFuture.channel().closeFuture().addListener(cfl -> {            close();            logger.info("客户端[" + channelFuture.channel().localAddress().toString() + "]已断开...");        });    }    /**     * 客户端关闭     */    private void close() {        //关闭客户端套接字        if(clientChannel!=null){            clientChannel.close();        }        //关闭客户端线程组        if (group != null) {            group.shutdownGracefully();        }    }    /**     * 客户端发送消息，将获取的Future句柄保存到clientFutures列表     * @return     * @throws InterruptedException     * @throws ExecutionException     */    public TinyWhaleFuture send(NettyMessage<NettyRequest> request) {        TinyWhaleFuture rpcFuture = new TinyWhaleFuture(request);        rpcFuture.addCallback(new TinyWhaleAsyncCallback() {            @Override            public void success(Object result) {            }            @Override            public void fail(Exception e) {                logger.error("发送失败", e);            }        });        clientFutures.put(request.getBody().getRequestId(), rpcFuture);        clientHandler.sendMessage(request);        return rpcFuture;    }}</span></span>