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

Merge branch 'cloud-native' of https://github.com/apache/dubbo into cloud-native

This commit is contained in:
ken.lj 2019-08-20 14:50:17 +08:00
parent 7b9742d1be 21e75e0252
commit ffa50bd8a7
34 changed files with 718 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-all/pom.xml
View File

@ -791,6 +791,10 @@
<resource>META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory
</resource>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscovery
</resource>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/dubbo/internal/org.apache.dubbo.metadata.definition.builder.TypeBuilder
</resource>

21
dubbo-bootstrap/pom.xml
View File

@ -33,6 +33,18 @@
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-etcd3</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>20.0</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-consul</artifactId>
@ -47,6 +59,13 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-report-etcd</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-configcenter-zookeeper</artifactId>
@ -110,4 +129,4 @@
</dependencies>
</project>
</project>

262
dubbo-bootstrap/src/main/java/org/apache/dubbo/bootstrap/DubboBootstrap.java
View File

@ -49,19 +49,21 @@ import org.apache.dubbo.config.metadata.ConfigurableMetadataServiceExporter;
import org.apache.dubbo.config.utils.ReferenceConfigCache;
import org.apache.dubbo.event.EventDispatcher;
import org.apache.dubbo.event.EventListener;
import org.apache.dubbo.event.GenericEventListener;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.MetadataServiceExporter;
import org.apache.dubbo.metadata.WritableMetadataService;
import org.apache.dubbo.metadata.report.MetadataReportInstance;
import org.apache.dubbo.registry.client.AbstractServiceDiscoveryFactory;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.event.ServiceDiscoveryInitializingEvent;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.SortedSet;
@ -73,13 +75,16 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import static java.util.Arrays.asList;
import static java.util.Collections.sort;
import static java.util.concurrent.Executors.newSingleThreadExecutor;
import static org.apache.dubbo.common.config.ConfigurationUtils.parseProperties;
import static org.apache.dubbo.common.config.configcenter.DynamicConfiguration.getDynamicConfiguration;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_REMOTE;
import static org.apache.dubbo.common.function.ThrowableAction.execute;
import static org.apache.dubbo.common.utils.StringUtils.isNotEmpty;
import static org.apache.dubbo.config.context.ConfigManager.getInstance;
import static org.apache.dubbo.metadata.WritableMetadataService.getExtension;
import static org.apache.dubbo.metadata.WritableMetadataService.getMetadataStorageType;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.setMetadataStorageType;
import static org.apache.dubbo.remoting.Constants.CLIENT_KEY;
@ -88,7 +93,7 @@ import static org.apache.dubbo.remoting.Constants.CLIENT_KEY;
*
* @since 2.7.4
*/
public class DubboBootstrap implements Lifecycle {
public class DubboBootstrap extends GenericEventListener implements Lifecycle {
public static final String DEFAULT_REGISTRY_ID = "REGISTRY#DEFAULT";
@ -137,10 +142,25 @@ public class DubboBootstrap implements Lifecycle {
private volatile MetadataServiceExporter metadataServiceExporter;
private volatile List<ServiceDiscovery> serviceDiscoveries = new LinkedList<>();
public DubboBootstrap() {
DubboShutdownHook.getDubboShutdownHook().register();
}
/**
* Store the {@link ServiceDiscovery} instances into {@link ServiceDiscoveryInitializingEvent}
*
* @param event {@link ServiceDiscoveryInitializingEvent}
* @see {@linkplan org.apache.dubbo.registry.client.EventPublishingServiceDiscovery}
*/
public void onServiceDiscoveryInitializing(ServiceDiscoveryInitializingEvent event) {
executeMutually(() -> {
serviceDiscoveries.add(event.getSource());
sort(serviceDiscoveries);
});
}
/**
* Set only register provider or not
*
@ -436,10 +456,14 @@ public class DubboBootstrap implements Lifecycle {
useRegistryAsConfigCenterIfNecessary();
initApplicationMetadata();
initMetadataService();
initMetadataServiceExporter();
initEventListener();
initialized = true;
if (logger.isInfoEnabled()) {
@ -451,56 +475,45 @@ public class DubboBootstrap implements Lifecycle {
return this;
}
/**
* Initialize {@link MetadataService} from {@link WritableMetadataService}'s extension
*/
private void initMetadataService() {
this.metadataService = getExtension(isDefaultMetadataStorageType());
private void initApplicationMetadata() {
String metadataStorageType = getMetadataStorageType(isDefaultMetadataStorageType());
getApplication().setMetadataStorageType(metadataStorageType);
}
/**
* Initialize {@link MetadataServiceExporter}
*/
private void initMetadataServiceExporter() {
this.metadataServiceExporter = new ConfigurableMetadataServiceExporter()
.setApplicationConfig(getApplication())
.setRegistries(configManager.getRegistries())
.setProtocols(configManager.getProtocols())
.metadataService(metadataService);
private void startConfigCenter() {
Collection<ConfigCenterConfig> configCenters = configManager.getConfigCenters();
if (CollectionUtils.isNotEmpty(configCenters)) {
CompositeDynamicConfiguration compositeDynamicConfiguration = new CompositeDynamicConfiguration();
for (ConfigCenterConfig configCenter : configCenters) {
configCenter.refresh();
compositeDynamicConfiguration.addConfiguration(prepareEnvironment(configCenter));
}
Environment.getInstance().setDynamicConfiguration(compositeDynamicConfiguration);
}
configManager.refreshAll();
}
private void loadRemoteConfigs() {
// registry ids to registry configs
List<RegistryConfig> tmpRegistries = new ArrayList<>();
Set<String> registryIds = configManager.getRegistryIds();
registryIds.forEach(id -> {
if (tmpRegistries.stream().noneMatch(reg -> reg.getId().equals(id))) {
tmpRegistries.add(configManager.getRegistry(id).orElseGet(() -> {
RegistryConfig registryConfig = new RegistryConfig();
registryConfig.setId(id);
registryConfig.refresh();
return registryConfig;
}));
private void startMetadataReport() {
ApplicationConfig applicationConfig = configManager.getApplication().orElseThrow(
() -> new IllegalStateException("There's no ApplicationConfig specified.")
);
String metadataType = applicationConfig.getMetadataStorageType();
// FIXME, multiple metadata config support.
Collection<MetadataReportConfig> metadataReportConfigs = configManager.getMetadataConfigs();
if (CollectionUtils.isEmpty(metadataReportConfigs)) {
if (METADATA_REMOTE.equals(metadataType)) {
throw new IllegalStateException("No MetadataConfig found, you must specify the remote Metadata Center address when set 'metadata=remote'.");
}
});
return;
}
MetadataReportConfig metadataReportConfig = metadataReportConfigs.iterator().next();
if (!metadataReportConfig.isValid()) {
return;
}
configManager.addRegistries(tmpRegistries);
// protocol ids to protocol configs
List<ProtocolConfig> tmpProtocols = new ArrayList<>();
Set<String> protocolIds = configManager.getProtocolIds();
protocolIds.forEach(id -> {
if (tmpProtocols.stream().noneMatch(prot -> prot.getId().equals(id))) {
tmpProtocols.add(configManager.getProtocol(id).orElseGet(() -> {
ProtocolConfig protocolConfig = new ProtocolConfig();
protocolConfig.setId(id);
protocolConfig.refresh();
return protocolConfig;
}));
}
});
configManager.addProtocols(tmpProtocols);
MetadataReportInstance.init(metadataReportConfig.toUrl());
}
/**
@ -537,8 +550,65 @@ public class DubboBootstrap implements Lifecycle {
startConfigCenter();
}
private void loadRemoteConfigs() {
// registry ids to registry configs
List<RegistryConfig> tmpRegistries = new ArrayList<>();
Set<String> registryIds = configManager.getRegistryIds();
registryIds.forEach(id -> {
if (tmpRegistries.stream().noneMatch(reg -> reg.getId().equals(id))) {
tmpRegistries.add(configManager.getRegistry(id).orElseGet(() -> {
RegistryConfig registryConfig = new RegistryConfig();
registryConfig.setId(id);
registryConfig.refresh();
return registryConfig;
}));
}
});
configManager.addRegistries(tmpRegistries);
// protocol ids to protocol configs
List<ProtocolConfig> tmpProtocols = new ArrayList<>();
Set<String> protocolIds = configManager.getProtocolIds();
protocolIds.forEach(id -> {
if (tmpProtocols.stream().noneMatch(prot -> prot.getId().equals(id))) {
tmpProtocols.add(configManager.getProtocol(id).orElseGet(() -> {
ProtocolConfig protocolConfig = new ProtocolConfig();
protocolConfig.setId(id);
protocolConfig.refresh();
return protocolConfig;
}));
}
});
configManager.addProtocols(tmpProtocols);
}
/**
* Initialize {@link MetadataService} from {@link WritableMetadataService}'s extension
*/
private void initMetadataService() {
this.metadataService = getExtension(isDefaultMetadataStorageType());
}
/**
* Initialize {@link MetadataServiceExporter}
*/
private void initMetadataServiceExporter() {
this.metadataServiceExporter = new ConfigurableMetadataServiceExporter(metadataService);
}
/**
* Initialize {@link EventListener}
*/
private void initEventListener() {
// Add current instance into listeners
addEventListener(this);
}
private Collection<ServiceDiscovery> getServiceDiscoveries() {
return AbstractServiceDiscoveryFactory.getDiscoveries();
return serviceDiscoveries;
}
/**
@ -631,8 +701,8 @@ public class DubboBootstrap implements Lifecycle {
public boolean isStarted() {
return started;
}
/* serve for builder apis, begin */
private ApplicationBuilder createApplicationBuilder(String name) {
return new ApplicationBuilder().name(name);
}
@ -660,43 +730,7 @@ public class DubboBootstrap implements Lifecycle {
private ConsumerBuilder createConsumerBuilder(String id) {
return new ConsumerBuilder().id(id);
}
/* serve for builder apis, end */
private void startMetadataReport() {
ApplicationConfig applicationConfig = configManager.getApplication().orElseThrow(
() -> new IllegalStateException("There's no ApplicationConfig specified.")
);
String metadataType = applicationConfig.getMetadata();
// FIXME, multiple metadata config support.
Collection<MetadataReportConfig> metadataReportConfigs = configManager.getMetadataConfigs();
if (CollectionUtils.isEmpty(metadataReportConfigs)) {
if (METADATA_REMOTE.equals(metadataType)) {
throw new IllegalStateException("No MetadataConfig found, you must specify the remote Metadata Center address when set 'metadata=remote'.");
}
return;
}
MetadataReportConfig metadataReportConfig = metadataReportConfigs.iterator().next();
if (!metadataReportConfig.isValid()) {
return;
}
MetadataReportInstance.init(metadataReportConfig.toUrl());
}
private void startConfigCenter() {
Collection<ConfigCenterConfig> configCenters = configManager.getConfigCenters();
if (CollectionUtils.isNotEmpty(configCenters)) {
CompositeDynamicConfiguration compositeDynamicConfiguration = new CompositeDynamicConfiguration();
for (ConfigCenterConfig configCenter : configCenters) {
configCenter.refresh();
compositeDynamicConfiguration.addConfiguration(prepareEnvironment(configCenter));
}
Environment.getInstance().setDynamicConfiguration(compositeDynamicConfiguration);
}
configManager.refreshAll();
}
private DynamicConfiguration prepareEnvironment(ConfigCenterConfig configCenter) {
if (configCenter.isValid()) {
@ -738,12 +772,10 @@ public class DubboBootstrap implements Lifecycle {
}
/**
* export {@link MetadataService} and get the exported {@link URL URLs}
*
* @return {@link MetadataServiceExporter#getExportedURLs()}
* export {@link MetadataService}
*/
private List<URL> exportMetadataService() {
return metadataServiceExporter.export().getExportedURLs();
private void exportMetadataService() {
metadataServiceExporter.export();
}
private void unexportMetadataService() {
@ -777,7 +809,7 @@ public class DubboBootstrap implements Lifecycle {
int port = exportedURL.getPort();
ServiceInstance serviceInstance = initServiceInstance(serviceName, host, port);
ServiceInstance serviceInstance = createServiceInstance(serviceName, host, port);
getServiceDiscoveries().forEach(serviceDiscovery -> serviceDiscovery.register(serviceInstance));
}
@ -801,42 +833,15 @@ public class DubboBootstrap implements Lifecycle {
return selectedURL;
}
/**
* Use rest protocol if there's one, otherwise, choose the first one available.
*
* @return
*/
private String findOneProtocolForServiceInstance(Set<String> protocols) {
String result = null;
for (String protocol : protocols) {
if ("rest".equalsIgnoreCase(protocol)) {
result = protocol;
break;
}
}
if (result == null) {
for (String protocol : protocols) {
if (!"injvm".equalsIgnoreCase(protocol) && "registry".equalsIgnoreCase(protocol)) {
result = protocol;
break;
}
}
}
return result;
}
private void unregisterServiceInstance() {
if (serviceInstance != null) {
getServiceDiscoveries().forEach(serviceDiscovery -> {
serviceDiscovery.unregister(serviceInstance);
});
}
}
private ServiceInstance initServiceInstance(String serviceName, String host, int port) {
private ServiceInstance createServiceInstance(String serviceName, String host, int port) {
this.serviceInstance = new DefaultServiceInstance(serviceName, host, port);
setMetadataStorageType(serviceInstance, isDefaultMetadataStorageType());
return this.serviceInstance;
@ -850,6 +855,8 @@ public class DubboBootstrap implements Lifecycle {
destroyReferences();
destroyServiceDiscoveries();
clear();
release();
@ -871,8 +878,20 @@ public class DubboBootstrap implements Lifecycle {
}
}
private void destroyServiceDiscoveries() {
getServiceDiscoveries().forEach(serviceDiscovery -> {
execute(() -> {
serviceDiscovery.destroy();
});
});
if (logger.isDebugEnabled()) {
logger.debug(NAME + "'s all ServiceDiscoveries have been destroyed.");
}
}
private void clear() {
clearConfigs();
clearServiceDiscoveries();
}
private void clearConfigs() {
@ -882,6 +901,13 @@ public class DubboBootstrap implements Lifecycle {
}
}
private void clearServiceDiscoveries() {
serviceDiscoveries.clear();
if (logger.isDebugEnabled()) {
logger.debug(NAME + "'s serviceDiscoveries have been clear.");
}
}
private void release() {
executeMutually(() -> {
while (awaited.compareAndSet(false, true)) {

2
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceConsumerBootstrap.java
View File

@ -31,10 +31,12 @@ public class DubboServiceConsumerBootstrap {
new DubboBootstrap()
.application("dubbo-consumer-demo")
.protocol(builder -> builder.port(20887).name("dubbo"))
// Zookeeper
.registry("zookeeper", builder -> builder.address("zookeeper://127.0.0.1:2181?registry.type=service&subscribed.services=dubbo-provider-demo"))
// .metadataReport(new MetadataReportConfig("zookeeper://127.0.0.1:2181"))
// Nacos
// .registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service&subscribed.services=dubbo-provider-demo"))
// .registry("consul", builder -> builder.address("consul://127.0.0.1:8500?registry.type=service&subscribed.services=dubbo-provider-demo").group("namespace1"))
.reference("echo", builder -> builder.interfaceClass(EchoService.class).protocol("dubbo"))
.reference("user", builder -> builder.interfaceClass(UserService.class).protocol("rest"))

6
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProviderBootstrap.java
View File

@ -25,10 +25,6 @@ import org.apache.dubbo.config.RegistryConfig;
import org.apache.dubbo.config.ServiceConfig;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import static org.apache.dubbo.common.constants.CommonConstants.METADATA_KEY;
/**
* Dubbo Provider Bootstrap
@ -67,7 +63,7 @@ public class DubboServiceProviderBootstrap {
// userService.setRegistries(Arrays.asList(interfaceRegistry, serviceRegistry));
ApplicationConfig applicationConfig = new ApplicationConfig("dubbo-provider-demo");
applicationConfig.setMetadata("remote");
applicationConfig.setMetadataStorageType("remote");
new DubboBootstrap()
.application(applicationConfig)
// Zookeeper in service registry type

60
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/EtcdDubboServiceConsumerBootstrap.java Normal file
View File

@ -0,0 +1,60 @@
/*
* 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.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.config.MetadataReportConfig;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.context.ConfigManager;
/**
* Dubbo Provider Bootstrap
*
* @since 2.7.4
*/
public class EtcdDubboServiceConsumerBootstrap {
public static void main(String[] args) throws Exception {
new DubboBootstrap()
.application("dubbo-consumer-demo")
.defaultMetadataStorageType(true)
// Zookeeper
.protocol(builder -> builder.port(20887).name("dubbo"))
.registry("zookeeper", builder -> builder.address("etcd3://127.0.0.1:2379?registry.type=service&subscribed.services=dubbo-provider-demo"))
.metadataReport(new MetadataReportConfig("etcd://127.0.0.1:2379"))
// Nacos
// .registry("consul", builder -> builder.address("consul://127.0.0.1:8500?registry.type=service&subscribed.services=dubbo-provider-demo").group("namespace1"))
.reference("echo", builder -> builder.interfaceClass(EchoService.class).protocol("dubbo"))
.reference("user", builder -> builder.interfaceClass(UserService.class).protocol("rest"))
.onlyRegisterProvider(true)
.start()
.await();
ConfigManager configManager = ConfigManager.getInstance();
ReferenceConfig<EchoService> referenceConfig = configManager.getReference("echo");
EchoService echoService = referenceConfig.get();
for (int i = 0; i < 500; i++) {
Thread.sleep(2000L);
System.out.println(echoService.echo("Hello,World"));
}
}
}

96
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/EtcdDubboServiceProviderBootstrap.java Normal file
View File

@ -0,0 +1,96 @@
/*
* 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.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.bootstrap.rest.UserServiceImpl;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.MetadataReportConfig;
import org.apache.dubbo.config.ProtocolConfig;
import org.apache.dubbo.config.RegistryConfig;
import org.apache.dubbo.config.ServiceConfig;
import java.util.Arrays;
/**
* Dubbo Provider Bootstrap
*
* @since 2.7.4
*/
public class EtcdDubboServiceProviderBootstrap {
public static void main(String[] args) {
multipleRegistries();
}
private static void multipleRegistries() {
ProtocolConfig restProtocol = new ProtocolConfig();
restProtocol.setName("rest");
restProtocol.setId("rest");
restProtocol.setPort(-1);
RegistryConfig interfaceRegistry = new RegistryConfig();
interfaceRegistry.setId("interfaceRegistry");
interfaceRegistry.setAddress("etcd3://127.0.0.1:2379");
RegistryConfig serviceRegistry = new RegistryConfig();
serviceRegistry.setId("serviceRegistry");
serviceRegistry.setAddress("etcd3://127.0.0.1:2379?registry.type=service");
ServiceConfig<EchoService> echoService = new ServiceConfig<>();
echoService.setInterface(EchoService.class.getName());
echoService.setRef(new EchoServiceImpl());
// echoService.setRegistries(Arrays.asList(interfaceRegistry, serviceRegistry));
ServiceConfig<UserService> userService = new ServiceConfig<>();
userService.setInterface(UserService.class.getName());
userService.setRef(new UserServiceImpl());
userService.setProtocol(restProtocol);
// userService.setRegistries(Arrays.asList(interfaceRegistry, serviceRegistry));
ApplicationConfig applicationConfig = new ApplicationConfig("dubbo-provider-demo");
applicationConfig.setMetadataStorageType("remote");
new DubboBootstrap()
.application(applicationConfig)
.defaultMetadataStorageType(true)
// Zookeeper in service registry type
// .registry("zookeeper", builder -> builder.address("zookeeper://127.0.0.1:2181?registry.type=service"))
// Nacos
// .registry("zookeeper", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service"))
.registries(Arrays.asList(interfaceRegistry, serviceRegistry))
// .registry(RegistryBuilder.newBuilder().address("consul://127.0.0.1:8500?registry.type=service").build())
.protocol(builder -> builder.port(-1).name("dubbo"))
.metadataReport(new MetadataReportConfig("etcd://127.0.0.1:2379"))
.service(echoService)
.service(userService)
.start()
.await();
}
private static void testSCCallDubbo() {
}
private static void testDubboCallSC() {
}
private static void testDubboTansormation() {
}
}

58
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceConsumerBootstrap.java Normal file
View File

@ -0,0 +1,58 @@
/*
* 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.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.config.MetadataReportConfig;
import org.apache.dubbo.config.ReferenceConfig;
import org.apache.dubbo.config.context.ConfigManager;
/**
* Dubbo Provider Bootstrap
*
* @since 2.7.4
*/
public class NacosDubboServiceConsumerBootstrap {
public static void main(String[] args) throws Exception {
new DubboBootstrap()
.application("dubbo-consumer-demo")
// Zookeeper
.registry("zookeeper", builder -> builder.address("zookeeper://127.0.0.1:2181?registry.type=service&subscribed.services=dubbo-provider-demo"))
.metadataReport(new MetadataReportConfig("zookeeper://127.0.0.1:2181"))
// Nacos
// .registry("consul", builder -> builder.address("consul://127.0.0.1:8500?registry.type=service&subscribed.services=dubbo-provider-demo").group("namespace1"))
.reference("echo", builder -> builder.interfaceClass(EchoService.class).protocol("dubbo"))
.reference("user", builder -> builder.interfaceClass(UserService.class).protocol("rest"))
.onlyRegisterProvider(true)
.start()
.await();
ConfigManager configManager = ConfigManager.getInstance();
ReferenceConfig<EchoService> referenceConfig = configManager.getReference("echo");
EchoService echoService = referenceConfig.get();
for (int i = 0; i < 500; i++) {
Thread.sleep(2000L);
System.out.println(echoService.echo("Hello,World"));
}
}
}

5
dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/DubboServiceProvider2Bootstrap.java → dubbo-bootstrap/src/test/java/org/apache/dubbo/bootstrap/NacosDubboServiceProviderBootstrap.java
View File

@ -18,22 +18,25 @@ package org.apache.dubbo.bootstrap;
import org.apache.dubbo.bootstrap.rest.UserService;
import org.apache.dubbo.bootstrap.rest.UserServiceImpl;
import org.apache.dubbo.config.MetadataReportConfig;
/**
* Dubbo Provider Bootstrap
*
* @since 2.7.4
*/
public class DubboServiceProvider2Bootstrap {
public class NacosDubboServiceProviderBootstrap {
public static void main(String[] args) {
new DubboBootstrap()
.defaultMetadataStorageType(false)
.application("dubbo-provider-demo")
// Zookeeper in service registry type
.registry("zookeeper", builder -> builder.address("zookeeper://127.0.0.1:2181?registry.type=service"))
// Nacos
// .registry("nacos", builder -> builder.address("nacos://127.0.0.1:8848?registry.type=service"))
// .registry(RegistryBuilder.newBuilder().address("etcd3://127.0.0.1:2379?registry.type=service").build())
.metadataReport(new MetadataReportConfig("zookeeper://127.0.0.1:2181"))
.protocol("dubbo", builder -> builder.port(20885).name("dubbo"))
.protocol("rest", builder -> builder.port(9090).name("rest"))
.service(builder -> builder.id("echo").interfaceClass(EchoService.class).ref(new EchoServiceImpl()).protocolIds("dubbo"))

11
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ApplicationConfig.java
View File

@ -150,7 +150,7 @@ public class ApplicationConfig extends AbstractConfig {
/**
* Metadata type, local or remote, if choose remote, you need to further specify metadata center.
*/
private String metadata;
private String metadataStorageType;
public ApplicationConfig() {
}
@ -334,6 +334,7 @@ public class ApplicationConfig extends AbstractConfig {
/**
* The format is the same as the springboot, including: getQosEnableCompatible(), getQosPortCompatible(), getQosAcceptForeignIpCompatible().
*
* @return
*/
@Parameter(key = QOS_ENABLE_COMPATIBLE, excluded = true)
@ -387,12 +388,12 @@ public class ApplicationConfig extends AbstractConfig {
return !StringUtils.isEmpty(name);
}
public String getMetadata() {
return metadata;
public String getMetadataStorageType() {
return metadataStorageType;
}
public void setMetadata(String metadata) {
this.metadata = metadata;
public void setMetadataStorageType(String metadataStorageType) {
this.metadataStorageType = metadataStorageType;
}
@Override

2
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/builders/ApplicationBuilder.java
View File

@ -177,7 +177,7 @@ public class ApplicationBuilder extends AbstractBuilder<ApplicationConfig, Appli
super.build(config);
config.setName(name);
config.setMetadata(metadata);
config.setMetadataStorageType(metadata);
config.setVersion(this.version);
config.setOwner(this.owner);
config.setOrganization(this.organization);

67
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporter.java
View File

@ -19,7 +19,6 @@ package org.apache.dubbo.config.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.config.AbstractConfig;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.ProtocolConfig;
import org.apache.dubbo.config.RegistryConfig;
@ -28,14 +27,15 @@ import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.MetadataServiceExporter;
import java.util.Collection;
import java.util.LinkedList;
import java.util.ArrayList;
import java.util.List;
import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
/**
* {@link MetadataServiceExporter} implementation based on {@link AbstractConfig Dubbo configurations}, the clients
* {@link MetadataServiceExporter} implementation based on {@link ConfigManager Dubbo configurations}, the clients
* should make sure the {@link ApplicationConfig}, {@link RegistryConfig} and {@link ProtocolConfig} are ready before
* {@link #export()}.
* <p>
@ -53,38 +53,15 @@ public class ConfigurableMetadataServiceExporter implements MetadataServiceExpor
private final Logger logger = LoggerFactory.getLogger(getClass());
private ApplicationConfig applicationConfig;
private final ConfigManager configManager;
private List<RegistryConfig> registries = new LinkedList<>();
private List<ProtocolConfig> protocols = new LinkedList<>();
private MetadataService metadataService;
private final MetadataService metadataService;
private ServiceConfig<MetadataService> serviceConfig;
public ConfigurableMetadataServiceExporter setApplicationConfig(ApplicationConfig applicationConfig) {
this.applicationConfig = applicationConfig;
return this;
}
public ConfigurableMetadataServiceExporter setRegistries(Collection<RegistryConfig> registries) {
this.registries.clear();
this.registries.addAll(registries);
return this;
}
public ConfigurableMetadataServiceExporter setProtocols(Collection<ProtocolConfig> protocols) {
this.protocols.clear();
// TODO only support "dubbo" protocol, add more in the future
protocols.stream().filter(protocolConfig -> "dubbo".equals(protocolConfig.getName()))
.forEach(this.protocols::add);
return this;
}
public ConfigurableMetadataServiceExporter metadataService(MetadataService metadataService) {
public ConfigurableMetadataServiceExporter(MetadataService metadataService) {
this.configManager = ConfigManager.getInstance();
this.metadataService = metadataService;
return this;
}
@Override
@ -93,12 +70,12 @@ public class ConfigurableMetadataServiceExporter implements MetadataServiceExpor
if (!isExported()) {
ServiceConfig<MetadataService> serviceConfig = new ServiceConfig<>();
serviceConfig.setApplication(applicationConfig);
serviceConfig.setRegistries(registries);
serviceConfig.setProtocols(protocols);
serviceConfig.setApplication(getApplicationConfig());
serviceConfig.setRegistries(getRegistries());
serviceConfig.setProtocols(getProtocols());
serviceConfig.setInterface(MetadataService.class);
serviceConfig.setRef(metadataService);
serviceConfig.setGroup(applicationConfig.getName());
serviceConfig.setGroup(getApplicationConfig().getName());
serviceConfig.setVersion(metadataService.version());
// export
@ -135,4 +112,24 @@ public class ConfigurableMetadataServiceExporter implements MetadataServiceExpor
public boolean isExported() {
return serviceConfig != null && serviceConfig.isExported();
}
private ApplicationConfig getApplicationConfig() {
return configManager.getApplication().get();
}
private List<RegistryConfig> getRegistries() {
return new ArrayList<>(configManager.getRegistries());
}
private List<ProtocolConfig> getProtocols() {
return asList(getDefaultProtocol());
}
private ProtocolConfig getDefaultProtocol() {
ProtocolConfig defaultProtocol = new ProtocolConfig();
defaultProtocol.setName(DUBBO);
// auto-increment port
defaultProtocol.setPort(-1);
return defaultProtocol;
}
}

4
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/DubboProviderBootstrap.java
View File

@ -26,6 +26,8 @@ import org.apache.dubbo.registry.client.DefaultServiceInstance;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceDiscoveryFactory;
import static org.apache.dubbo.metadata.WritableMetadataService.getDefaultExtension;
/**
* Dubbo Provider Bootstrap
*/
@ -60,7 +62,7 @@ public class DubboProviderBootstrap {
// 暴露及注册服务
service.export();
MetadataServiceExporter exporter = new ConfigurableMetadataServiceExporter();
MetadataServiceExporter exporter = new ConfigurableMetadataServiceExporter(getDefaultExtension());
// 暴露 MetadataService 服务
exporter.export();

4
dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/metadata/ConfigurableMetadataServiceExporterTest.java
View File

@ -22,6 +22,7 @@ import org.apache.dubbo.config.ProtocolConfig;
import org.apache.dubbo.config.RegistryConfig;
import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.store.InMemoryWritableMetadataService;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@ -69,8 +70,7 @@ public class ConfigurableMetadataServiceExporterTest {
@Test
public void testExportAndUnexport() {
ConfigurableMetadataServiceExporter exporter = new ConfigurableMetadataServiceExporter();
exporter.setApplicationConfig(ConfigManager.getInstance().getApplication().get());
ConfigurableMetadataServiceExporter exporter = new ConfigurableMetadataServiceExporter(new InMemoryWritableMetadataService());
List<URL> urls = exporter.export().getExportedURLs();
assertEquals(1, urls.size());

5
dubbo-dependencies-bom/pom.xml
View File

@ -428,6 +428,11 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.etcd</groupId>
<artifactId>jetcd-launcher</artifactId>
<version>${jetcd_version}</version>
</dependency>
<!-- Log libs -->
<dependency>
<groupId>org.slf4j</groupId>

8
dubbo-registry/dubbo-registry-api/pom.xml
View File

@ -31,11 +31,19 @@
<skip_maven_deploy>false</skip_maven_deploy>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-common</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-config-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-cluster</artifactId>

43
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
View File

@ -17,51 +17,26 @@
package org.apache.dubbo.registry.client;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* Abstract {@link ServiceDiscoveryFactory} implementation with cache, the subclass
* should implement {@link #createDiscovery(URL)} method to create an instance of {@link ServiceDiscovery}
*
* @see ServiceDiscoveryFactory
* @since 2.7.4
*/
public abstract class AbstractServiceDiscoveryFactory implements ServiceDiscoveryFactory {
private static final Logger logger = LoggerFactory.getLogger(AbstractServiceDiscoveryFactory.class);
private final ConcurrentMap<String, ServiceDiscovery> discoveries = new ConcurrentHashMap<>();
private static ConcurrentMap<String, ServiceDiscovery> discoveries = new ConcurrentHashMap<>();
public static Collection<ServiceDiscovery> getDiscoveries() {
return Collections.unmodifiableCollection(discoveries.values());
}
/**
* Close all created registries
*/
public static void destroyAll() {
if (logger.isInfoEnabled()) {
logger.info("Closing all ServiceDiscovery instances: " + getDiscoveries());
}
for (ServiceDiscovery discovery : getDiscoveries()) {
try {
discovery.destroy();
} catch (Throwable e) {
logger.error("Error trying to close ServiceDiscovery instance.", e);
}
}
discoveries.clear();
}
/**
* @param registryURL "zookeeper://ip:port/RegistryService?xxx"
* @return
*/
@Override
public ServiceDiscovery getServiceDiscovery(URL registryURL) {
String key = registryURL.toServiceStringWithoutResolving();
return discoveries.computeIfAbsent(key, k -> createDiscovery(registryURL));
}
protected abstract ServiceDiscovery createDiscovery(URL url);
protected abstract ServiceDiscovery createDiscovery(URL registryURL);
}

17
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceDiscoveryFactory.java
View File

@ -34,20 +34,15 @@ import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoad
*/
public class DefaultServiceDiscoveryFactory extends AbstractServiceDiscoveryFactory {
@Override
protected ServiceDiscovery createDiscovery(URL connectionURL) {
ServiceDiscovery serviceDiscovery = load(connectionURL);
return new EventPublishingServiceDiscovery(serviceDiscovery);
}
/**
* Load the {@link ServiceDiscovery} by {@link URL#getProtocol() the protocol} from {@link URL connection URL}
* Create the {@link ServiceDiscovery} by {@link URL#getProtocol() the protocol} from {@link URL connection URL}
*
* @param connectionURL the {@link URL url} to connect
* @return non-null
* @param registryURL
* @return
*/
private ServiceDiscovery load(URL connectionURL) {
String protocol = connectionURL.getProtocol();
@Override
protected ServiceDiscovery createDiscovery(URL registryURL) {
String protocol = registryURL.getProtocol();
ExtensionLoader<ServiceDiscovery> loader = getExtensionLoader(ServiceDiscovery.class);
return loader.getOrDefaultExtension(protocol);
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/EventPublishingServiceDiscovery.java
View File

@ -235,9 +235,9 @@ final class EventPublishingServiceDiscovery implements ServiceDiscovery {
}
executeWithEvents(
of(new ServiceDiscoveryInitializingEvent(serviceDiscovery)),
of(new ServiceDiscoveryInitializingEvent(this, serviceDiscovery)),
() -> serviceDiscovery.initialize(registryURL),
of(new ServiceDiscoveryInitializedEvent(serviceDiscovery))
of(new ServiceDiscoveryInitializedEvent(this, serviceDiscovery))
);
// doesn't start -> started
@ -257,9 +257,9 @@ final class EventPublishingServiceDiscovery implements ServiceDiscovery {
}
executeWithEvents(
of(new ServiceDiscoveryDestroyingEvent(serviceDiscovery)),
of(new ServiceDiscoveryDestroyingEvent(this, serviceDiscovery)),
serviceDiscovery::destroy,
of(new ServiceDiscoveryDestroyedEvent(serviceDiscovery))
of(new ServiceDiscoveryDestroyedEvent(this, serviceDiscovery))
);
// doesn't stop -> stopped
@ -273,7 +273,7 @@ final class EventPublishingServiceDiscovery implements ServiceDiscovery {
try {
action.execute();
} catch (Exception e) {
dispatchEvent(new ServiceDiscoveryExceptionEvent(serviceDiscovery, e));
dispatchEvent(new ServiceDiscoveryExceptionEvent(this, serviceDiscovery, e));
}
afterEvent.ifPresent(this::dispatchEvent);
}

41
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistry.java
View File

@ -103,7 +103,7 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
public ServiceDiscoveryRegistry(URL registryURL) {
super(registryURL);
this.serviceDiscovery = getServiceDiscovery(registryURL);
this.serviceDiscovery = createServiceDiscovery(registryURL);
this.subscribedServices = getSubscribedServices(registryURL);
this.serviceNameMapping = ServiceNameMapping.getDefaultExtension();
String metadataStorageType = getMetadataStorageType(registryURL);
@ -120,14 +120,14 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
}
/**
* Get the {@link ServiceDiscovery} from the connection {@link URL}
* Create the {@link ServiceDiscovery} from the registry {@link URL}
*
* @param registryURL the {@link URL} to connect the registry
* @return non-null
*/
protected ServiceDiscovery getServiceDiscovery(URL registryURL) {
ServiceDiscoveryFactory factory = getExtension(registryURL);
ServiceDiscovery serviceDiscovery = factory.getServiceDiscovery(registryURL);
protected ServiceDiscovery createServiceDiscovery(URL registryURL) {
ServiceDiscovery originalServiceDiscovery = getServiceDiscovery(registryURL);
ServiceDiscovery serviceDiscovery = enhanceEventPublishing(originalServiceDiscovery);
execute(() -> {
serviceDiscovery.initialize(registryURL.addParameter(INTERFACE_KEY, ServiceDiscovery.class.getName())
.removeParameter(REGISTRY_TYPE_KEY));
@ -135,6 +135,28 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
return serviceDiscovery;
}
/**
* Get the instance {@link ServiceDiscovery} from the registry {@link URL} using
* {@link ServiceDiscoveryFactory} SPI
*
* @param registryURL the {@link URL} to connect the registry
* @return
*/
private ServiceDiscovery getServiceDiscovery(URL registryURL) {
ServiceDiscoveryFactory factory = getExtension(registryURL);
return factory.getServiceDiscovery(registryURL);
}
/**
* Enhance the original {@link ServiceDiscovery} with event publishing feature
*
* @param original the original {@link ServiceDiscovery}
* @return {@link EventPublishingServiceDiscovery} instance
*/
private ServiceDiscovery enhanceEventPublishing(ServiceDiscovery original) {
return new EventPublishingServiceDiscovery(original);
}
protected boolean shouldRegister(URL providerURL) {
String side = providerURL.getParameter(SIDE_KEY);
@ -506,13 +528,4 @@ public class ServiceDiscoveryRegistry extends FailbackRegistry {
public static boolean supports(URL registryURL) {
return SERVICE_REGISTRY_TYPE.equalsIgnoreCase(registryURL.getParameter(REGISTRY_TYPE_KEY));
}
/**
* Get the instance of {@link ServiceDiscovery}
*
* @return non-null
*/
public ServiceDiscovery getServiceDiscovery() {
return serviceDiscovery;
}
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryDestroyedEvent.java
View File

@ -26,14 +26,8 @@ import org.apache.dubbo.registry.client.ServiceDiscovery;
*/
public class ServiceDiscoveryDestroyedEvent extends ServiceDiscoveryEvent {
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The instance of {@link ServiceDiscovery} as source
* @throws IllegalArgumentException if source is null.
*/
public ServiceDiscoveryDestroyedEvent(ServiceDiscovery serviceDiscovery) {
super(serviceDiscovery);
public ServiceDiscoveryDestroyedEvent(ServiceDiscovery source, ServiceDiscovery original) {
super(source, original);
}
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryDestroyingEvent.java
View File

@ -26,14 +26,8 @@ import org.apache.dubbo.registry.client.ServiceDiscovery;
*/
public class ServiceDiscoveryDestroyingEvent extends ServiceDiscoveryEvent {
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The instance of {@link ServiceDiscovery} as source
* @throws IllegalArgumentException if source is null.
*/
public ServiceDiscoveryDestroyingEvent(ServiceDiscovery serviceDiscovery) {
super(serviceDiscovery);
public ServiceDiscoveryDestroyingEvent(ServiceDiscovery source, ServiceDiscovery original) {
super(source, original);
}
}

26
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryEvent.java
View File

@ -28,14 +28,23 @@ import org.apache.dubbo.registry.client.ServiceDiscovery;
*/
public abstract class ServiceDiscoveryEvent extends Event {
private final ServiceDiscovery original;
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The {@link ServiceDiscovery} on which the Event initially occurred.
* @param source The object on which the Event initially occurred.
* @param original The original {@link ServiceDiscovery}
* @throws IllegalArgumentException if source is null.
*/
public ServiceDiscoveryEvent(ServiceDiscovery serviceDiscovery) {
super(serviceDiscovery);
public ServiceDiscoveryEvent(ServiceDiscovery source, ServiceDiscovery original) {
super(source);
this.original = original;
}
@Override
public ServiceDiscovery getSource() {
return (ServiceDiscovery) super.getSource();
}
/**
@ -44,6 +53,15 @@ public abstract class ServiceDiscoveryEvent extends Event {
* @return {@link ServiceDiscovery} instance
*/
public final ServiceDiscovery getServiceDiscovery() {
return (ServiceDiscovery) getSource();
return getSource();
}
/**
* Get the original {@link ServiceDiscovery}
*
* @return the original {@link ServiceDiscovery}
*/
public final ServiceDiscovery getOriginal() {
return original;
}
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryExceptionEvent.java
View File

@ -29,14 +29,8 @@ public class ServiceDiscoveryExceptionEvent extends ServiceDiscoveryEvent {
private final Exception cause;
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The {@link ServiceDiscovery} on which the Event initially occurred.
* @throws IllegalArgumentException if any argument is null.
*/
public ServiceDiscoveryExceptionEvent(ServiceDiscovery serviceDiscovery, Exception cause) {
super(serviceDiscovery);
public ServiceDiscoveryExceptionEvent(ServiceDiscovery source, ServiceDiscovery original, Exception cause) {
super(source, original);
if (cause == null) {
throw new NullPointerException("The cause of Exception must not null");
}

11
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryInitializedEvent.java
View File

@ -27,14 +27,7 @@ import org.apache.dubbo.registry.client.ServiceDiscovery;
*/
public class ServiceDiscoveryInitializedEvent extends ServiceDiscoveryEvent {
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The instance of {@link ServiceDiscovery} as source
* @throws IllegalArgumentException if source is null.
*/
public ServiceDiscoveryInitializedEvent(ServiceDiscovery serviceDiscovery) {
super(serviceDiscovery);
public ServiceDiscoveryInitializedEvent(ServiceDiscovery source, ServiceDiscovery original) {
super(source, original);
}
}

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/event/ServiceDiscoveryInitializingEvent.java
View File

@ -27,13 +27,7 @@ import org.apache.dubbo.registry.client.ServiceDiscovery;
*/
public class ServiceDiscoveryInitializingEvent extends ServiceDiscoveryEvent {
/**
* Constructs a prototypical Event.
*
* @param serviceDiscovery The instance of {@link ServiceDiscovery} as source
* @throws IllegalArgumentException if source is null.
*/
public ServiceDiscoveryInitializingEvent(ServiceDiscovery serviceDiscovery) {
super(serviceDiscovery);
public ServiceDiscoveryInitializingEvent(ServiceDiscovery source, ServiceDiscovery original) {
super(source, original);
}
}

12
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/RefreshServiceMetadataCustomizer.java
View File

@ -4,7 +4,7 @@ import org.apache.dubbo.metadata.WritableMetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.ServiceInstanceCustomizer;
import static org.apache.dubbo.metadata.WritableMetadataService.DEFAULT_EXTENSION;
import static org.apache.dubbo.metadata.WritableMetadataService.getExtension;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getExportedServicesRevision;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getMetadataStorageType;
import static org.apache.dubbo.registry.client.metadata.ServiceInstanceMetadataUtils.getSubscribedServicesRevision;
@ -22,12 +22,12 @@ public class RefreshServiceMetadataCustomizer implements ServiceInstanceCustomiz
@Override
public void customize(ServiceInstance serviceInstance) {
// FIXME to define the constant
String metadataStoredType = getMetadataStorageType(serviceInstance);
WritableMetadataService remoteWritableMetadataService =
WritableMetadataService.getExtension(metadataStoredType == null ? DEFAULT_EXTENSION : metadataStoredType);
remoteWritableMetadataService.refreshMetadata(getExportedServicesRevision(serviceInstance),
String metadataStoredType = getMetadataStorageType(serviceInstance);
WritableMetadataService writableMetadataService = getExtension(metadataStoredType);
writableMetadataService.refreshMetadata(getExportedServicesRevision(serviceInstance),
getSubscribedServicesRevision(serviceInstance));
}
}

22
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/ServiceInstanceMetadataUtils.java
View File

@ -17,6 +17,8 @@
package org.apache.dubbo.registry.client.metadata;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.ApplicationConfig;
import org.apache.dubbo.config.context.ConfigManager;
import org.apache.dubbo.metadata.MetadataService;
import org.apache.dubbo.metadata.WritableMetadataService;
import org.apache.dubbo.registry.client.ServiceInstance;
@ -187,10 +189,10 @@ public class ServiceInstanceMetadataUtils {
*
* @param registryURL the {@link URL} to connect the registry
* @return if not found in {@link URL#getParameters() parameters} of {@link URL registry URL}, return
* {@link WritableMetadataService#DEFAULT_METADATA_STORAGE_TYPE "default"}
* {@link #getDefaultMetadataStorageType()}
*/
public static String getMetadataStorageType(URL registryURL) {
return registryURL.getParameter(METADATA_STORAGE_TYPE_KEY, DEFAULT_METADATA_STORAGE_TYPE);
return registryURL.getParameter(METADATA_STORAGE_TYPE_KEY, getDefaultMetadataStorageType());
}
/**
@ -198,11 +200,23 @@ public class ServiceInstanceMetadataUtils {
*
* @param serviceInstance the specified {@link ServiceInstance}
* @return if not found in {@link ServiceInstance#getMetadata() metadata} of {@link ServiceInstance}, return
* {@link WritableMetadataService#DEFAULT_METADATA_STORAGE_TYPE "default"}
* {@link #getDefaultMetadataStorageType()}
*/
public static String getMetadataStorageType(ServiceInstance serviceInstance) {
Map<String, String> metadata = serviceInstance.getMetadata();
return metadata.getOrDefault(METADATA_STORAGE_TYPE_KEY, DEFAULT_METADATA_STORAGE_TYPE);
return metadata.getOrDefault(METADATA_STORAGE_TYPE_KEY, getDefaultMetadataStorageType());
}
/**
* Get the default Metadata storage type from {@link ApplicationConfig} if present, or
* {@link WritableMetadataService#DEFAULT_METADATA_STORAGE_TYPE "default"}
*
* @return non-null
*/
public static String getDefaultMetadataStorageType() {
return ConfigManager.getInstance().getApplication()
.map(ApplicationConfig::getMetadataStorageType)
.orElse(DEFAULT_METADATA_STORAGE_TYPE);
}
/**

8
dubbo-registry/dubbo-registry-api/src/test/java/org/apache/dubbo/registry/client/event/listener/LoggingEventListenerTest.java
View File

@ -59,10 +59,10 @@ public class LoggingEventListenerTest {
serviceDiscovery.initialize(connectionURL);
// ServiceDiscoveryStartingEvent
listener.onEvent(new ServiceDiscoveryInitializingEvent(serviceDiscovery));
listener.onEvent(new ServiceDiscoveryInitializingEvent(serviceDiscovery, serviceDiscovery));
// ServiceDiscoveryStartedEvent
listener.onEvent(new ServiceDiscoveryInitializedEvent(serviceDiscovery));
listener.onEvent(new ServiceDiscoveryInitializedEvent(serviceDiscovery, serviceDiscovery));
// ServiceInstancePreRegisteredEvent
listener.onEvent(new ServiceInstancePreRegisteredEvent(serviceDiscovery, createInstance()));
@ -80,9 +80,9 @@ public class LoggingEventListenerTest {
listener.onEvent(new ServiceInstanceUnregisteredEvent(serviceDiscovery, createInstance()));
// ServiceDiscoveryStoppingEvent
listener.onEvent(new ServiceDiscoveryDestroyingEvent(serviceDiscovery));
listener.onEvent(new ServiceDiscoveryDestroyingEvent(serviceDiscovery, serviceDiscovery));
// ServiceDiscoveryStoppedEvent
listener.onEvent(new ServiceDiscoveryDestroyedEvent(serviceDiscovery));
listener.onEvent(new ServiceDiscoveryDestroyedEvent(serviceDiscovery, serviceDiscovery));
}
}

26
dubbo-registry/dubbo-registry-etcd3/src/main/java/org/apache/dubbo/registry/etcd/EtcdServiceDiscovery.java
View File

@ -20,10 +20,12 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.event.EventDispatcher;
import org.apache.dubbo.event.EventListener;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.client.DefaultServiceInstance;
import org.apache.dubbo.registry.client.ServiceDiscovery;
import org.apache.dubbo.registry.client.ServiceInstance;
import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
@ -38,13 +40,17 @@ import org.apache.dubbo.rpc.RpcException;
import com.google.gson.Gson;
import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import static org.apache.dubbo.common.constants.RegistryConstants.DYNAMIC_KEY;
/**
* 2019-07-08
*/
@ -103,6 +109,7 @@ public class EtcdServiceDiscovery implements ServiceDiscovery, EventListener<Ser
try {
this.serviceInstance = serviceInstance;
String path = toPath(serviceInstance);
// etcdClient.createEphemeral(path);
etcdClient.putEphemeral(path, new Gson().toJson(serviceInstance));
services.add(serviceInstance.getServiceName());
} catch (Throwable e) {
@ -118,6 +125,10 @@ public class EtcdServiceDiscovery implements ServiceDiscovery, EventListener<Ser
+ ":" + serviceInstance.getPort();
}
String toParentPath(String serviceName) {
return root + File.separator + serviceName;
}
@Override
public void update(ServiceInstance serviceInstance) throws RuntimeException {
try {
@ -155,6 +166,21 @@ public class EtcdServiceDiscovery implements ServiceDiscovery, EventListener<Ser
dispatcher.addEventListener(listener);
}
@Override
public List<ServiceInstance> getInstances(String serviceName) {
List<String> children = etcdClient.getChildren(toParentPath(serviceName));
if (CollectionUtils.isEmpty(children)) {
return Collections.EMPTY_LIST;
}
List<ServiceInstance> list = new ArrayList<>(children.size());
for (String child : children) {
ServiceInstance serviceInstance = new Gson().fromJson(etcdClient.getKVValue(child), DefaultServiceInstance.class);
list.add(serviceInstance);
}
return list;
}
protected void registerServiceWatcher(String serviceName) {
String path = root + File.separator + serviceName;
/*

2
dubbo-registry/dubbo-registry-etcd3/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscovery
View File

@ -1 +1 @@
etcd=org.apache.dubbo.registry.etcd.EtcdServiceDiscovery
etcd3=org.apache.dubbo.registry.etcd.EtcdServiceDiscovery

15
dubbo-remoting/dubbo-remoting-etcd3/pom.xml
View File

@ -32,6 +32,7 @@
<description>The etcd3 remoting module of Dubbo project</description>
<properties>
<skip_maven_deploy>false</skip_maven_deploy>
<assertj.version>3.13.2</assertj.version>
</properties>
<dependencies>
<dependency>
@ -48,10 +49,24 @@
<groupId>io.etcd</groupId>
<artifactId>jetcd-core</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/io.etcd/jetcd-launcher -->
<dependency>
<groupId>io.etcd</groupId>
<artifactId>jetcd-launcher</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>${assertj.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

2
dubbo-remoting/dubbo-remoting-etcd3/src/main/java/org/apache/dubbo/remoting/etcd/jetcd/JEtcdClient.java
View File

@ -205,7 +205,7 @@ public class JEtcdClient extends AbstractEtcdClient<JEtcdClient.EtcdWatcher> {
@Override
public boolean putEphemeral(String key, String value) {
return clientWrapper.put(key, value);
return clientWrapper.putEphemeral(key, value);
}
public ManagedChannel getChannel() {

122
dubbo-remoting/dubbo-remoting-etcd3/src/test/java/org/apache/dubbo/remoting/etcd/jetcd/LeaseTest.java Normal file
View File

@ -0,0 +1,122 @@
package org.apache.dubbo.remoting.etcd.jetcd;
import com.google.common.base.Charsets;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.CloseableClient;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.Observers;
import io.etcd.jetcd.launcher.EtcdCluster;
import io.etcd.jetcd.launcher.EtcdClusterFactory;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.options.PutOption;
import io.grpc.stub.StreamObserver;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author cvictory ON 2019-08-16
*/
public class LeaseTest {
private static EtcdCluster cluster;
private KV kvClient;
private Client client;
private Lease leaseClient;
private static final ByteSequence KEY = ByteSequence.from("foo", Charsets.UTF_8);
private static final ByteSequence KEY_2 = ByteSequence.from("foo2", Charsets.UTF_8);
private static final ByteSequence VALUE = ByteSequence.from("bar", Charsets.UTF_8);
@BeforeAll
public static void beforeClass() {
cluster = EtcdClusterFactory.buildCluster("etcd-lease", 3, false);
cluster.start();
}
@AfterAll
public static void afterClass() {
cluster.close();
}
@BeforeEach
public void setUp() {
client = Client.builder().endpoints(cluster.getClientEndpoints()).build();
kvClient = client.getKVClient();
leaseClient = client.getLeaseClient();
}
@AfterEach
public void tearDown() {
if (client != null) {
client.close();
}
}
@Test
public void testGrant() throws Exception {
long leaseID = leaseClient.grant(5).get().getID();
kvClient.put(KEY, VALUE, PutOption.newBuilder().withLeaseId(leaseID).build()).get();
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(1);
Thread.sleep(6000);
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(0);
}
@Test
public void testRevoke() throws Exception {
long leaseID = leaseClient.grant(5).get().getID();
kvClient.put(KEY, VALUE, PutOption.newBuilder().withLeaseId(leaseID).build()).get();
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(1);
leaseClient.revoke(leaseID).get();
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(0);
}
@Test
public void testKeepAliveOnce() throws ExecutionException, InterruptedException {
long leaseID = leaseClient.grant(2).get().getID();
kvClient.put(KEY, VALUE, PutOption.newBuilder().withLeaseId(leaseID).build()).get();
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(1);
LeaseKeepAliveResponse rp = leaseClient.keepAliveOnce(leaseID).get();
assertThat(rp.getTTL()).isGreaterThan(0);
}
@Test
public void testKeepAlive() throws ExecutionException, InterruptedException {
long leaseID = leaseClient.grant(2).get().getID();
kvClient.put(KEY, VALUE, PutOption.newBuilder().withLeaseId(leaseID).build()).get();
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(1);
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<LeaseKeepAliveResponse> responseRef = new AtomicReference<>();
StreamObserver<LeaseKeepAliveResponse> observer = Observers.observer(response -> {
responseRef.set(response);
latch.countDown();
});
try (CloseableClient c = leaseClient.keepAlive(leaseID, observer)) {
latch.await(5, TimeUnit.SECONDS);
LeaseKeepAliveResponse response = responseRef.get();
assertThat(response.getTTL()).isGreaterThan(0);
}
Thread.sleep(3000);
assertThat(kvClient.get(KEY).get().getCount()).isEqualTo(0);
}
}