natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'apache-3.0' into apache-3.1 

# Conflicts:
#	dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/AbstractRegistry.java
This commit is contained in:
Albumen Kevin 2022-08-15 11:16:43 +08:00
parent e069a7af9e c84b745bb1
commit 551f820a7e
16 changed files with 288 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
dubbo-common/src/main/java/org/apache/dubbo/common/concurrent/AbortPolicy.java Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.
*/
package org.apache.dubbo.common.concurrent;
import java.util.Queue;
/**
* A handler for rejected element that throws a {@code RejectException}.
*/
public class AbortPolicy<E> implements Rejector<E> {
@Override
public void reject(final E e, final Queue<E> queue) {
throw new RejectException("no more memory can be used !");
}
}

31
dubbo-common/src/main/java/org/apache/dubbo/common/concurrent/DiscardOldestPolicy.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.
*/
package org.apache.dubbo.common.concurrent;
import java.util.Queue;
/**
* A handler for rejected element that discards the oldest element.
*/
public class DiscardOldestPolicy<E> implements Rejector<E> {
@Override
public void reject(final E e, final Queue<E> queue) {
queue.poll();
queue.offer(e);
}
}

31
dubbo-common/src/main/java/org/apache/dubbo/common/concurrent/DiscardPolicy.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.
*/
package org.apache.dubbo.common.concurrent;
import java.util.Queue;
/**
* A handler for rejected element that silently discards the rejected element.
*/
public class DiscardPolicy<E> implements Rejector<E> {
@Override
public void reject(final E e, final Queue<E> queue) {
}
}

64
dubbo-common/src/main/java/org/apache/dubbo/common/concurrent/RejectException.java Normal file
View File

@ -0,0 +1,64 @@
/*
* 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.
*/
package org.apache.dubbo.common.concurrent;
import org.apache.dubbo.common.threadpool.MemorySafeLinkedBlockingQueue;
/**
* Exception thrown by an {@link MemorySafeLinkedBlockingQueue} when a element cannot be accepted.
*/
public class RejectException extends RuntimeException {
private static final long serialVersionUID = -3240015871717170195L;
/**
* Constructs a {@code RejectException} with no detail message. The cause is not initialized, and may subsequently be initialized by a
* call to {@link #initCause(Throwable) initCause}.
*/
public RejectException() {
}
/**
* Constructs a {@code RejectException} with the specified detail message. The cause is not initialized, and may subsequently be
* initialized by a call to {@link #initCause(Throwable) initCause}.
*
* @param message the detail message
*/
public RejectException(final String message) {
super(message);
}
/**
* Constructs a {@code RejectException} with the specified detail message and cause.
*
* @param message the detail message
* @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method)
*/
public RejectException(final String message, final Throwable cause) {
super(message, cause);
}
/**
* Constructs a {@code RejectException} with the specified cause. The detail message is set to {@code (cause == null ? null :
* cause.toString())} (which typically contains the class and detail message of {@code cause}).
*
* @param cause the cause (which is saved for later retrieval by the {@link #getCause()} method)
*/
public RejectException(final Throwable cause) {
super(cause);
}
}

43
dubbo-common/src/main/java/org/apache/dubbo/common/concurrent/Rejector.java Normal file
View File

@ -0,0 +1,43 @@
/*
* 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.
*/
package org.apache.dubbo.common.concurrent;
import java.util.Queue;
/**
* RejectHandler, it works when you need to custom reject action.
*
* @see AbortPolicy
* @see DiscardPolicy
* @see DiscardOldestPolicy
*/
public interface Rejector<E> {
/**
* Method that may be invoked by a Queue when Queue has remained memory
* return true. This may occur when no more memory are available because their bounds would be exceeded.
*
* <p>In the absence of other alternatives, the method may throw an unchecked
* {@link RejectException}, which will be propagated to the caller.
*
* @param e the element requested to be added
* @param queue the queue attempting to add this element
*
* @throws RejectException if there is no more memory
*/
void reject(E e, Queue<E> queue);
}

4
dubbo-common/src/main/java/org/apache/dubbo/common/json/impl/FastJsonImpl.java
View File

