Skip to content

Commit

Permalink
[Dubbo-6720] fix bug same interface unexport and export fail. also su…
Browse files Browse the repository at this point in the history
…pport hotload service (#6720)

* delegate exportmap & support hotload

* add header

* delete unused import & change test code

* delete unused import

* check codestyle

* endline format

* add code

* add res code exporters func

* DelegateExporterMap referencing a map

* fix test  DelegateExporterMap interface change
  • Loading branch information
owen200008 committed Jun 15, 2021
1 parent ba4d88e commit 222bf15
Show file tree
Hide file tree
Showing 10 changed files with 534 additions and 432 deletions.
@@ -1,116 +1,116 @@
/*
* 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.rpc.protocol;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.support.ProtocolUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

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

/**
* abstract ProtocolSupport.
*/
public abstract class AbstractProtocol implements Protocol {

protected final Logger logger = LoggerFactory.getLogger(getClass());

protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>();

/**
* <host:port, ProtocolServer>
*/
protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();

//TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();

protected static String serviceKey(URL url) {
int port = url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
return serviceKey(port, url.getPath(), url.getParameter(VERSION_KEY), url.getParameter(GROUP_KEY));
}

protected static String serviceKey(int port, String serviceName, String serviceVersion, String serviceGroup) {
return ProtocolUtils.serviceKey(port, serviceName, serviceVersion, serviceGroup);
}

public List<ProtocolServer> getServers() {
return Collections.unmodifiableList(new ArrayList<>(serverMap.values()));
}

@Override
public void destroy() {
for (Invoker<?> invoker : invokers) {
if (invoker != null) {
invokers.remove(invoker);
try {
if (logger.isInfoEnabled()) {
logger.info("Destroy reference: " + invoker.getUrl());
}
invoker.destroy();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
for (String key : new ArrayList<String>(exporterMap.keySet())) {
Exporter<?> exporter = exporterMap.remove(key);
if (exporter != null) {
try {
if (logger.isInfoEnabled()) {
logger.info("Unexport service: " + exporter.getInvoker().getUrl());
}
exporter.unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
}

@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
}

protected abstract <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException;

public Map<String, Exporter<?>> getExporterMap() {
return exporterMap;
}

public Collection<Exporter<?>> getExporters() {
return Collections.unmodifiableCollection(exporterMap.values());
}
}
/*
* 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.rpc.protocol;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProtocolServer;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.support.ProtocolUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

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

/**
* abstract ProtocolSupport.
*/
public abstract class AbstractProtocol implements Protocol {

protected final Logger logger = LoggerFactory.getLogger(getClass());

protected final DelegateExporterMap exporterMap = new DelegateExporterMap();

/**
* <host:port, ProtocolServer>
*/
protected final Map<String, ProtocolServer> serverMap = new ConcurrentHashMap<>();

//TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();

protected static String serviceKey(URL url) {
int port = url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
return serviceKey(port, url.getPath(), url.getParameter(VERSION_KEY), url.getParameter(GROUP_KEY));
}

protected static String serviceKey(int port, String serviceName, String serviceVersion, String serviceGroup) {
return ProtocolUtils.serviceKey(port, serviceName, serviceVersion, serviceGroup);
}

@Override
public List<ProtocolServer> getServers() {
return Collections.unmodifiableList(new ArrayList<>(serverMap.values()));
}

@Override
public void destroy() {
for (Invoker<?> invoker : invokers) {
if (invoker != null) {
invokers.remove(invoker);
try {
if (logger.isInfoEnabled()) {
logger.info("Destroy reference: " + invoker.getUrl());
}
invoker.destroy();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
for (Map.Entry<String, Exporter<?>> item : exporterMap.getExporterMap().entrySet()) {
if (exporterMap.removeExportMap(item.getKey(), item.getValue())) {
try {
if (logger.isInfoEnabled()) {
logger.info("Unexport service: " + item.getValue().getInvoker().getUrl());
}
item.getValue().unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
}

@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
}

protected abstract <T> Invoker<T> protocolBindingRefer(Class<T> type, URL url) throws RpcException;

public Map<String, Exporter<?>> getExporterMap() {
return exporterMap.getExporterMap();
}

public Collection<Exporter<?>> getExporters() {
return exporterMap.getExporters();
}
}
Expand Up @@ -77,7 +77,7 @@ public void setProxyFactory(ProxyFactory proxyFactory) {
@SuppressWarnings("unchecked")
public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
final String uri = serviceKey(invoker.getUrl());
Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
Exporter<T> exporter = (Exporter<T>) exporterMap.getExport(uri);
if (exporter != null) {
// When modifying the configuration through override, you need to re-expose the newly modified service.
if (Objects.equals(exporter.getInvoker().getUrl(), invoker.getUrl())) {
Expand All @@ -88,7 +88,7 @@ public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
exporter = new AbstractExporter<T>(invoker) {
@Override
public void afterUnExport() {
exporterMap.remove(uri);
exporterMap.removeExportMap(uri, this);
if (runnable != null) {
try {
runnable.run();
Expand All @@ -98,7 +98,7 @@ public void afterUnExport() {
}
}
};
exporterMap.put(uri, exporter);
exporterMap.addExportMap(uri, exporter);
return exporter;
}

Expand Down Expand Up @@ -277,5 +277,4 @@ public boolean isClosed() {
}
}


}
@@ -0,0 +1,86 @@
/*
* 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.rpc.protocol;

import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.Exporter;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
* delegate exportermap oper
*/
public class DelegateExporterMap {
protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>();

/**
* check is empty map
* @return
*/
public boolean isEmpty() {
return CollectionUtils.isEmptyMap(exporterMap);
}

/**
* get export
* @param key
* @return
*/
public Exporter<?> getExport(String key) {
return exporterMap.get(key);
}

/**
* add
* @param key
* @param exporter
*/
public void addExportMap(String key, Exporter<?> exporter) {
exporterMap.put(key, exporter);
}

/**
* delete
* @param key
*/
public boolean removeExportMap(String key, Exporter<?> exporter) {
Exporter<?> findExporter = exporterMap.get(key);
if(findExporter == exporter){
exporterMap.remove(key);
return true;
}
return false;
}

/**
* get the exports
* @return
*/
public Map<String, Exporter<?>> getExporterMap() {
return exporterMap;
}

/**
* get all exports
* @return
*/
public Collection<Exporter<?>> getExporters() {
return Collections.unmodifiableCollection(exporterMap.values());
}
}
Expand Up @@ -16,11 +16,9 @@
*/
package org.apache.dubbo.rpc.protocol.dubbo;

import org.apache.dubbo.rpc.Exporter;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.protocol.AbstractExporter;

import java.util.Map;
import org.apache.dubbo.rpc.protocol.DelegateExporterMap;

/**
* DubboExporter
Expand All @@ -29,17 +27,16 @@ public class DubboExporter<T> extends AbstractExporter<T> {

private final String key;

private final Map<String, Exporter<?>> exporterMap;
private final DelegateExporterMap delegateExporterMap;

public DubboExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {
public DubboExporter(Invoker<T> invoker, String key, DelegateExporterMap delegateExporterMap) {
super(invoker);
this.key = key;
this.exporterMap = exporterMap;
this.delegateExporterMap = delegateExporterMap;
}

@Override
public void afterUnExport() {
exporterMap.remove(key);
delegateExporterMap.removeExportMap(key, this);
}

}

0 comments on commit 222bf15

Please sign in to comment.