服务对象调用

功能

实现 Client 端到服务端之间的固定服务 pojo 调用。

接口

服务

我们希望实现一个计算功能。

public interface Calculator { /** * 计算加法 * @param request 请求入参 * @return 返回结果 */ CalculateResponse sum(final CalculateRequest request); } 服务端实现 public class CalculatorService implements Calculator { @Override public CalculateResponse sum(CalculateRequest request) { int sum = request.getOne()+request.getTwo(); return new CalculateResponse(true, sum); } }

pojo 信息

入参和出参如下：

CalculateRequest.java public class CalculateRequest implements Serializable { private static final long serialVersionUID = 6420751004355300996L; /** * 参数一 */ private int one; /** * 参数二 */ private int two; //Getter & Setter } CalculateResponse.java public class CalculateResponse implements Serializable { private static final long serialVersionUID = -1972014736222511341L; /** * 是否成功 */ private boolean success; /** * 二者的和 */ private int sum; //Getter & Setter }

服务端核心代码

RpcServerHandler

主要处理客户端的调用请求。

import com.github.houbb.log.integration.core.Log; import com.github.houbb.log.integration.core.LogFactory; 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 com.github.houbb.rpc.server.service.CalculatorService; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; /** * @author binbin.hou * @since 0.0.1 */ public class RpcServerHandler extends SimpleChannelInboundHandler { private static final Log log = LogFactory.getLog(RpcServerHandler.class); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { final String id = ctx.channel().id().asLongText(); log.info("[Server] channel {} connected " + id); } @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { final String id = ctx.channel().id().asLongText(); CalculateRequest request = (CalculateRequest)msg; log.info("[Server] receive channel {} request: {} from ", id, request); Calculator calculator = new CalculatorService(); CalculateResponse response = calculator.sum(request); // 回写到 client 端 ctx.writeAndFlush(response); log.info("[Server] channel {} response {}", id, response); } }

编码器 & 解码器

可以使得我们 handler 中直接操作对象即可。

CalculateResponseEncoder.java import com.github.houbb.rpc.common.model.CalculateResponse; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; /** * @author binbin.hou * @since 0.0.3 */ public class CalculateResponseEncoder extends MessageToByteEncoder<CalculateResponse> { @Override protected void encode(ChannelHandlerContext ctx, CalculateResponse msg, ByteBuf out) throws Exception { boolean success = msg.isSuccess(); int result = msg.getSum(); out.writeBoolean(success); out.writeInt(result); } } CalculateRequestDecoder.java import com.github.houbb.rpc.common.model.CalculateRequest; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import java.util.List; /** * 请求参数解码 * @author binbin.hou * @since 0.0.3 */ public class CalculateRequestDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { int one = in.readInt(); int two = in.readInt(); CalculateRequest request = new CalculateRequest(one, two); out.add(request); } }

客户端核心代码

RpcClientHandler.java

向客户端发送请求，并且处理服务端响应结果。

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.CalculateRequest; 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); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { CalculateRequest request = new CalculateRequest(1, 2); ctx.writeAndFlush(request); log.info("[Client] request is :{}", request); } @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { CalculateResponse response = (CalculateResponse)msg; log.info("[Client] response is :{}", response); } }

编码器 & 解码器

CalculateRequestEncoder.java import com.github.houbb.rpc.common.model.CalculateRequest; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; /** * @author binbin.hou * @since 0.0.3 */ public class CalculateRequestEncoder extends MessageToByteEncoder<CalculateRequest> { @Override protected void encode(ChannelHandlerContext ctx, CalculateRequest msg, ByteBuf out) throws Exception { int one = msg.getOne(); int two = msg.getTwo(); out.writeInt(one); out.writeInt(two); } } CalculateResponseDecoder.java import com.github.houbb.rpc.common.model.CalculateResponse; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.ByteToMessageDecoder; import java.util.List; /** * 响应参数解码 * @author binbin.hou * @since 0.0.3 */ public class CalculateResponseDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { boolean success = in.readBoolean(); int sum = in.readInt(); CalculateResponse response = new CalculateResponse(success, sum); out.add(response); } }

测试代码

服务端启动

测试代码 new RpcServer().start(); 日志 [DEBUG] [2019-11-01 14:18:23.630] [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:18:23.634] [Thread-0] [c.g.h.r.s.c.RpcServer.run] - RPC 服务开始启动服务端 十一月 01, 2019 2:18:24 下午 io.netty.handler.logging.LoggingHandler channelRegistered 信息: [id: 0xbd1a8caf] REGISTERED 十一月 01, 2019 2:18:24 下午 io.netty.handler.logging.LoggingHandler bind 信息: [id: 0xbd1a8caf] BIND: 0.0.0.0/0.0.0.0:9527 十一月 01, 2019 2:18:24 下午 io.netty.handler.logging.LoggingHandler channelActive 信息: [id: 0xbd1a8caf, L:/0:0:0:0:0:0:0:0:9527] ACTIVE [INFO] [2019-11-01 14:18:24.708] [Thread-0] [c.g.h.r.s.c.RpcServer.run] - RPC 服务端启动完成，监听【9527】端口

客户端启动

测试代码 new RpcClient().start(); 日志 [INFO] [2019-11-01 14:19:04.621] [Thread-0] [c.g.h.r.c.c.RpcClient.run] - RPC 服务开始启动客户端 [INFO] [2019-11-01 14:19:05.627] [Thread-0] [c.g.h.r.c.c.RpcClient.run] - RPC 服务启动客户端完成，监听端口：9527 十一月 01, 2019 2:19:05 下午 io.netty.handler.logging.LoggingHandler channelRegistered 信息: [id: 0x250bb351] REGISTERED 十一月 01, 2019 2:19:05 下午 io.netty.handler.logging.LoggingHandler connect 信息: [id: 0x250bb351] CONNECT: /127.0.0.1:9527 十一月 01, 2019 2:19:05 下午 io.netty.handler.logging.LoggingHandler channelActive 信息: [id: 0x250bb351, L:/127.0.0.1:56605 - R:/127.0.0.1:9527] ACTIVE 十一月 01, 2019 2:19:05 下午 io.netty.handler.logging.LoggingHandler write 信息: [id: 0x250bb351, L:/127.0.0.1:56605 - R:/127.0.0.1:9527] WRITE: 8B +-------------------------------------------------+ | 0 1 2 3 4 5 6 7 8 9 a b c d e f | +--------+-------------------------------------------------+----------------+ |00000000| 00 00 00 01 00 00 00 02 |........ | +--------+-------------------------------------------------+----------------+ 十一月 01, 2019 2:19:05 下午 io.netty.handler.logging.LoggingHandler flush 信息: [id: 0x250bb351, L:/127.0.0.1:56605 - R:/127.0.0.1:9527] FLUSH [INFO] [2019-11-01 14:19:05.662] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelActive] - [Client] request is :CalculateRequest{one=1, two=2} [INFO] [2019-11-01 14:19:05.693] [nioEventLoopGroup-2-1] [c.g.h.r.c.c.RpcClient.channelRead0] - [Client] response is :CalculateResponse{success=true, sum=3}