我们上一章的例子中，我们的调用是在客户端启动的时候完成的。
实际使用中，我们希望调用可以有客户端主动发起。
通过 calculate 方法，我们就可以主动发起请求。
import com.github.houbb.log.integration.core.Log; import com.github.houbb.log.integration.core.LogFactory; import com.github.houbb.rpc.client.decoder.CalculateResponseDecoder; import com.github.houbb.rpc.client.encoder.CalculateRequestEncoder; import com.github.houbb.rpc.client.handler.RpcClientHandler; import com.github.houbb.rpc.common.constant.RpcConstant; import com.github.houbb.rpc.common.model.CalculateRequest; import com.github.houbb.rpc.common.model.CalculateResponse; import io.netty.bootstrap.Bootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import java.util.concurrent.TimeUnit; /** * <p> rpc 客户端 </p> * * <pre> Created: 2019/10/16 11:21 下午 </pre> * <pre> Project: rpc </pre> * * @author houbinbin * @since 0.0.2 */ public class RpcClient { private static final Log log = LogFactory.getLog(RpcClient.class); /** * 监听端口号 */ private final int port; /** * channel 信息 * @since 0.0.4 */ private ChannelFuture channelFuture; /** * 客户端处理 handler * @since 0.0.4 */ private RpcClientHandler channelHandler; public RpcClient(int port) { this.port = port; } public RpcClient() { this(RpcConstant.PORT); } /** * 开始运行 */ public void start() { // 启动服务端 log.info("RPC 服务开始启动客户端"); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); channelFuture = bootstrap.group(workerGroup) .channel(NioSocketChannel.class) .option(ChannelOption.SO_KEEPALIVE, true) .handler(new ChannelInitializer<Channel>(){ @Override protected void initChannel(Channel ch) throws Exception { channelHandler = new RpcClientHandler(); ch.pipeline() .addLast(new LoggingHandler(LogLevel.INFO)) .addLast(new CalculateRequestEncoder()) .addLast(new CalculateResponseDecoder()) .addLast(channelHandler); } }) .connect(RpcConstant.ADDRESS, port) .syncUninterruptibly(); log.info("RPC 服务启动客户端完成，监听端口：" + port); } catch (Exception e) { log.error("RPC 客户端遇到异常", e); throw new RuntimeException(e); } // 不要关闭线程池！！！ } /** * 调用计算 * @param request 请求信息 * @return 结果 * @since 0.0.4 */ public CalculateResponse calculate(final CalculateRequest request) { // 发送请求 final Channel channel = channelFuture.channel(); log.info("RPC 客户端发送请求，request: {}", request); // 关闭当前线程，以获取对应的信息 channel.writeAndFlush(request); channel.closeFuture().syncUninterruptibly(); return channelHandler.getResponse(); } }客户端处理类，编码解码和上次一样。
import com.github.houbb.log.integration.core.Log; import com.github.houbb.log.integration.core.LogFactory; import com.github.houbb.rpc.client.core.RpcClient; import com.github.houbb.rpc.common.model.CalculateResponse; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; /** * <p> 客户端处理类 </p> * * <pre> Created: 2019/10/16 11:30 下午 </pre> * <pre> Project: rpc </pre> * * @author houbinbin * @since 0.0.2 */ public class RpcClientHandler extends SimpleChannelInboundHandler { private static final Log log = LogFactory.getLog(RpcClient.class); /** * 响应信息 * @since 0.0.4 */ private CalculateResponse response; @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { CalculateResponse response = (CalculateResponse)msg; this.response = response; log.info("[Client] response is :{}", response); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { // 每次用完要关闭，不然拿不到response，我也不知道为啥（目测得了解netty才行） // 个人理解：如果不关闭，则永远会被阻塞。 ctx.flush(); ctx.close(); } public CalculateResponse getResponse() { return response; } }计算的代理实现。
import com.github.houbb.rpc.client.core.RpcClient; import com.github.houbb.rpc.common.model.CalculateRequest; import com.github.houbb.rpc.common.model.CalculateResponse; import com.github.houbb.rpc.common.service.Calculator; /** * @author binbin.hou * @since 0.0.4 */ public class CalculatorProxy implements Calculator { /** * rpc 客户端 */ private RpcClient rpcClient; /** * 创建类 * （1）默认初始化 client 端 */ public CalculatorProxy() { rpcClient = new RpcClient(); rpcClient.start(); } @Override public CalculateResponse sum(CalculateRequest request) { return rpcClient.calculate(request); } }和原来一样，此处不再赘述。
直接启动即可。
我们主动发起一次调用。
import com.github.houbb.rpc.client.proxy.CalculatorProxy; import com.github.houbb.rpc.common.model.CalculateRequest; import com.github.houbb.rpc.common.model.CalculateResponse; import com.github.houbb.rpc.common.service.Calculator; import org.junit.Ignore; import java.util.concurrent.TimeUnit; /** * rpc 客户端测试代码 * @author binbin.hou * @since 0.0.2 */ @Ignore public class RpcClientTest { /** * 服务启动代码测试 * @param args 参数 */ public static void main(String[] args) { Calculator calculator = new CalculatorProxy(); CalculateRequest request = new CalculateRequest(); request.setOne(5); request.setTwo(6); CalculateResponse response = calculator.sum(request); System.out.println("rpc call result: " + response); } } 日志 [DEBUG] [2019-11-01 14:48:33.523] [main] [c.g.h.l.i.c.LogFactory.setImplementation] - Logging initialized using 'class com.github.houbb.log.integration.adaptors.stdout.StdOutExImpl' adapter. [INFO] [2019-11-01 14:48:33.527] [main] [c.g.h.r.c.c.RpcClient.start] - RPC 服务开始启动客户端 [INFO] [2019-11-01 14:48:34.546] [main] [c.g.h.r.c.c.RpcClient.start] - RPC 服务启动客户端完成，监听端口：9527 [INFO] [2019-11-01 14:48:34.548] [main] [c.g.h.r.c.c.RpcClient.calculate] - RPC 客户端发送请求，request: CalculateRequest{one=5, two=6} [INFO] [2019-11-01 14:48:34.600] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelRead0] - [Client] response is :CalculateResponse{success=true, sum=11} rpc call result: CalculateResponse{success=true, sum=11}