public class RpcResponse<T> implements Serializable {
private static final long serialVersionUID = 715745410605631233L; private String requestId;//RPC响应的唯一标识符 /** * response code */ private Integer code; /** * response message */ private String message; /** * response body */ private T data;
public static <T> RpcResponse<T> success(T data, String requestId) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(RpcResponseCodeEnum.SUCCESS.getCode()); response.setMessage(RpcResponseCodeEnum.SUCCESS.getMessage()); response.setRequestId(requestId); if (null != data) { response.setData(data); } return response; }
public static <T> RpcResponse<T> fail(RpcResponseCodeEnum rpcResponseCodeEnum) { RpcResponse<T> response = new RpcResponse<>(); response.setCode(rpcResponseCodeEnum.getCode()); response.setMessage(rpcResponseCodeEnum.getMessage()); return response; }
}
success和fail都很简单,就是根据成功与否去设置消息的内容
Netty客户端
NettyRpcClient
@Slf4j public final class NettyRpcClient implements RpcRequestTransport { private final ServiceDiscovery serviceDiscovery;//服务发现的实例，用于查找服务的地址 private final UnprocessedRequests unprocessedRequests;//未处理的请求实例，用于存储未完成的RPC请求。 private final ChannelProvider channelProvider;//通道提供者的实例，用于管理和提供与服务器地址相关联的通道。 private final Bootstrap bootstrap;//Netty的Bootstrap实例，用于配置和初始化Netty客户端。 private final EventLoopGroup eventLoopGroup;//Netty的EventLoopGroup实例，用于处理事件循环。
public NettyRpcClient() { // initialize resources such as EventLoopGroup, Bootstrap eventLoopGroup = new NioEventLoopGroup(); bootstrap = new Bootstrap(); bootstrap.group(eventLoopGroup) .channel(NioSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) // The timeout period of the connection. // If this time is exceeded or the connection cannot be established, the connection fails. .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) { ChannelPipeline p = ch.pipeline(); // If no data is sent to the server within 15 seconds, a heartbeat request is sent心跳机制 p.addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS));//用于处理连接的空闲状态 p.addLast(new RpcMessageEncoder()); p.addLast(new RpcMessageDecoder()); p.addLast(new NettyRpcClientHandler()); } }); this.serviceDiscovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension(ServiceDiscoveryEnum.ZK.getName()); this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class); this.channelProvider = SingletonFactory.getInstance(ChannelProvider.class); }
/** * connect server and get the channel ,so that you can send rpc message to server * * @param inetSocketAddress server address * @return the channel */ @SneakyThrows public Channel doConnect(InetSocketAddress inetSocketAddress) { CompletableFuture<Channel> completableFuture = new CompletableFuture<>(); bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("The client has connected [{}] successful!", inetSocketAddress.toString()); completableFuture.complete(future.channel()); } else { throw new IllegalStateException(); } }); return completableFuture.get(); }
@Override public Object sendRpcRequest(RpcRequest rpcRequest) { // build return value CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();//这个异步操作的结果是一个包含泛型类型为 Object 的 RpcResponse 对象。 // get server address InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest);//使用 serviceDiscovery（服务发现）查找并获取RPC请求的服务地址。 // get server address related channel Channel channel = getChannel(inetSocketAddress);//通过 getChannel 方法获取与指定服务器地址关联的 Channel。 if (channel.isActive()) { // put unprocessed request unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);//将 RPC 请求放入未处理请求的集合中，以便后续处理响应。 RpcMessage rpcMessage = RpcMessage.builder().data(rpcRequest) .codec(SerializationTypeEnum.HESSIAN.getCode()) .compress(CompressTypeEnum.GZIP.getCode()) .messageType(RpcConstants.REQUEST_TYPE).build(); channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> { if (future.isSuccess()) { log.info("client send message: [{}]", rpcMessage); } else { future.channel().close(); resultFuture.completeExceptionally(future.cause()); log.error("Send failed:", future.cause()); } }); } else { throw new IllegalStateException(); }
public class UnprocessedRequests { public static final Map<String, CompletableFuture<RpcResponse<Object>>> UNPROCESSED_RESPONSE_FUTURES=new ConcurrentHashMap<>(); public void put(String requestId, CompletableFuture<RpcResponse<Object>> future) { UNPROCESSED_RESPONSE_FUTURES.put(requestId, future); }
public void complete(RpcResponse<Object> rpcResponse) { CompletableFuture<RpcResponse<Object>> future = UNPROCESSED_RESPONSE_FUTURES.remove(rpcResponse.getRequestId()); if(null!=future) { future.complete(rpcResponse);//在这个时候，future里面才存入了rpcResponse }else { throw new IllegalArgumentException(); } } }
private final Map<String, Channel> channelMap;//使用 ConcurrentHashMap 作为存储通道映射关系的容器。ConcurrentHashMap 是线程安全的，适用于并发环境。
public ChannelProvider() { channelMap = new ConcurrentHashMap<>(); }
public Channel get(InetSocketAddress inetSocketAddress) { String key = inetSocketAddress.toString(); // determine if there is a connection for the corresponding address if (channelMap.containsKey(key)) { Channel channel = channelMap.get(key); // if so, determine if the connection is available, and if so, get it directly if (channel != null && channel.isActive()) { return channel; } else { channelMap.remove(key); } } return null; }
get：获得channel
set:设置
public void set(InetSocketAddress inetSocketAddress, Channel channel){ String key=inetSocketAddress.toString(); channelMap.put(key,channel); }
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { if(msg instanceof RpcMessage) { log.info("server receive msg: [{}] ", msg); byte messageType = ((RpcMessage) msg).getMessageType(); RpcMessage rpcMessage = new RpcMessage(); rpcMessage.setCodec(SerializationTypeEnum.HESSIAN.getCode()); rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode()); //如果是心跳请求，那么就返回个pong回去 if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) { rpcMessage.setMessageType(RpcConstants.HEARTBEAT_RESPONSE_TYPE); rpcMessage.setData(RpcConstants.PONG); }else {//否则就是有内容的了 RpcRequest rpcRequest = (RpcRequest) ((RpcMessage) msg).getData(); // Execute the target method (the method the client needs to execute) and return the method result Object result = rpcRequestHandler.handle(rpcRequest);//执行请求，获得结果 log.info(String.format("server get result: %s", result.toString())); rpcMessage.setMessageType(RpcConstants.RESPONSE_TYPE); //把结果写到response里面 if(ctx.channel().isActive()&&ctx.channel().isWritable()) { RpcResponse<Object> rpcResponse = RpcResponse.success(result, rpcRequest.getRequestId()); rpcMessage.setData(rpcResponse); }else//否则失败了，把失败的结果写到里面 { RpcResponse<Object> rpcResponse = RpcResponse.fail(RpcResponseCodeEnum.FAIL); rpcMessage.setData(rpcResponse); log.error("not writable now, message dropped"); } } ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);//把结果写回给客户端 } }finally { ReferenceCountUtil.release(msg); } }
还有两个比较简单，不分析了
RpcRequestHandler
这个在handler包下，用于真正去调用方法，并不是pipeline里面的handler
@Slf4j public class RpcRequestHandler { private final ServiceProvider serviceProvider;
public RpcRequestHandler() { serviceProvider = SingletonFactory.getInstance(ZkServiceProviderImpl.class); }
/** * Processing rpcRequest: call the corresponding method, and then return the method */ public Object handle(RpcRequest rpcRequest) { Object service = serviceProvider.getService(rpcRequest.getRpcServiceName()); return invokeTargetMethod(rpcRequest, service); }
/** * get method execution results * * @param rpcRequest client request * @param service service object * @return the result of the target method execution */ private Object invokeTargetMethod(RpcRequest rpcRequest, Object service) { Object result; try { Method method = service.getClass().getMethod(rpcRequest.getMethodName(), rpcRequest.getParamTypes()); result = method.invoke(service, rpcRequest.getParameters()); log.info("service:[{}] successful invoke method:[{}]", rpcRequest.getInterfaceName(), rpcRequest.getMethodName()); } catch (NoSuchMethodException | IllegalArgumentException | InvocationTargetException | IllegalAccessException e) { throw new RpcException(e.getMessage(), e); } return result; } }
public Object handle(RpcRequest rpcRequest) { Object service = serviceProvider.getService(rpcRequest.getRpcServiceName()); return invokeTargetMethod(rpcRequest, service); }
先是获取了service对象
service其实就是一个Object：
public class RpcServiceConfig { /** * service version */ private String version = ""; /** * when the interface has multiple implementation classes, distinguish by group */ private String group = "";