Skip to content

Commit

Permalink
Qos enhancement, take apache#2153 back to 2.6.x
Browse files Browse the repository at this point in the history
  • Loading branch information
diecui1202 committed Sep 6, 2018
1 parent 52e9290 commit 5656a0b
Show file tree
Hide file tree
Showing 3 changed files with 25 additions and 13 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -71,10 +71,13 @@ public class ApplicationConfig extends AbstractConfig {
// directory for saving thread dump
private String dumpDirectory;

// whether to enable qos or not
private Boolean qosEnable;

// the qos port to listen
private Integer qosPort;

// should we accept foreign ip or not?
private Boolean qosAcceptForeignIp;

// customized parameters
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.qos.common.QosConstants;
import com.alibaba.dubbo.qos.server.Server;
import com.alibaba.dubbo.rpc.Exporter;
import com.alibaba.dubbo.rpc.Invoker;
Expand All @@ -29,8 +32,12 @@
import static com.alibaba.dubbo.common.Constants.ACCEPT_FOREIGN_IP;
import static com.alibaba.dubbo.common.Constants.QOS_ENABLE;
import static com.alibaba.dubbo.common.Constants.QOS_PORT;
import static com.alibaba.dubbo.qos.common.QosConstants.DEFAULT_PORT;

public class QosProtocolWrapper implements Protocol {

private final Logger logger = LoggerFactory.getLogger(QosProtocolWrapper.class);

private static AtomicBoolean hasStarted = new AtomicBoolean(false);

private Protocol protocol;
Expand Down Expand Up @@ -76,20 +83,23 @@ private void startQosServer(URL url) {
}

try {
boolean qosEnable = Boolean.parseBoolean(url.getParameter(QOS_ENABLE,"true"));
boolean qosEnable = url.getParameter(QOS_ENABLE,true);
if (!qosEnable) {
logger.info("qos won't be started because it is disabled. " +
"Please check dubbo.application.qos.enable is configured either in system property, " +
"dubbo.properties or XML/spring boot configuration.");
return;
}

int port = Integer.parseInt(url.getParameter(QOS_PORT,"22222"));
boolean acceptForeignIp = Boolean.parseBoolean(url.getParameter(ACCEPT_FOREIGN_IP,"true"));
int port = url.getParameter(QOS_PORT, DEFAULT_PORT);
boolean acceptForeignIp = Boolean.parseBoolean(url.getParameter(ACCEPT_FOREIGN_IP,"false"));
Server server = com.alibaba.dubbo.qos.server.Server.getInstance();
server.setPort(port);
server.setAcceptForeignIp(acceptForeignIp);
server.start();

} catch (Throwable throwable) {
//throw new RpcException("fail to start qos server", throwable);
logger.warn("Fail to start qos server: ", throwable);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -135,26 +135,26 @@ public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcExceptio

//registry provider
final Registry registry = getRegistry(originInvoker);
final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);

//to judge to delay publish whether or not
boolean register = registedProviderUrl.getParameter("register", true);
boolean register = registeredProviderUrl.getParameter("register", true);

ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registedProviderUrl);
ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);

if (register) {
register(registryUrl, registedProviderUrl);
register(registryUrl, registeredProviderUrl);
ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
}

// Subscribe the override data
// FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
//Ensure that a new exporter instance is returned every time export
return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registedProviderUrl);
return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
}

@SuppressWarnings("unchecked")
Expand Down Expand Up @@ -219,18 +219,17 @@ private URL getRegistryUrl(Invoker<?> originInvoker) {
* @param originInvoker
* @return
*/
private URL getRegistedProviderUrl(final Invoker<?> originInvoker) {
private URL getRegisteredProviderUrl(final Invoker<?> originInvoker) {
URL providerUrl = getProviderUrl(originInvoker);
//The address you see at the registry
final URL registedProviderUrl = providerUrl.removeParameters(getFilteredKeys(providerUrl))
return providerUrl.removeParameters(getFilteredKeys(providerUrl))
.removeParameter(Constants.MONITOR_KEY)
.removeParameter(Constants.BIND_IP_KEY)
.removeParameter(Constants.BIND_PORT_KEY)
.removeParameter(QOS_ENABLE)
.removeParameter(QOS_PORT)
.removeParameter(ACCEPT_FOREIGN_IP)
.removeParameter(VALIDATION_KEY);
return registedProviderUrl;
}

private URL getSubscribedOverrideUrl(URL registedProviderUrl) {
Expand Down

0 comments on commit 5656a0b

Please sign in to comment.