From 99ae6a50be906e7bde06efb99c2dcb3b1871f294 Mon Sep 17 00:00:00 2001 From: Vincent Date: Mon, 25 May 2026 18:53:48 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=86=85=E5=AD=98?= =?UTF-8?q?=E6=B3=84=E6=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../edp/caf/rpc/client/CAFRpcClientImpl.java | 581 +++++++----------- 1 file changed, 230 insertions(+), 351 deletions(-) diff --git a/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java b/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java index f122e2074..1f5385ddb 100644 --- a/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java +++ b/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java @@ -19,7 +19,6 @@ package io.iec.edp.caf.rpc.client; import com.google.protobuf.ByteString; import io.grpc.Metadata; import io.grpc.StatusRuntimeException; -//import io.iec.edp.caf.boot.context.CAFContext; import io.iec.edp.caf.common.JSONSerializer; import io.iec.edp.caf.commons.core.SerializerFactory; import io.iec.edp.caf.commons.core.enums.SerializeType; @@ -29,7 +28,6 @@ import io.iec.edp.caf.commons.exception.entity.DefaultExceptionProperties; import io.iec.edp.caf.commons.exception.entity.ExceptionErrorCode; import io.iec.edp.caf.commons.runtime.CafEnvironment; import io.iec.edp.caf.commons.runtime.thread.CAFThreadPoolExecutor; -//import io.iec.edp.caf.core.session.ICafSessionService; import io.iec.edp.caf.logging.CommonConstant; import io.iec.edp.caf.msu.client.exception.ServiceUnitNotFoundException; import io.iec.edp.caf.rpc.api.channel.RpcAbstractChannel; @@ -48,12 +46,6 @@ import io.iec.edp.caf.rpc.api.utils.Utils; import io.iec.edp.caf.rpc.api.utils.Validator; import io.iec.edp.caf.rpc.client.local.RpcLocalChecker; import io.iec.edp.caf.rpc.client.local.RpcLocalInvoker; -//import io.iec.edp.caf.tenancy.api.ITenantRouteService; -//import io.iec.edp.caf.tenancy.api.ITenantService; -//import io.iec.edp.caf.tenancy.api.context.RequestTenantContextHolder; -//import io.iec.edp.caf.tenancy.api.context.RequestTenantContextInfo; -//import io.iec.edp.caf.tenancy.api.entity.TenancyMode; -//import io.iec.edp.caf.tenancy.api.exception.TenantNotFoundException; import io.iec.edp.caf.rpc.api.spi.RpcContextResolver; import io.iec.edp.caf.rpc.api.spi.RpcContextTransmittee; import lombok.SneakyThrows; @@ -82,8 +74,6 @@ public class CAFRpcClientImpl implements CAFRpcClient { private final InternalServiceManageService management; private final RpcLocalInvoker rpcLocalInvoker; -// private final ITenantService tenantService; -// private final ITenantRouteService tenantRouteService; private String cookieName; private final RpcClientEventBroker clientEventBroker; private static final String HTTP_THREAD_NAME="http-nio-"; @@ -105,17 +95,11 @@ public class CAFRpcClientImpl implements CAFRpcClient { public CAFRpcClientImpl(RpcClientEventBroker clientEventBroker, InternalServiceManageService management, - RpcLocalInvoker rpcLocalInvoker -// ITenantService tenantService, -// ITenantRouteService tenantRouteService - ) { - + RpcLocalInvoker rpcLocalInvoker) { this.clientEventBroker = clientEventBroker; this.management = management; this.rpcLocalInvoker = rpcLocalInvoker; this.cookieName = CafEnvironment.getEnvironment().getProperty("caf-security.general.cookie-name"); -// this.tenantService = tenantService; -// this.tenantRouteService = tenantRouteService; if (this.cookieName == null) { this.cookieName = ConstanceVarible.COOKIE_SESSION_NAME; } @@ -143,7 +127,6 @@ public class CAFRpcClientImpl implements CAFRpcClient { return this.invoke(String.class, serviceId, version, serviceUnitName, parameters, channelType,filterType,context); } - @SneakyThrows @Override public T invoke(Type t, String serviceId,String version, String serviceUnitName, LinkedHashMap parameters, RpcChannelType channelType, String filterType,HashMap context) throws IllegalAccessException, InstantiationException, IOException, NoSuchMethodException, InvocationTargetException, ClassNotFoundException { @@ -155,13 +138,6 @@ public class CAFRpcClientImpl implements CAFRpcClient { filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT; } - //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现 -// if(channelType!=null&&channelType.equals(RpcChannelType.REST)){ -// RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class); -// assert rpcClient != null; -// return rpcClient.invoke(t,serviceId,serviceUnitName,parameters,context); -// } - try { var start = System.currentTimeMillis(); var end = System.currentTimeMillis(); @@ -171,7 +147,6 @@ public class CAFRpcClientImpl implements CAFRpcClient { //check cookie only log warning not break invoke Validator.CheckCksSession(cks,serviceId); - T result; //get rpc service definition from local memory variable RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId); //get extend tenant id @@ -186,6 +161,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { log.info("start rpc invoke，current serviceId：{}，current target su:{}",serviceId,serviceUnitName); Object retvalue = null; + T result; //is local invoke if (RpcLocalChecker.isLocalInvoke(serviceUnitName)) { //local memory has no rpc service definition @@ -213,59 +189,60 @@ public class CAFRpcClientImpl implements CAFRpcClient { retvalue = result; } else { - channelType = channelType==null?RpcChannelType.GRPC:channelType; - filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; - //create rpc channel by protocol - RpcChannelOptions options = RpcChannelOptions.builder() - .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) - .build(); - var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); - assert channel != null; - if(log.isInfoEnabled()){ - log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); - } - - //prepare event args - eventContext.put(ConstanceVarible.IS_LOCAL, "false"); - eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); - eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); - HashMap localDict = new HashMap<>(); - localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); - //before invoke event - this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); - String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); - - //set gsp context to header - HashMap gspContext = new HashMap<>(); - setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); - - //add additional headers to channel - Map headers = new HashMap<>(); - headers.put(CommonConstant.TRACE_ID_HEADER,traceId); - headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); - channel.addHeaders(headers); - - //in RPC2.0 use the protobuf serialize as default - LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); -// LinkedHashMap params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition); - - start = System.currentTimeMillis(); - var returnString = channel.invokeRemoteService(serviceId,version,params,null); - channel.stopClient(); -// retvalue = RpcSerializeUtil.deSerializeReturnValue(t, returnString ,serviceMethodDefinition); - retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString ,serviceMethodDefinition); - - var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); - var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); - serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; - this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); - - end = System.currentTimeMillis(); + RpcAbstractChannel channel = null; + try { + channelType = channelType==null?RpcChannelType.GRPC:channelType; + filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; + //create rpc channel by protocol + RpcChannelOptions options = RpcChannelOptions.builder() + .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) + .build(); + channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); + assert channel != null; + if(log.isInfoEnabled()){ + log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); + } - if(log.isInfoEnabled()){ - log.info("rpc remote invoke client total time:{} ms",end-start); + //prepare event args + eventContext.put(ConstanceVarible.IS_LOCAL, "false"); + eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); + eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); + HashMap localDict = new HashMap<>(); + localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); + //before invoke event + this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); + String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); + + //set gsp context to header + HashMap gspContext = new HashMap<>(); + setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); + + //add additional headers to channel + Map headers = new HashMap<>(); + headers.put(CommonConstant.TRACE_ID_HEADER,traceId); + headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); + channel.addHeaders(headers); + + //in RPC2.0 use the protobuf serialize as default + LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); + + start = System.currentTimeMillis(); + var returnString = channel.invokeRemoteService(serviceId,version,params,null); + retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString ,serviceMethodDefinition); + + var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); + var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); + serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; + this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); + + end = System.currentTimeMillis(); + + if(log.isInfoEnabled()){ + log.info("rpc remote invoke client total time:{} ms",end-start); + } + } finally { + if (channel!=null) channel.stopClient(); } - } return (T)retvalue; @@ -290,15 +267,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT; } - //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现 -// if(channelType!=null&&channelType.equals(RpcChannelType.REST)){ -// RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class); -// assert rpcClient != null; -// return rpcClient.invokeStream(serviceId,version,serviceUnitName,parameters,context); -// } - try { - var start = System.currentTimeMillis(); var end = System.currentTimeMillis(); @@ -345,57 +314,59 @@ public class CAFRpcClientImpl implements CAFRpcClient { end = System.currentTimeMillis(); log.info("rpc local invoke client total time：{} ms",end-start); } else { - channelType = channelType==null?RpcChannelType.GRPC:channelType; - filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; - //create rpc channel by protocol - RpcChannelOptions options = RpcChannelOptions.builder() - .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) - .build(); - var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); - assert channel != null; - if(log.isInfoEnabled()){ - log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); - } - - //prepare event args - eventContext.put(ConstanceVarible.IS_LOCAL, "false"); - eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); - eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); - HashMap localDict = new HashMap<>(); - localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); - //before invoke event - this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); - String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); - - //prepare rpc context - HashMap gspContext = new HashMap<>(); - setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); - - //add additional headers to channel - Map headers = new HashMap<>(); - headers.put(CommonConstant.TRACE_ID_HEADER,traceId); - headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); - channel.addHeaders(headers); - - //in RPC2.0 use the protobuf serialize as default - LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); -// LinkedHashMap params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition); - - start = System.currentTimeMillis(); - retvalue = channel.invokeRemoteServiceStream(serviceId,version,params,null); - channel.stopClient(); - - var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); - var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); - serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; - this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); - - end = System.currentTimeMillis(); + RpcAbstractChannel channel = null; + try { + channelType = channelType==null?RpcChannelType.GRPC:channelType; + filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; + //create rpc channel by protocol + RpcChannelOptions options = RpcChannelOptions.builder() + .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) + .build(); + channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); + assert channel != null; + if(log.isInfoEnabled()){ + log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); + } - if(log.isInfoEnabled()){ - log.info("rpc remote invoke client total time:{} ms",end-start); + //prepare event args + eventContext.put(ConstanceVarible.IS_LOCAL, "false"); + eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); + eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); + HashMap localDict = new HashMap<>(); + localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); + //before invoke event + this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); + String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); + + //prepare rpc context + HashMap gspContext = new HashMap<>(); + setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); + + //add additional headers to channel + Map headers = new HashMap<>(); + headers.put(CommonConstant.TRACE_ID_HEADER,traceId); + headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); + channel.addHeaders(headers); + + //in RPC2.0 use the protobuf serialize as default + LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); + + start = System.currentTimeMillis(); + retvalue = channel.invokeRemoteServiceStream(serviceId,version,params,null); + + var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); + var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); + serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; + this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); + + end = System.currentTimeMillis(); + + if(log.isInfoEnabled()){ + log.info("rpc remote invoke client total time:{} ms",end-start); + } + } finally { + if (channel!=null) channel.stopClient(); } - } return retvalue; @@ -419,15 +390,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT; } - //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现 -// if(channelType!=null&&channelType.equals(RpcChannelType.REST)){ -// RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class); -// assert rpcClient != null; -// return rpcClient.invokeStream(serviceId,version,serviceUnitName,parameters,context); -// } - try { - var start = System.currentTimeMillis(); var end = System.currentTimeMillis(); @@ -479,60 +442,61 @@ public class CAFRpcClientImpl implements CAFRpcClient { end = System.currentTimeMillis(); log.info("rpc local invoke client total time：{} ms",end-start); } else { - channelType = channelType==null?RpcChannelType.GRPC:channelType; - filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; - //create rpc channel by protocol - RpcChannelOptions options = RpcChannelOptions.builder() - .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) - .build(); - var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); - assert channel != null; - - if(log.isInfoEnabled()){ - log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); - } - - //prepare event args - eventContext.put(ConstanceVarible.IS_LOCAL, "false"); - eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); - eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); - HashMap localDict = new HashMap<>(); - localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); - //before invoke event - this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); - String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); - - //prepare rpc context - HashMap gspContext = new HashMap<>(); - setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); - - //add additional headers to channel - Map headers = new HashMap<>(); - headers.put(CommonConstant.TRACE_ID_HEADER,traceId); - headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); - channel.addHeaders(headers); - - //in RPC2.0 use the protobuf serialize as default - LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); -// LinkedHashMap params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition); - - start = System.currentTimeMillis(); - channel.invokeRemoteServiceStreamAsyn(serviceId,version,params,null,callBack); - channel.stopClient(); - - var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); - var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); - serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; - this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, null, localDict); - - end = System.currentTimeMillis(); + RpcAbstractChannel channel = null; + try { + channelType = channelType==null?RpcChannelType.GRPC:channelType; + filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; + //create rpc channel by protocol + RpcChannelOptions options = RpcChannelOptions.builder() + .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) + .build(); + channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); + assert channel != null; + + if(log.isInfoEnabled()){ + log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); + } - if(log.isInfoEnabled()){ - log.info("rpc remote invoke client total time:{} ms",end-start); + //prepare event args + eventContext.put(ConstanceVarible.IS_LOCAL, "false"); + eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); + eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); + HashMap localDict = new HashMap<>(); + localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); + //before invoke event + this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); + String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); + + //prepare rpc context + HashMap gspContext = new HashMap<>(); + setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); + + //add additional headers to channel + Map headers = new HashMap<>(); + headers.put(CommonConstant.TRACE_ID_HEADER,traceId); + headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); + channel.addHeaders(headers); + + //in RPC2.0 use the protobuf serialize as default + LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); + + start = System.currentTimeMillis(); + channel.invokeRemoteServiceStreamAsyn(serviceId,version,params,null,callBack); + + var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); + var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); + serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; + this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, null, localDict); + + end = System.currentTimeMillis(); + + if(log.isInfoEnabled()){ + log.info("rpc remote invoke client total time:{} ms",end-start); + } + } finally { + if (channel!=null) channel.stopClient(); } - } - } catch (Exception e) { log.debug("rpc client exception type:{}",e.getClass().getName()); var ex = handleException(e); @@ -553,15 +517,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { filterType = ConstanceVarible.RPC_FILTER_MODE_DEFAULT; } - //兼容RPC1.0接口，后续可以基于RPC2.0的REST接口实现 -// if(channelType!=null&&channelType.equals(RpcChannelType.REST)){ -// RpcClient rpcClient = SpringBeanUtils.getBean(RpcClient.class); -// assert rpcClient != null; -// return rpcClient.invokeStream(stream,t,serviceId,version,serviceUnitName,parameters,context); -// } - try { - var start = System.currentTimeMillis(); var end = System.currentTimeMillis(); @@ -570,7 +526,6 @@ public class CAFRpcClientImpl implements CAFRpcClient { //check cookie only log warning not break invoke Validator.CheckCksSession(cks,serviceId); -// T result; //get rpc service definition from local memory variable RpcServiceMethodDefinition serviceMethodDefinition = this.management.getRpcMethodDefinition(serviceId); //get extend tenant id @@ -611,63 +566,64 @@ public class CAFRpcClientImpl implements CAFRpcClient { log.info("rpc local invoke client total time：{} ms",end-start); } else { - channelType = channelType==null?RpcChannelType.GRPC:channelType; - filterType = filterType==null? ConstanceVarible.RPC_FILTER_MODE_DEFAULT :filterType; - //create rpc channel by protocol - RpcChannelOptions options = RpcChannelOptions.builder() - .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) - .build(); - var channel = (RpcAbstractChannel)RpcChannelFactory.buildChannelClient(serviceUnitName,channelType,filterType,eventContext,options); - assert channel != null; - - if(log.isInfoEnabled()){ - log.info("rpc remote address：{}",channel.getHost()+":"+channel.getPort()); - } - - //prepare event args - eventContext.put(ConstanceVarible.IS_LOCAL, "false"); - eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL,channel.getHost()+":"+channel.getPort()); - eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); - HashMap localDict = new HashMap<>(); - localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); - //before invoke event - this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); - String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); - - //prepare rpc context - HashMap gspContext = new HashMap<>(); - setContext(gspContext,cks,serviceUnitName,targetTenantId,eventContext); - - //add additional headers to channel - Map headers = new HashMap<>(); - headers.put(CommonConstant.TRACE_ID_HEADER,traceId); - headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); - channel.addHeaders(headers); - - //here parse the inputstream parameter in parameters - var stream = Utils.handleStreamFromParameters(parameters); - //in RPC2.0 use the protobuf serialize as default - LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); -// LinkedHashMap params = RpcSerializeUtil.serializeParameter(parameters, serviceMethodDefinition); - - start = System.currentTimeMillis(); - - var returnString = channel.invokeRemoteServiceStream(stream,serviceId,version,params,null); - channel.stopClient(); -// retvalue = RpcSerializeUtil.deSerializeReturnValue(t, returnString ,serviceMethodDefinition); - retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString ,serviceMethodDefinition); - - var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); - var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); - serverContextDict = serverContextDict==null?new LinkedHashMap():serverContextDict; - this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); - - end = System.currentTimeMillis(); + RpcAbstractChannel channel = null; + try { + channelType = channelType == null ? RpcChannelType.GRPC : channelType; + filterType = filterType == null ? ConstanceVarible.RPC_FILTER_MODE_DEFAULT : filterType; + //create rpc channel by protocol + RpcChannelOptions options = RpcChannelOptions.builder() + .serviceDiscoveryRetryPolicy((ServiceDiscoveryRetryPolicy) RpcThreadCacheHolder.getObject(RPC_SERVICEDISCOVER_RETRY_POLICY)) + .build(); + channel = (RpcAbstractChannel) RpcChannelFactory.buildChannelClient(serviceUnitName, channelType, filterType, eventContext, options); + assert channel != null; + + if (log.isInfoEnabled()) { + log.info("rpc remote address：{}", channel.getHost() + ":" + channel.getPort()); + } - if(log.isInfoEnabled()){ - log.info("rpc remote invoke client total time:{} ms",end-start); + //prepare event args + eventContext.put(ConstanceVarible.IS_LOCAL, "false"); + eventContext.put(ConstanceVarible.CURRENT_REMOTEBASE_URL, channel.getHost() + ":" + channel.getPort()); + eventContext.put(ConstanceVarible.CURRENT_SERVICE_INTERFACE, serviceMethodDefinition == null ? null : serviceMethodDefinition.getParentDefinition().getClassName()); + HashMap localDict = new HashMap<>(); + localDict.put(CommonConstant.LOG_MSU_ID, serviceUnitName); + //before invoke event + this.clientEventBroker.firePreRpcInvokeEvent(eventContext, parameters, localDict); + String traceId = (String) localDict.get(CommonConstant.LOG_TRACE_ID); + + //prepare rpc context + HashMap gspContext = new HashMap<>(); + setContext(gspContext, cks, serviceUnitName, targetTenantId, eventContext); + + //add additional headers to channel + Map headers = new HashMap<>(); + headers.put(CommonConstant.TRACE_ID_HEADER, traceId); + headers.put(ConstanceVarible.GSP_CONTEXT, JSONSerializer.serialize(gspContext)); + channel.addHeaders(headers); + + //here parse the inputstream parameter in parameters + var stream = Utils.handleStreamFromParameters(parameters); + //in RPC2.0 use the protobuf serialize as default + LinkedHashMap params = RpcSerializeUtil.serializeParameter4RPC2(parameters, serviceMethodDefinition); + + start = System.currentTimeMillis(); + + var returnString = channel.invokeRemoteServiceStream(stream, serviceId, version, params, null); + retvalue = RpcSerializeUtil.deSerializeReturnValue4RPC2(t, returnString, serviceMethodDefinition); + + var serverContextString = RpcThreadCacheHolder.getValue(ConstanceVarible.GSP_RPC_SERVER_ENVENT); + var serverContextDict = JSONSerializer.deserialize(serverContextString, LinkedHashMap.class); + serverContextDict = serverContextDict == null ? new LinkedHashMap() : serverContextDict; + this.clientEventBroker.firePostRpcInvokeEvent(serverContextDict, retvalue, localDict); + + end = System.currentTimeMillis(); + + if (log.isInfoEnabled()) { + log.info("rpc remote invoke client total time:{} ms", end - start); + } + } finally { + if (channel != null) channel.stopClient(); } - } return (T)retvalue; @@ -760,117 +716,40 @@ public class CAFRpcClientImpl implements CAFRpcClient { } - -// /** -// * get extend tenantid to route -// * by implement {@link ITenantRouteService} interface -// * @param msu msu code -// * @param context extend parameters to choose route tenant -// * @return if {@link ITenantRouteService} implement it will reture the ITenantRouteService`s tenantid,if not implement -// * {@link ITenantRouteService} it will found tenantid in http header by CAFContextFilter,if both not found it -// * will return null -// */ -// private Integer getTargetTenantId(String msu, HashMap context) { -// -// if (this.tenantService.getTenancyMode() != TenancyMode.group) -// return null; -// -// context = context == null ? new HashMap<>() : context; -// String tenantIdString = context.get(ConstanceVarible.ROUTE_TENANTID); -// if (tenantIdString != null) { -// return Integer.valueOf(tenantIdString); -// } else { -// String tenantDim1 = context.get(ConstanceVarible.ROUTE_FIRSTDIMENSION_KEY); -// String tenantDim2 = context.get(ConstanceVarible.ROUTE_SECONDDIMENSION_KEY); -// if (!StringUtils.isEmpty(tenantDim1)) { -// try { -// return tenantRouteService.route(msu, tenantDim1, tenantDim2); -// } catch (TenantNotFoundException e) { -// log.error(String.format("tenant route failed su：%s first dimension：%s ,second dimension：%s", msu, tenantDim1, tenantDim2)); -// } -// } -// -// //get requset header tenant if set -// RequestTenantContextInfo contextInfo = RequestTenantContextHolder.get(); -// return contextInfo == null ? null : contextInfo.getTenantId(); -// } -// } - /** * https://javakk.com/2558.html * use metadata to deal with exception */ @SneakyThrows - private Throwable handleException(Throwable e){ - - if(e instanceof ServiceUnitNotFoundException){ + private Throwable handleException(Throwable e) { + if (e instanceof ServiceUnitNotFoundException) { return e; - }else if(e instanceof StatusRuntimeException){ - Metadata trailers = ((StatusRuntimeException)e).getTrailers(); + } else if (e instanceof StatusRuntimeException) { + Metadata trailers = ((StatusRuntimeException) e).getTrailers(); var serializer = SerializerFactory.getSerializer(SerializeType.Protobuf); String code = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER)); - int level = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER))==null? - 2:serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)),Integer.class); + int level = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)) == null ? + 2 : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)), Integer.class); String message = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)); String innerMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER)); - OffsetDateTime date =trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER))==null? - OffsetDateTime.now():serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)),OffsetDateTime.class); - String detail = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER))==null? - "":serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER)),String.class); + OffsetDateTime date = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)) == null ? + OffsetDateTime.now() : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)), OffsetDateTime.class); + String detail = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER)) == null ? + "" : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER)), String.class); String requestId = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER)); String exceptionType = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER)); - boolean isBizException=trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER))==null? - false:serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER)),Boolean.class);; - HashMap extensionMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))==null? - new HashMap<>():serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)),HashMap.class); -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// code = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// level = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// message = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// innerMessage = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)); -// if(temp!=null){ -// date = serializer.deserialize(temp,OffsetDateTime.class); -// } -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// detail = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// requestId = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// exceptionType = ; -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER))!=null){ -// var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER)); -// if(temp!=null){ -// isBizException = serializer.deserialize(temp,Boolean.class); -// } -// } -// if(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER))){ -// var temp = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)); -// if(temp!=null){ -// extensionMessage = serializer.deserialize(temp,HashMap.class); -// } -// } - + boolean isBizException = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER)) == null ? + false : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER)), Boolean.class); + HashMap extensionMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)) == null ? + new HashMap<>() : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)), HashMap.class); if (exceptionType != null && exceptionType.equalsIgnoreCase(CAFRuntimeException.class.getTypeName())) { - return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, code, message + "\n" + detail, null, ExceptionLevel.getLevel(level), isBizException); + return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, code, message + detail, null, ExceptionLevel.getLevel(level), isBizException); } else { return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, - ExceptionErrorCode.rpcError, message + "\n" + detail, null, ExceptionLevel.Error, false); + ExceptionErrorCode.rpcError, message + detail, null, ExceptionLevel.Error, false); } - }else{ + } else { return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, DefaultExceptionProperties.RESOURCE_FILE, ExceptionErrorCode.rpcError, -- Gitee From 4ac77d196da48ffc1b02d48542272ffeb2cd3a66 Mon Sep 17 00:00:00 2001 From: Vincent Date: Thu, 28 May 2026 09:00:47 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E4=BC=98=E5=8C=96grpc=E6=8A=9B=E5=87=BA?= =?UTF-8?q?=E5=BC=82=E5=B8=B8=E6=97=B6=E7=9A=84=E5=A4=84=E7=90=86;?= =?UTF-8?q?=E5=A2=9E=E5=A4=A7=E9=BB=98=E8=AE=A4header=E5=A4=A7=E5=B0=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rpc/api/grpc/GrpcExceptionMetaData.java | 44 +++++++++---------- .../edp/caf/rpc/client/CAFRpcClientImpl.java | 33 +++++++------- .../rpc/remote/grpc/GrpcChannelBuilder.java | 7 +-- .../caf/rpc/remote/grpc/GrpcServerInvoke.java | 35 ++++++--------- 4 files changed, 56 insertions(+), 63 deletions(-) diff --git a/caf-rpc/caf-rpc-api/src/main/java/io/iec/edp/caf/rpc/api/grpc/GrpcExceptionMetaData.java b/caf-rpc/caf-rpc-api/src/main/java/io/iec/edp/caf/rpc/api/grpc/GrpcExceptionMetaData.java index 20a382fd2..f82120e2c 100644 --- a/caf-rpc/caf-rpc-api/src/main/java/io/iec/edp/caf/rpc/api/grpc/GrpcExceptionMetaData.java +++ b/caf-rpc/caf-rpc-api/src/main/java/io/iec/edp/caf/rpc/api/grpc/GrpcExceptionMetaData.java @@ -23,7 +23,6 @@ import lombok.Builder; import lombok.var; import java.time.OffsetDateTime; -import java.util.Base64; import java.util.HashMap; @Builder @@ -36,42 +35,39 @@ public class GrpcExceptionMetaData { private String detail; private String requestId; private String exceptionType; - private Boolean isBizException=false; + private Boolean isBizException = false; private HashMap extensionMessage; - public Metadata buildMetaData(){ + public Metadata buildMetaData() { var metadata = new Metadata(); var serializer = SerializerFactory.getSerializer(SerializeType.Protobuf); - if(code!=null && code.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER),code); + if (code != null && code.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER), code); } - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER),serializer.serializeToString(level)); - if(message!=null && message.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER),message); + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(level)); + if (message != null && message.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(message)); } - if(innerMessage!=null && innerMessage.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER),innerMessage); + if (innerMessage != null && innerMessage.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(innerMessage)); } - if(date!=null){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER),serializer.serializeToString(date)); + if (date != null) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(date)); } - if(detail!=null && detail.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER),serializer.serializeToString(detail)); + if (detail != null && detail.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(detail)); } - if(requestId!=null && requestId.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER),requestId); + if (requestId != null && requestId.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_REQUESTID, Metadata.ASCII_STRING_MARSHALLER), requestId); } - if(extensionMessage!=null && extensionMessage.size()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER), - serializer.serializeToString(extensionMessage)); + if (extensionMessage != null && extensionMessage.size() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(extensionMessage)); } - if(exceptionType!=null && exceptionType.length()>0){ - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER), - exceptionType); + if (exceptionType != null && exceptionType.length() > 0) { + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_TYPE, Metadata.ASCII_STRING_MARSHALLER), exceptionType); } - metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER), - serializer.serializeToString(isBizException)); + metadata.put(Metadata.Key.of(GrpcVariable.GRPC_EX_BIZEXCEPTION, Metadata.ASCII_STRING_MARSHALLER), serializer.serializeToString(isBizException)); return metadata; } diff --git a/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java b/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java index 1f5385ddb..7415b301b 100644 --- a/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java +++ b/caf-rpc/caf-rpc-client/src/main/java/io/iec/edp/caf/rpc/client/CAFRpcClientImpl.java @@ -248,8 +248,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { return (T)retvalue; } catch (Exception e) { log.debug("rpc client exception type:{}",e.getClass().getName()); - var ex = handleException(e); - throw ex; + throw handleException(serviceId, serviceUnitName, e); }finally { RpcThreadCacheHolder.clear(); } @@ -372,8 +371,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { return retvalue; } catch (Exception e) { log.debug("rpc client exception type:{}",e.getClass().getName()); - var ex = handleException(e); - throw ex; + throw handleException(serviceId, serviceUnitName, e); }finally { RpcThreadCacheHolder.clear(); } @@ -499,8 +497,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { } } catch (Exception e) { log.debug("rpc client exception type:{}",e.getClass().getName()); - var ex = handleException(e); - throw ex; + throw handleException(serviceId, serviceUnitName, e); }finally { RpcThreadCacheHolder.clear(); } @@ -628,9 +625,8 @@ public class CAFRpcClientImpl implements CAFRpcClient { return (T)retvalue; } catch (Exception e) { - log.debug("rpc client exception type:{}",e.getClass().getName()); - var ex = handleException(e); - throw ex; + log.debug("rpc client exception type:{}", e.getClass().getName()); + throw handleException(serviceId, serviceUnitName, e); }finally { RpcThreadCacheHolder.clear(); } @@ -721,7 +717,7 @@ public class CAFRpcClientImpl implements CAFRpcClient { * use metadata to deal with exception */ @SneakyThrows - private Throwable handleException(Throwable e) { + private Throwable handleException(String serviceId, String su, Throwable e) { if (e instanceof ServiceUnitNotFoundException) { return e; } else if (e instanceof StatusRuntimeException) { @@ -730,8 +726,10 @@ public class CAFRpcClientImpl implements CAFRpcClient { String code = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_CODE, Metadata.ASCII_STRING_MARSHALLER)); int level = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)) == null ? 2 : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_LEVEL, Metadata.ASCII_STRING_MARSHALLER)), Integer.class); - String message = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)); - String innerMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER)); + String message = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)) == null ? + "" : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)), String.class); + String innerMessage = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER)) == null ? + "" : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_INNERMESSAGE, Metadata.ASCII_STRING_MARSHALLER)), String.class); OffsetDateTime date = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)) == null ? OffsetDateTime.now() : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DATE, Metadata.ASCII_STRING_MARSHALLER)), OffsetDateTime.class); String detail = trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_DETAIL, Metadata.ASCII_STRING_MARSHALLER)) == null ? @@ -744,10 +742,15 @@ public class CAFRpcClientImpl implements CAFRpcClient { new HashMap<>() : serializer.deserialize(trailers.get(Metadata.Key.of(GrpcVariable.GRPC_EX_EXTENSION_MESSAGE, Metadata.ASCII_STRING_MARSHALLER)), HashMap.class); if (exceptionType != null && exceptionType.equalsIgnoreCase(CAFRuntimeException.class.getTypeName())) { - return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, code, message + detail, null, ExceptionLevel.getLevel(level), isBizException); + if (detail != null) log.error("gRpc服务端报错, 详细堆栈如下: \n\n{}", detail); + return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, code, + String.format("Exception in gRpc calling, serviceId: %s, su: %s, exception: %s", serviceId, su, message), + e, ExceptionLevel.getLevel(level), isBizException); } else { - return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, - ExceptionErrorCode.rpcError, message + detail, null, ExceptionLevel.Error, false); + if (detail != null) log.error("gRpc服务端报错, 详细堆栈如下: \n\n{}", detail); + return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, ExceptionErrorCode.rpcError, + String.format("Exception in gRpc calling, serviceId: %s, su: %s, exception: %s", serviceId, su, message), + e, ExceptionLevel.Error, false); } } else { return new CAFRuntimeException(DefaultExceptionProperties.SERVICE_UNIT, diff --git a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcChannelBuilder.java b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcChannelBuilder.java index 9fb0634e7..6bcfc7e12 100644 --- a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcChannelBuilder.java +++ b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcChannelBuilder.java @@ -18,9 +18,6 @@ package io.iec.edp.caf.rpc.remote.grpc; import io.grpc.*; -import java.net.MalformedURLException; -import java.net.URL; - /** * gRpc客户端构造器标准方法 */ @@ -35,6 +32,10 @@ class GrpcChannelBuilder { public static ManagedChannelBuilder getChannelBuilder(String host, int port){ return ManagedChannelBuilder.forAddress(host, port) + // 设置最大 header 为 64K, 原默认值为 8k, 当前策略堆栈借助 header 返回, 当堆栈过大会导致整个header异常 + .maxInboundMetadataSize(64 * 1024) + // 设置最大传输大小为 64M + .maxInboundMessageSize(64 * 1024 * 1024) // 使用非安全机制传输 .usePlaintext(); } diff --git a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java index 6ead4dd0d..1047a48f1 100644 --- a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java +++ b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java @@ -26,7 +26,9 @@ import io.iec.edp.caf.commons.core.enums.SerializeType; import io.iec.edp.caf.commons.exception.CAFRuntimeException; import io.iec.edp.caf.commons.exception.ExceptionLevel; import io.iec.edp.caf.commons.exception.entity.DefaultExceptionProperties; +import io.iec.edp.caf.commons.exception.entity.ExceptionContext; import io.iec.edp.caf.commons.exception.entity.ExceptionErrorCode; +import io.iec.edp.caf.commons.exception.service.CafExceptionHandler; import io.iec.edp.caf.commons.utils.SpringBeanUtils; import io.iec.edp.caf.commons.utils.StringUtils; import io.iec.edp.caf.rpc.api.common.GspSerializeType; @@ -57,9 +59,6 @@ import java.util.LinkedHashMap; import java.util.Map; import java.util.UUID; -import static io.grpc.stub.ServerCalls.asyncUnimplementedStreamingCall; -import static io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall; - @Slf4j public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImplBase { @@ -122,20 +121,16 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp var intermanager = SpringBeanUtils.getBean(InternalServiceManageService.class); var rpcServiceMethodDefinition = intermanager.getRpcMethodDefinition(serviceId); var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params); -// var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params); result = rpcServer.invokeService(serviceId, "", parameters, GspSerializeType.Protobuf).toString(); } catch (Throwable e) { - var statusRuntimeException = handleException(e); responseObserver.onError(statusRuntimeException); serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException); throw e; } finally { RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue()); -// restoreTenant(tenantId); } - var contextDict = new HashMap(); this.serverEventBroker.firePostRpcInvokeEvent(contextDict, result, localDict); var eventContextStr = SerializerFactory.getSerializer(SerializeType.Json).serializeToString(contextDict); @@ -203,7 +198,6 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp var intermanager = SpringBeanUtils.getBean(InternalServiceManageService.class); var rpcServiceMethodDefinition = intermanager.getRpcMethodDefinition(serviceId); var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params); -// var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params); result = rpcServer.invokeServiceStream(serviceId, "", parameters); // download file as chunk byte[] bytes = new byte[4096]; @@ -217,14 +211,12 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp responseObserver.onNext(resp); } } catch (Throwable e) { - var statusRuntimeException = handleException(e); responseObserver.onError(statusRuntimeException); serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException); throw e; } finally { RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue()); -// restoreTenant(tenantId); if(result!=null){ result.close(); } @@ -303,14 +295,12 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp swapStream = new ByteArrayInputStream(content); params = Utils.handleStreamParam(rpcServiceMethodDefinition,params,swapStream); var parameters = RpcSerializeUtil.deSerializeParameter4RPC2(rpcServiceMethodDefinition,params); -// var parameters = RpcSerializeUtil.deSerializeParameter(rpcServiceMethodDefinition,params); result = rpcServer.invokeService(serviceId, "", parameters, GspSerializeType.Protobuf).toString(); } catch (Throwable e) { this.onError(e); } finally { RpcContextResolver.restore(rpcContext,RpcContextTransmittee.TransmitteType.TENANT.getValue()); -// restoreTenant(tenantId); swapStream.close(); } @@ -322,7 +312,6 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp @Override public void onError(Throwable throwable) { -// var errerMessage = throwable.getMessage(); var statusRuntimeException = handleException(throwable); responseObserver.onError(statusRuntimeException); serverEventBroker.fireExceptionRpcInvokeEvent(eventContextDict, params, localDict, statusRuntimeException); @@ -339,8 +328,13 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp }; } - private StatusRuntimeException handleException(Throwable e){ - log.error(e.getMessage(),e); + private StatusRuntimeException handleException(Throwable e) { + log.error(e.getMessage(), e); + + if (e instanceof Exception) { + // 异常处理, 类似 RESTExceptionMapper 的处理, 解决 InvocationTargetException 的问头 + e = SpringBeanUtils.getBean(CafExceptionHandler.class).handle(new ExceptionContext((Exception) e, null)); + } /** * grpc处理自定义异常的两种方式：1.基于metadata处理，2.基于google.rpc.Status方式 * 方式2简单，但是目前不是gRPC的官方方式，针对所有语言不具有通用性，这里优先使用方式1 @@ -349,7 +343,7 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp * https://cloud.tencent.com/developer/article/1618969 * */ Metadata metadata = null; - if(e instanceof CAFRuntimeException && ((CAFRuntimeException)e).isBizException()){ + if (e instanceof CAFRuntimeException && ((CAFRuntimeException) e).isBizException()) { metadata = GrpcExceptionMetaData.builder() .exceptionType(CAFRuntimeException.class.getTypeName()) .level(ExceptionLevel.Error.ordinal()) @@ -362,8 +356,7 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp .extensionMessage(((CAFRuntimeException) e).getExtensionMessage()) .build() .buildMetaData(); - - }else{ + } else { metadata = GrpcExceptionMetaData.builder() .code(ExceptionErrorCode.rpcError) .exceptionType(e.getClass().getTypeName()) @@ -375,9 +368,9 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp .build() .buildMetaData(); } - var statusRuntimeException = - Status.UNAVAILABLE.withDescription(e.getMessage()).asRuntimeException(metadata); - return statusRuntimeException; + return Status.UNAVAILABLE + .withDescription(e.getMessage()) + .asRuntimeException(metadata); } } -- Gitee From 80bc22c0212a20c44afad6fcbde8f90dbfc55ff9 Mon Sep 17 00:00:00 2001 From: Vincent Date: Thu, 28 May 2026 09:16:01 +0800 Subject: [PATCH 3/3] =?UTF-8?q?=E6=B3=A8=E9=87=8A=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java index 1047a48f1..c05e37baf 100644 --- a/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java +++ b/caf-rpc/caf-rpc-remote-grpc/src/main/java/io/iec/edp/caf/rpc/remote/grpc/GrpcServerInvoke.java @@ -332,7 +332,7 @@ public class GrpcServerInvoke extends GrpcInvokeServiceGrpc.GrpcInvokeServiceImp log.error(e.getMessage(), e); if (e instanceof Exception) { - // 异常处理, 类似 RESTExceptionMapper 的处理, 解决 InvocationTargetException 的问头 + // 异常处理, 类似 RESTExceptionMapper 的处理, 解决 InvocationTargetException 的问题 e = SpringBeanUtils.getBean(CafExceptionHandler.class).handle(new ExceptionContext((Exception) e, null)); } /** -- Gitee