概述
1.4 通过代理bean调用远程方法。
- 底层调用方式为异步调用，
demoService 是一个动态代理，在执行方法时，先执行InvokerInvocationHandler.invoke()方法，代码如下：
//InvokerInvocationHandler.invoke()
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
//方法来自Object方法
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
其中的invoker是一个包含目标类DubboInvoker的代理类AsyncToSyncInvoker的实例，AsyncToSyncInvoker的invoke()如下。
通过调用目标类获取远程方法调用结果，
//AsyncToSyncInvoker.invoke()
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult = invoker.invoke(invocation);
try {
if (InvokeMode.SYNC == ((RpcInvocation) invocation).getInvokeMode()) {
asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
}
} catch (InterruptedException e) {
throw new RpcException("Interrupted unexpectedly while waiting for remoting result to return! method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (ExecutionException e) {
Throwable t = e.getCause();
if (t instanceof TimeoutException) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} else if (t instanceof RemotingException) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
} catch (Throwable e) {
throw new RpcException(e.getMessage(), e);
}
DubboInvoker.invoke的方法继承自AbstractInvoker，代码如下：
//AbstractInvoker.invoke
public Result invoke(Invocation inv) throws RpcException {
// if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
if (destroyed.get()) {
logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
+ ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
}
RpcInvocation invocation = (RpcInvocation) inv;
invocation.setInvoker(this);
if (CollectionUtils.isNotEmptyMap(attachment)) {
//attachment是一个map,参数{interface=org.apache.dubbo.demo.DemoService}
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/
//暂时未空
invocation.addAttachments(contextAttachments);
}
//设置调用模式
invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation));
//设置调用id,自增
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
return doInvoke(invocation);
} catch (InvocationTargetException e) { // biz exception
Throwable te = e.getTargetException();
if (te == null) {
return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
} else {
if (te instanceof RpcException) {
((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
}
return AsyncRpcResult.newDefaultAsyncResult(null, te, invocation);
}
} catch (RpcException e) {
if (e.isBiz()) {
return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
} else {
throw e;
}
} catch (Throwable e) {
return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
}
}
调用DubboInvoker.doInvoke()方法，代码如下：
//DubboInvoker.doInvoke()
protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(PATH_KEY, getUrl().getPath());
inv.setAttachment(VERSION_KEY, version);
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = getUrl().getMethodPositiveParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
if (isOneway) {
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
return AsyncRpcResult.newDefaultAsyncResult(invocation);
} else {
AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
asyncRpcResult.subscribeTo(responseFuture);
// save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
FutureContext.getContext().setCompatibleFuture(responseFuture);
return asyncRpcResult;
}
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
}
}
上述代码中主要调用方法时currentClient.request(inv, timeout)；currentClient来源于clients，由3.3可知，currentClient.request执行的是ReferenceCountExchangeClient.request方法，代码如下：
//ReferenceCountExchangeClient.request
@Override
public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
return client.request(request, timeout);
}
ReferenceCountExchangeClient是一个代理类。
目标类是LazyConnectExchangeClient或HeaderExchangeClient。
1.4.1 非懒加载HeaderExchangeClient.request
@Override
public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
return channel.request(request, timeout);
//this.channel = new HeaderExchangeChannel(client); client是NettyClient的实例
}
//HeaderExchangeChannel. request
@Override
public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
}
// create request.
Request req = new Request();
req.setVersion(Version.getProtocolVersion());
req.setTwoWay(true);
req.setData(request);
DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout);
try {
//channel是NettyClient的实例
channel.send(req);
} catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
ExchangeClient[]中存放的是ReferenceCountExchangeClient，ReferenceCountExchangeClient是一个代理类，目标类是LazyConnectExchangeClient（xml中配置lazy=true）或者HeaderExchangeClient，其中HeaderExchangeClient中client存放NettyClient的实例。
1.4.1 懒加载LazyConnectExchangeClient.request
//LazyConnectExchangeClient.request
public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
this.warning();
this.initClient();
return this.client.request(request, timeout);
}
private void initClient() throws RemotingException {
if (client != null) {
return;
}
if (logger.isInfoEnabled()) {
logger.info("Lazy connect to " + url);
}
connectLock.lock();
try {
if (client != null) {
return;
}
this.client = Exchangers.connect(url, requestHandler);
} finally {
connectLock.unlock();
}
}
懒加载是在调用的时候才会调用Exchangers.connect(url, requestHandler)获取client，
最后
以上就是动听钻石为你收集整理的3.4 dubbo-消费方通过代理bean调用远程方法的全部内容，希望文章能够帮你解决3.4 dubbo-消费方通过代理bean调用远程方法所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
发表评论 取消回复