@ -18,6 +18,8 @@ package org.apache.dubbo.common.json.impl;
import org.apache.dubbo.common.utils.ClassUtils;
import com.alibaba.fastjson.serializer.SerializerFeature;
import java.lang.reflect.Type;
import java.util.List;
@ -45,6 +47,6 @@ public class FastJsonImpl extends AbstractJSONImpl {
@Override
public String toJson(Object obj) {
return com.alibaba.fastjson.JSON.toJSONString(obj);
return com.alibaba.fastjson.JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect);
}
}

31
dubbo-common/src/main/java/org/apache/dubbo/common/threadpool/MemorySafeLinkedBlockingQueue.java
View File

@ -17,6 +17,9 @@
package org.apache.dubbo.common.threadpool;
import org.apache.dubbo.common.concurrent.DiscardPolicy;
import org.apache.dubbo.common.concurrent.Rejector;
import java.util.Collection;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@ -36,6 +39,8 @@ public class MemorySafeLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
private int maxFreeMemory;
private Rejector<E> rejector;
public MemorySafeLinkedBlockingQueue() {
this(THE_256_MB);
}
@ -43,12 +48,16 @@ public class MemorySafeLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
public MemorySafeLinkedBlockingQueue(final int maxFreeMemory) {
super(Integer.MAX_VALUE);
this.maxFreeMemory = maxFreeMemory;
//default as DiscardPolicy to ensure compatibility with the old version
this.rejector = new DiscardPolicy<>();
}
public MemorySafeLinkedBlockingQueue(final Collection<? extends E> c,
final int maxFreeMemory) {
super(c);
this.maxFreeMemory = maxFreeMemory;
//default as DiscardPolicy to ensure compatibility with the old version
this.rejector = new DiscardPolicy<>();
}
/**
@ -69,6 +78,15 @@ public class MemorySafeLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
return maxFreeMemory;
}
/**
* set the rejector.
*
* @param rejector the rejector
*/
public void setRejector(final Rejector<E> rejector) {
this.rejector = rejector;
}
/**
* determine if there is any remaining free memory.
*
@ -83,15 +101,24 @@ public class MemorySafeLinkedBlockingQueue<E> extends LinkedBlockingQueue<E> {
if (hasRemainedMemory()) {
super.put(e);
}
rejector.reject(e, this);
}
@Override
public boolean offer(final E e, final long timeout, final TimeUnit unit) throws InterruptedException {
return hasRemainedMemory() && super.offer(e, timeout, unit);
if (!hasRemainedMemory()) {
rejector.reject(e, this);
return false;
}
return super.offer(e, timeout, unit);
}
@Override
public boolean offer(final E e) {
return hasRemainedMemory() && super.offer(e);
if (!hasRemainedMemory()) {
rejector.reject(e, this);
return false;
}
return super.offer(e);
}
}

11
dubbo-common/src/test/java/org/apache/dubbo/common/threadpool/MemorySafeLinkedBlockingQueueTest.java
View File

@ -18,12 +18,15 @@
package org.apache.dubbo.common.threadpool;
import net.bytebuddy.agent.ByteBuddyAgent;
import org.apache.dubbo.common.concurrent.AbortPolicy;
import org.apache.dubbo.common.concurrent.RejectException;
import org.junit.jupiter.api.Test;
import java.lang.instrument.Instrumentation;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class MemorySafeLinkedBlockingQueueTest {
@Test
@ -43,4 +46,12 @@ public class MemorySafeLinkedBlockingQueueTest {
assertThat(queue.offer(() -> {
}), is(true));
}
@Test
public void testCustomReject() throws Exception {
MemorySafeLinkedBlockingQueue<Runnable> queue = new MemorySafeLinkedBlockingQueue<>(Integer.MAX_VALUE);
queue.setRejector(new AbortPolicy<>());
assertThrows(RejectException.class, () -> queue.offer(() -> {
}));
}
}

3
dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/beans/factory/annotation/ServiceBeanNameBuilder.java
View File

@ -95,8 +95,9 @@ public class ServiceBeanNameBuilder {
}
private static void append(StringBuilder builder, String value) {
builder.append(SEPARATOR);
if (StringUtils.hasText(value)) {
builder.append(SEPARATOR).append(value);
builder.append(value);
}
}

23
dubbo-config/dubbo-config-spring/src/main/java/org/apache/dubbo/config/spring/reference/ReferenceBeanManager.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.config.spring.reference;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.spring.ReferenceBean;
import org.apache.dubbo.config.spring.util.DubboBeanUtils;
@ -34,6 +35,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@ -67,8 +69,10 @@ public class ReferenceBeanManager implements ApplicationContextAware {
" the BeanPostProcessor has not been loaded at this time, which may cause abnormalities in some components (such as seata): " +
referenceBeanName + " = " + ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext));
}
String referenceKey = ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext);
String referenceKey = getReferenceKeyByBeanName(referenceBeanName);
if (StringUtils.isEmpty(referenceKey)) {
referenceKey = ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext);
}
ReferenceBean oldReferenceBean = referenceBeanMap.get(referenceBeanName);
if (oldReferenceBean != null) {
if (referenceBean != oldReferenceBean) {
@ -88,6 +92,16 @@ public class ReferenceBeanManager implements ApplicationContextAware {
}
}
private String getReferenceKeyByBeanName(String referenceBeanName){
Set<Map.Entry<String, List<String>>> entries = referenceKeyMap.entrySet();
for (Map.Entry<String, List<String>> entry : entries) {
if (entry.getValue().contains(referenceBeanName)) {
return entry.getKey();
}
}
return null;
}
public void registerReferenceKeyAndBeanName(String referenceKey, String referenceBeanNameOrAlias) {
List<String> list = referenceKeyMap.computeIfAbsent(referenceKey, (key) -> new ArrayList<>());
if (!list.contains(referenceBeanNameOrAlias)) {
@ -148,7 +162,10 @@ public class ReferenceBeanManager implements ApplicationContextAware {
// TOTO check same unique service name but difference reference key (means difference attributes).
// reference key
String referenceKey = ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext);
String referenceKey = getReferenceKeyByBeanName(referenceBean.getId());
if (StringUtils.isEmpty(referenceKey)) {
referenceKey = ReferenceBeanSupport.generateReferenceKey(referenceBean, applicationContext);
}
ReferenceConfig referenceConfig = referenceConfigMap.get(referenceKey);
if (referenceConfig == null) {

2
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/beans/factory/annotation/ServiceAnnotationPostProcessorTest.java
View File

@ -95,7 +95,7 @@ public class ServiceAnnotationPostProcessorTest {
Assertions.assertEquals(3, serviceBeansMap.size());
ServiceBean demoServiceBean = serviceBeansMap.get("ServiceBean:org.apache.dubbo.config.spring.api.DemoService:2.5.7");
ServiceBean demoServiceBean = serviceBeansMap.get("ServiceBean:org.apache.dubbo.config.spring.api.DemoService:2.5.7:");
Assertions.assertNotNull(demoServiceBean.getMethods());

10
dubbo-config/dubbo-config-spring/src/test/java/org/apache/dubbo/config/spring/beans/factory/annotation/ServiceBeanNameBuilderTest.java
View File

@ -72,4 +72,14 @@ public class ServiceBeanNameBuilderTest {
builder.build());
}
@Test
public void testServiceNameBuild() {
ServiceBeanNameBuilder vBuilder = ServiceBeanNameBuilder.create(INTERFACE_CLASS, environment);
String vBeanName = vBuilder.version("DUBBO").build();
ServiceBeanNameBuilder gBuilder = ServiceBeanNameBuilder.create(INTERFACE_CLASS, environment);
String gBeanName = gBuilder.group("DUBBO").build();
Assertions.assertNotEquals(vBeanName, gBeanName);
}
}

2
dubbo-configcenter/dubbo-configcenter-zookeeper/src/main/java/org/apache/dubbo/configcenter/support/zookeeper/ZookeeperDynamicConfiguration.java
View File

@ -155,7 +155,7 @@ public class ZookeeperDynamicConfiguration extends TreePathDynamicConfiguration
@Override
protected void doRemoveListener(String pathKey, ConfigurationListener listener) {
ZookeeperDataListener zookeeperDataListener = cacheListener.removeListener(pathKey, listener);
if (CollectionUtils.isEmpty(zookeeperDataListener.getListeners())) {
if (zookeeperDataListener != null && CollectionUtils.isEmpty(zookeeperDataListener.getListeners())) {
zkClient.removeDataListener(pathKey, zookeeperDataListener);
}
}

13
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/support/AbstractRegistry.java
View File

@ -49,7 +49,8 @@ import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
@ -82,6 +83,8 @@ public abstract class AbstractRegistry implements Registry {
private static final String URL_SPLIT = "\\s+";
// Max times to retry to save properties to local cache file
private static final int MAX_RETRY_TIMES_SAVE_PROPERTIES = 3;
// Default interval in millisecond for saving properties to local cache file
private static final long DEFAULT_INTERVAL_SAVE_PROPERTIES = 500L;
// Log output
protected final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(getClass());
@ -89,7 +92,7 @@ public abstract class AbstractRegistry implements Registry {
// Local disk cache, where the special key value.registries records the list of registry centers, and the others are the list of notified service providers
private final Properties properties = new Properties();
// File cache timing writing
private final ExecutorService registryCacheExecutor;
private final ScheduledExecutorService registryCacheExecutor;
private final AtomicLong lastCacheChanged = new AtomicLong();
private final AtomicInteger savePropertiesRetryTimes = new AtomicInteger();
private final Set<URL> registered = new ConcurrentHashSet<>();
@ -111,7 +114,7 @@ public abstract class AbstractRegistry implements Registry {
registryManager = url.getOrDefaultApplicationModel().getBeanFactory().getBean(RegistryManager.class);
localCacheEnabled = url.getParameter(REGISTRY_LOCAL_FILE_CACHE_ENABLED, true);
registryCacheExecutor = url.getOrDefaultFrameworkModel().getBeanFactory()
.getBean(FrameworkExecutorRepository.class).getSharedExecutor();
.getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor();
if (localCacheEnabled) {
// Start file save timer
syncSaveFile = url.getParameter(REGISTRY_FILESAVE_SYNC_KEY, false);
@ -276,7 +279,7 @@ public abstract class AbstractRegistry implements Registry {
savePropertiesRetryTimes.set(0);
return;
} else {
registryCacheExecutor.execute(() -> doSaveProperties(lastCacheChanged.incrementAndGet()));
registryCacheExecutor.schedule(() -> doSaveProperties(lastCacheChanged.incrementAndGet()), DEFAULT_INTERVAL_SAVE_PROPERTIES, TimeUnit.MILLISECONDS);
}
if (!(e instanceof OverlappingFileLockException)) {
@ -548,7 +551,7 @@ public abstract class AbstractRegistry implements Registry {
if (syncSaveFile) {
doSaveProperties(version);
} else {
registryCacheExecutor.execute(() -> doSaveProperties(version));
registryCacheExecutor.schedule(() -> doSaveProperties(version), DEFAULT_INTERVAL_SAVE_PROPERTIES, TimeUnit.MILLISECONDS);
}
} catch (Throwable t) {
logger.warn(t.getMessage(), t);

2
dubbo-spring-boot/dubbo-spring-boot-actuator/src/test/java/org/apache/dubbo/spring/boot/actuate/autoconfigure/DubboEndpointAnnotationAutoConfigurationTest.java
View File

@ -179,7 +179,7 @@ public class DubboEndpointAnnotationAutoConfigurationTest {
Assert.assertEquals(1, services.size());
Map<String, Object> demoServiceMeta = services.get("ServiceBean:org.apache.dubbo.spring.boot.actuate.autoconfigure.DubboEndpointAnnotationAutoConfigurationTest$DemoService:1.0.0");
Map<String, Object> demoServiceMeta = services.get("ServiceBean:org.apache.dubbo.spring.boot.actuate.autoconfigure.DubboEndpointAnnotationAutoConfigurationTest$DemoService:1.0.0:");
Assert.assertEquals("1.0.0", demoServiceMeta.get("version"));

2
dubbo-spring-boot/dubbo-spring-boot-compatible/actuator/src/test/java/org/apache/dubbo/spring/boot/actuate/autoconfigure/DubboEndpointAutoConfigurationTest.java
View File

@ -166,7 +166,7 @@ public class DubboEndpointAutoConfigurationTest {
Assert.assertEquals(1, services.size());
Map<String, Object> demoServiceMeta = services.get("ServiceBean:org.apache.dubbo.spring.boot.actuate.autoconfigure.DubboEndpointAutoConfigurationTest$DemoService:1.0.0");
Map<String, Object> demoServiceMeta = services.get("ServiceBean:org.apache.dubbo.spring.boot.actuate.autoconfigure.DubboEndpointAutoConfigurationTest$DemoService:1.0.0:");
Assert.assertEquals("1.0.0", demoServiceMeta.get("version"));