Skip to content

Commit

Permalink
Bugfix of retries and missing field (apache#182)
Browse files Browse the repository at this point in the history 
Co-authored-by: x-shadow-man <1494445739@qq.com>
  • Loading branch information
@wxbty
wxbty and x-shadow-man authored Dec 16, 2022
1 parent 12a5931 commit e983761
Show file tree
Hide file tree
Showing 20 changed files with 251 additions and 209 deletions.
40 changes: 40 additions & 0 deletions dubbo-cluster-extensions/dubbo-cluster-specify-address-common/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>dubbo-cluster-extensions</artifactId>
<groupId>org.apache.dubbo.extensions</groupId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>

<artifactId>dubbo-cluster-specify-address-common</artifactId>
<version>1.0.2-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-common</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>


</project>
6 changes: 6 additions & 0 deletions ...o/rpc/cluster/specifyaddress/Address.java → ...o/rpc/cluster/specifyaddress/Address.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
import java.util.Objects;

public class Address implements Serializable {

public static final String name = "specifyAddress";

// ip - priority: 3
private String ip;

Expand All @@ -38,6 +41,9 @@ public Address(String ip, int port) {
this.urlAddress = null;
}

/**
* disableRetry default value is true, will disable failover
*/
public Address(String ip, int port, boolean needToCreate) {
this.ip = ip;
this.port = port;
Expand Down
0 ...cifyaddress/UserSpecifiedAddressUtil.java → ...cifyaddress/UserSpecifiedAddressUtil.java
View file
File renamed without changes.
9 changes: 4 additions & 5 deletions .../cluster/specifyaddress/InvokerCache.java → ...r/specifyaddress/common/InvokerCache.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,23 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.cluster.specifyaddress;
package org.apache.dubbo.rpc.cluster.specifyaddress.common;

import org.apache.dubbo.rpc.Invoker;

public class InvokerCache<T> {
private long lastAccess = System.currentTimeMillis();
private final Invoker<T> invoker;
private final T invoker;

public InvokerCache(Invoker<T> invoker) {
public InvokerCache(T invoker) {
this.invoker = invoker;
}

public long getLastAccess() {
return lastAccess;
}

public Invoker<T> getInvoker() {
public T getInvoker() {
lastAccess = System.currentTimeMillis();
return invoker;
}
Expand Down
5 changes: 5 additions & 0 deletions dubbo-cluster-extensions/dubbo-cluster-specify-address-dubbo2/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,11 @@
<version>2.7.18</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.dubbo.extensions</groupId>
<artifactId>dubbo-cluster-specify-address-common</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>

</project>
37 changes: 20 additions & 17 deletions ...cifyaddress/UserSpecifiedAddressUtil.java → ...ess/AddressSpecifyClusterInterceptor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,26 +16,29 @@
*/
package org.apache.dubbo.rpc.cluster.specifyaddress;

import org.apache.dubbo.common.threadlocal.InternalThreadLocal;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.cluster.interceptor.ClusterInterceptor;
import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;

public class UserSpecifiedAddressUtil {
private final static InternalThreadLocal<Address> ADDRESS = new InternalThreadLocal<>();
/**
* The SPECIFY ADDRESS field is handed over to the attachment by the thread
*/
@Activate(group = CommonConstants.CONSUMER)
public class AddressSpecifyClusterInterceptor implements ClusterInterceptor {

/**
* Set specified address to next invoke
*
* @param address specified address
*/
public static void setAddress(Address address) {
ADDRESS.set(address);
@Override
public void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {
Address current = UserSpecifiedAddressUtil.getAddress();
if (current != null) {
invocation.put(Address.name, current);
}
}

public static Address getAddress() {
try {
return ADDRESS.get();
} finally {
// work once
ADDRESS.remove();
}

@Override
public void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {

}
}
25 changes: 15 additions & 10 deletions ...src/main/java/org/apache/dubbo/rpc/cluster/specifyaddress/UserSpecifiedAddressRouter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,24 +30,25 @@
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.AbstractRouter;
import org.apache.dubbo.rpc.cluster.specifyaddress.common.InvokerCache;

import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Iterator;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.MONITOR_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;


public class UserSpecifiedAddressRouter<T> extends AbstractRouter {
Expand All @@ -66,7 +67,7 @@ public class UserSpecifiedAddressRouter<T> extends AbstractRouter {
private final AtomicBoolean launchRemovalTask = new AtomicBoolean(false);


private final Map<URL, InvokerCache<T>> newInvokerCache = new LinkedHashMap<>(16, 0.75f, true);
private final Map<URL, InvokerCache<Invoker<T>>> newInvokerCache = new LinkedHashMap<>(16, 0.75f, true);

public UserSpecifiedAddressRouter(URL referenceUrl) {
super(referenceUrl);
Expand All @@ -85,14 +86,18 @@ public <T> void notify(List<Invoker<T>> invokers) {
}

@Override
@SuppressWarnings("unchecked")
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
Address address = UserSpecifiedAddressUtil.getAddress();

Object addressObj = invocation.get(Address.name);

// 1. check if set address in ThreadLocal
if (address == null) {
if (addressObj == null) {
return invokers;
}

Address address = (Address) addressObj;

List<Invoker<T>> result = new LinkedList<>();

// 2. check if set address url
Expand Down Expand Up @@ -259,7 +264,7 @@ private URL copyConsumerUrl(URL url, String ip, int port, Map<String, String> pa
}

public URL rebuildAddress(Address address, URL consumerUrl) {
URL url = address.getUrlAddress();
URL url = (URL) address.getUrlAddress();
Map<String, String> parameters = new HashMap<>(url.getParameters());
parameters.put(VERSION_KEY, consumerUrl.getParameter(VERSION_KEY, "0.0.0"));
parameters.put(GROUP_KEY, consumerUrl.getParameter(GROUP_KEY));
Expand All @@ -270,7 +275,7 @@ public URL rebuildAddress(Address address, URL consumerUrl) {
private Invoker<T> getOrBuildInvokerCache(URL url) {
logger.info("Unable to find a proper invoker from directory. Try to create new invoker. New URL: " + url);

InvokerCache<T> cache;
InvokerCache<Invoker<T>> cache;
cacheLock.lock();
try {
cache = newInvokerCache.get(url);
Expand Down Expand Up @@ -314,9 +319,9 @@ public void run() {
cacheLock.lock();
try {
if (newInvokerCache.size() > 0) {
Iterator<Map.Entry<URL, InvokerCache<T>>> iterator = newInvokerCache.entrySet().iterator();
Iterator<Map.Entry<URL, InvokerCache<Invoker<T>>>> iterator = newInvokerCache.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<URL, InvokerCache<T>> entry = iterator.next();
Map.Entry<URL, InvokerCache<Invoker<T>>> entry = iterator.next();
if (System.currentTimeMillis() - entry.getValue().getLastAccess() > EXPIRE_TIME) {
iterator.remove();
entry.getValue().getInvoker().destroy();
Expand Down
1 change: 1 addition & 0 deletions ...main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.interceptor.ClusterInterceptor
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
disableRetry=org.apache.dubbo.rpc.cluster.specifyaddress.AddressSpecifyClusterInterceptor
3 changes: 2 additions & 1 deletion ...ss-dubbo2/src/test/java/org/apache/dubbo/rpc/cluster/specifyaddress/InvokerCacheTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,14 +18,15 @@

import org.apache.dubbo.rpc.Invoker;

import org.apache.dubbo.rpc.cluster.specifyaddress.common.InvokerCache;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

public class InvokerCacheTest {
@Test
public void test() throws InterruptedException {
InvokerCache<Object> cache = new InvokerCache<>(Mockito.mock(Invoker.class));
InvokerCache<Invoker<Object>> cache = new InvokerCache<>(Mockito.mock(Invoker.class));
long originTime = cache.getLastAccess();
Thread.sleep(5);
cache.getInvoker();
Expand Down
Loading

0 comments on commit e983761

Please sign in to comment.