You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2020/11/13 10:18:28 UTC
[dubbo] branch 3.0 updated: save current effective invoker to
consumer model on migration rule change
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new 81890a8 save current effective invoker to consumer model on migration rule change
81890a8 is described below
commit 81890a85fa9b7f8ac02dc93072697c8f704970e9
Author: ken.lj <ke...@gmail.com>
AuthorDate: Fri Nov 13 18:17:34 2020 +0800
save current effective invoker to consumer model on migration rule change
---
.../java/org/apache/dubbo/rpc/cluster/Directory.java | 1 +
.../java/org/apache/dubbo/rpc/cluster/RouterChain.java | 4 ++++
.../listener/ServiceInstancesChangedListener.java | 2 +-
.../registry/client/migration/MigrationInvoker.java | 18 ++++++++++++++++++
.../client/migration/MigrationRuleHandler.java | 10 +++++++---
.../main/java/org/apache/dubbo/rpc/RpcInvocation.java | 2 +-
.../dubbo/rpc/proxy/InvokerInvocationHandler.java | 4 ++--
7 files changed, 34 insertions(+), 7 deletions(-)
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
index 208b0c2..e6a20fd 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
@@ -55,4 +55,5 @@ public interface Directory<T> extends Node {
void discordAddresses();
+ RouterChain<T> getRouterChain();
}
\ No newline at end of file
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
index e340e3d..e69a90a 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/RouterChain.java
@@ -83,6 +83,10 @@ public class RouterChain<T> {
this.routers = newRouters;
}
+ public List<Router> getRouters() {
+ return routers;
+ }
+
private void sort() {
Collections.sort(routers);
}
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
index 7eb6f76..aff9c22 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/listener/ServiceInstancesChangedListener.java
@@ -308,7 +308,7 @@ public class ServiceInstancesChangedListener implements ConditionalEventListener
listeners.forEach((key, notifyListener) -> {
//FIXME, group wildcard match
List<URL> urls = toUrlsWithEmpty(serviceUrls.get(key));
- logger.info("Notify service " + key + " with services " + urls.size());
+ logger.info("Notify service " + key + " with urls " + urls.size());
notifyListener.notify(urls);
});
}
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
index f9c7265..f8a78cb 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
@@ -33,6 +33,8 @@ import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Cluster;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import org.apache.dubbo.rpc.cluster.Directory;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.model.ConsumerModel;
import java.util.List;
import java.util.Set;
@@ -286,6 +288,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
protected synchronized void destroyServiceDiscoveryInvoker(ClusterInvoker<?> serviceDiscoveryInvoker) {
if (this.invoker != null) {
this.currentAvailableInvoker = this.invoker;
+ updateConsumerModel(currentAvailableInvoker, serviceDiscoveryInvoker);
}
if (serviceDiscoveryInvoker != null) {
if (logger.isDebugEnabled()) {
@@ -298,6 +301,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
protected synchronized void discardServiceDiscoveryInvokerAddress(ClusterInvoker<T> serviceDiscoveryInvoker) {
if (this.invoker != null) {
this.currentAvailableInvoker = this.invoker;
+ updateConsumerModel(currentAvailableInvoker, serviceDiscoveryInvoker);
}
if (serviceDiscoveryInvoker != null) {
if (logger.isDebugEnabled()) {
@@ -337,6 +341,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
protected synchronized void destroyInterfaceInvoker(ClusterInvoker<T> invoker) {
if (checkInvokerAvailable(this.serviceDiscoveryInvoker)) {
this.currentAvailableInvoker = this.serviceDiscoveryInvoker;
+ updateConsumerModel(currentAvailableInvoker, invoker);
}
if (invoker != null) {
if (logger.isDebugEnabled()) {
@@ -349,6 +354,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
protected synchronized void discardInterfaceInvokerAddress(ClusterInvoker<T> invoker) {
if (this.serviceDiscoveryInvoker != null) {
this.currentAvailableInvoker = this.serviceDiscoveryInvoker;
+ updateConsumerModel(currentAvailableInvoker, invoker);
}
if (invoker != null) {
if (logger.isDebugEnabled()) {
@@ -378,4 +384,16 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
public boolean checkInvokerAvailable(ClusterInvoker<T> invoker) {
return invoker != null && !invoker.isDestroyed() && invoker.isAvailable();
}
+
+ private void updateConsumerModel(ClusterInvoker<?> workingInvoker, ClusterInvoker<?> backInvoker) {
+ ConsumerModel consumerModel = ApplicationModel.getConsumerModel(consumerUrl.getServiceKey());
+ if (consumerModel != null) {
+ if (workingInvoker != null) {
+ consumerModel.getServiceMetadata().addAttribute("currentClusterInvoker", workingInvoker);
+ }
+ if (backInvoker != null) {
+ consumerModel.getServiceMetadata().addAttribute("backupClusterInvoker", backInvoker);
+ }
+ }
+ }
}
diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleHandler.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleHandler.java
index 766368f..9c4071b 100644
--- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleHandler.java
+++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleHandler.java
@@ -94,9 +94,13 @@ public class MigrationRuleHandler<T> {
}
public void setCurrentStepAndThreshold(MigrationStep currentStep, Float currentThreshold) {
- this.currentStep = currentStep;
- this.currentThreshold = currentThreshold;
- this.migrationInvoker.setMigrationStep(currentStep);
+ if (currentThreshold != null) {
+ this.currentThreshold = currentThreshold;
+ }
+ if (currentStep != null) {
+ this.currentStep = currentStep;
+ this.migrationInvoker.setMigrationStep(currentStep);
+ }
}
public void setMigrationRule(MigrationRule rule) {
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
index 4a78cd6..be830d7 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/RpcInvocation.java
@@ -70,7 +70,7 @@ public class RpcInvocation implements Invocation, Serializable {
/**
* Only used on the caller side, will not appear on the wire.
*/
- private Map<Object, Object> attributes = new HashMap<Object, Object>();
+ private transient Map<Object, Object> attributes = new HashMap<Object, Object>();
private transient Invoker<?> invoker;
diff --git a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java
index 69f1d06..fb7ec51 100644
--- a/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java
+++ b/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/proxy/InvokerInvocationHandler.java
@@ -69,11 +69,11 @@ public class InvokerInvocationHandler implements InvocationHandler {
return invoker.equals(args[0]);
}
RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface().getName(), protocolServiceKey, args);
- String serviceKey = invoker.getUrl().getServiceKey();
+ String serviceKey = url.getServiceKey();
rpcInvocation.setTargetServiceUniqueName(serviceKey);
// invoker.getUrl() returns consumer url.
- RpcContext.setRpcContext(invoker.getUrl());
+ RpcContext.setRpcContext(url);
if (consumerModel != null) {
rpcInvocation.put(Constants.CONSUMER_MODEL, consumerModel);