The introduction
What is remote exposure? Consider this scenario: Suppose we add A new server, A, dedicated to sending SMS alerts to selected users. So the question is, once our Message service goes live, how do we tell the calling server that server A provides Message functionality? So can we expose services that are currently provided in a place where callers know that a particular machine provides a particular function? With this assumption in mind, today we are going to talk about remote exposure of Dubbo service exposure!
Service remote exposure
Going back to the previous article, where we talked about the Export () method of ServiceConfig and analyzed the local exposure of the service, we’ll pick up where we left off and talk about remote exposure of service exposure.
// export to remote if the config is not local (export to local only when config is local)
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
...
if(! Constants.SCOPE_LOCAL.equalsIgnoreCase(scope)) {if (logger.isInfoEnabled()) {
logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
}
if(registryURLs ! = null && ! registryURLs.isEmpty()) {for(URL registryURL : registryURLs) { url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY)); // To help you read, omit some code... Invoker<? > invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); Exporter<? > exporter = protocol.export(wrapperInvoker); exporters.add(exporter); }}else{... }}... }Copy the codeHere we will focus only on the core code:
Invoker<? > invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); DelegateProviderMetaDataInvoker wrapperInvoker =new DelegateProviderMetaDataInvoker(invoker, this); Exporter<? > exporter = protocol.export(wrapperInvoker);Copy the codeBuild the Invoker object
Let’s start with how the Invoker object is created! This involves the Dubbo SPI mechanism, Calling process roughly StubProxyFactoryWrapper. GetInvoker () = = > JavassistProxyFactory. GetInvoker () look at the detail JavassistProxyFactory class GetInvoker method
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName, Class
[] parameterTypes, Object[] arguments) throws Throwable {
returnwrapper.invokeMethod(proxy, methodName, parameterTypes, arguments); }}; }Copy the codeIt’s worth focusing on the getWrapper() method of the Wrapper class!!
public static Wrapper getWrapper(Class
c) {
while (ClassGenerator.isDynamicClass(c)) // can not wrapper on dynamic class.
{
c = c.getSuperclass();
}
if (c == Object.class) {
return OBJECT_WRAPPER;
}
Wrapper ret = WRAPPER_MAP.get(c);
if (ret == null) {
ret = makeWrapper(c);
WRAPPER_MAP.put(c, ret);
}
return ret;
}
Copy the codeThe key value c is used to value from the WRAPPER_MAP cache. If there is no corresponding value, the makeWrapper() method is called to build a Wrapper class using Javassist technology. Assuming the value of the current parameter c is demoService, the resulting dynamic class is:
public class Wrapper0 extends Wrapper implements DC {
public static String[] pns;
public static Map pts;
public static String[] mns;
public static String[] dmns;
public static Class[] mts0;
public Wrapper0(a) {}public Class getPropertyType(String var1) {
return (Class)pts.get(var1);
}
public Object invokeMethod(Object var1, String var2, Class[] var3, Object[] var4) throws InvocationTargetException {
DemoService var5;
try {
var5 = (DemoService)var1;
} catch (Throwable var8) {
throw new IllegalArgumentException(var8);
}
try {
if("sayHello".equals(var2) && var3.length == 1) {
return var5.sayHello((String)var4[0]); }}catch (Throwable var9) {
throw new InvocationTargetException(var9);
}
throw new NoSuchMethodException("Not found method \"" + var2 + "\" in class org.apache.dubbo.demo.DemoService.");
}
public String[] getPropertyNames() {
return pns;
}
public Object getPropertyValue(Object var1, String var2) {
try {
DemoService var3 = (DemoService)var1;
} catch (Throwable var5) {
throw new IllegalArgumentException(var5);
}
throw new NoSuchPropertyException("Not found property \"" + var2 + "\" field or setter method in class org.apache.dubbo.demo.DemoService.");
}
public void setPropertyValue(Object var1, String var2, Object var3) {
try {
DemoService var4 = (DemoService)var1;
} catch (Throwable var6) {
throw new IllegalArgumentException(var6);
}
throw new NoSuchPropertyException("Not found property \"" + var2 + "\" field or setter method in class org.apache.dubbo.demo.DemoService.");
}
public String[] getDeclaredMethodNames() {
return dmns;
}
public boolean hasProperty(String var1) {
return pts.containsKey(var1);
}
public String[] getMethodNames() {
returnmns; }}Copy the codeFinally, back to the JavassistProxyFactory class’s getInvoker method, you can see that it actually returns an AbstractProxyInvoker object, When the AbstractProxyInvoker class’s doInvoke() method is called, the Wrapper class’s invokeMethod() method is actually called! This is very important! We’ll revisit this when we talk about Dubbo remote calls!
The build of an Exporter
Exporter<? > exporter = protocol.export(wrapperInvoker);Copy the codeTake a look at the second half of the code. The export() method of the RegistryProtocol class is finally called. If in doubt, see the Dubbo SPI mechanism in this article. Look directly at the RegistryProtocol export() method:
RegistryProtocol.export()
public class RegistryProtocol implements Protocol {
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
//export invoker
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
URL registryUrl = getRegistryUrl(originInvoker);
//registry provider
final Registry registry = getRegistry(originInvoker);
final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);
//to judge to delay publish whether or not
boolean register = registeredProviderUrl.getParameter("register".true);
ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);
if (register) {
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(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 newDestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl); }}Copy the codeThe registryProtocol.export () method is very important!! This is the core of service remote exposure. Without further ado, let’s take a look line by line!
doLocalExport()
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) {
ProviderUrl = originInvoker url.parameters; // Get originInvoker url.parameters
String key = getCacheKey(originInvoker);
ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
synchronized (bounds) {
exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
finalInvoker<? > invokerDelegete =new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
exporter = newExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker); bounds.put(key, exporter); }}}return exporter;
}
Copy the codeLet’s start with what the doLocalExport() method does:
- Obtained from the getCacheKey() method, the value corresponding to the key export is added to the parameters collection of the URL in the following code. And then we take the corresponding values here.
Invoker<? > invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));Copy the code- Try to fetch the exporter corresponding to the current key from the Bounds cache.
- If the cache is null, create a new exporter and return. The protoCL object here is Protocol$Adaptive. It is not hard to see that the final execution is actually DubboProtocol’s export() method.
To summarize: doLocalExport () using ExporterChangeableWrapper proxy class wraps the protocol. The export () method returns the exporter of object, the last on the bounds set in the cache.
DubboPrtocol.export()
DubboProtocol.java
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
// export service.
String key = serviceKey(url);
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
exporterMap.put(key, exporter);
//export an stub service for dispatching event
Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
if(isStubSupportEvent && ! isCallbackservice) { String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
if (logger.isWarnEnabled()) {
logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) +
"], has set stubproxy support event ,but no stub methods founded.")); }}else {
stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
}
}
openServer(url);
optimizeSerialization(url);
return exporter;
}
Copy the codeContinue with the dubboprotocol.export () method to create an exporter:
- The serviceKey() method is called to construct the serviceKey. The resulting key is in the form of group/path:version:port
- New DubboExporter
- OpenServer (URL), the URL is providerUrl passed by RegistryProtocol. The purpose of openServer() will be analyzed later.
- Dubboprotocol.export () returns an object called DubboExporter. Worth noting is the openServer() method behind!
openServer()
private void openServer(URL url) {
// find server.
String key = url.getAddress();
//client can export a service which's only for server to invoke
boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
if (isServer) {
ExchangeServer server = serverMap.get(key);
if (server == null) {
synchronized (this) {
server = serverMap.get(key);
if (server == null) { serverMap.put(key, createServer(url)); }}}else {
// server supports reset, use together with overrideserver.reset(url); }}}Copy the codeOpenServer () light from the method name looks like it opens the service connection. The method is relatively simple: take the URL address as the key and try to obtain the corresponding value from the serverMap. If value is null, the createServer(URL) method is called to create the server and add it to the serverMap. The createServer() method has a rather lengthy process. Here we use a sequence diagram to illustrate the internal call flow of the method:
The figure above omits the conversion process from ServiceConfig to RegistryProtocol and from RegistryProtocol to DubboProtocol. This part of the content involves the Dubbo SPI mechanism, if you have questions, please see: Dubbo source reading series. Here is a simple conversion process
- ServiceConfig RegistryProtocol: Protocol$Adaptive == “ProtocolFilterWrapper ==” ProtocolListenerWrapper == “RegistryProtocol
- RegistryProtocol to DubboProtocol Protocol$Adaptive == “ProtocolFilterWrapper ==” ProtocolListenerWrapper == “DubboProtocol
Finally, let’s focus on the NettyServer doOpen() method:
protected void doOpen(a) throws Throwable {
NettyHelper.setNettyLoggerFactory();
ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss".true));
ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker".true));
ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
bootstrap = new ServerBootstrap(channelFactory);
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
channels = nettyHandler.getChannels();
// https://issues.jboss.org/browse/NETTY-365
// https://issues.jboss.org/browse/NETTY-379
// final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
bootstrap.setOption("child.tcpNoDelay".true);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@Override
public ChannelPipeline getPipeline(a) {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
ChannelPipeline pipeline = Channels.pipeline();
/*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); } * /
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
returnpipeline; }});// bind
channel = bootstrap.bind(getBindAddress());
}
Copy the codeYou can see this code is more classic Netty server start code… This means that the openServer() method is used for Netty server startup. We know that Netty is often used for communication between clients and servers. Here we have the server enabled, so where will the corresponding client be enabled? What kind of interaction do they have with each other? We’ll save that for a follow-up article.
Exposure of services
With all this said, does it feel like remote exposure doesn’t have much to do with services? How exactly are our services perceived by other machines? How do people know that our machine provides SMS service? In fact, the curtain has already begun! Let’s continue. Recall the registryprotocol.export () method:
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
//export invoker
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
URL registryUrl = getRegistryUrl(originInvoker);
//registry provider
final Registry registry = getRegistry(originInvoker);
final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);
//to judge to delay publish whether or not
boolean register = registeredProviderUrl.getParameter("register".true);
ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);
if (register) {
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(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, registeredProviderUrl);
}
Copy the codeNow that we’ve finished talking about the doLocalExport() method, move on to the second half of the export() method:
RegistryProtocol.java
final Registry registry = getRegistry(originInvoker);
private Registry getRegistry(finalInvoker<? > originInvoker) {
URL registryUrl = getRegistryUrl(originInvoker);
return registryFactory.getRegistry(registryUrl);
}
Copy the codeThe registryFactory here is registryFactory $Adaptive (the Dubbo source code is riddled with numerous uses of the SPI extension mechanism, which will not be described here). In short, we obtained the extension class as ZookeeperRegistryFactory, which is derived from the AbstractRegistryFactory class. So you end up calling the getRegistry() method of the AbstractRegistryFactory class.
@Override
public Registry getRegistry(URL url) {
url = url.setPath(RegistryService.class.getName())
.addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
.removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
String key = url.toServiceStringWithoutResolving();
// Lock the registry access process to ensure a single instance of the registry
LOCK.lock();
try {
Registry registry = REGISTRIES.get(key);
if(registry ! =null) {
return registry;
}
registry = createRegistry(url);
if (registry == null) {
throw new IllegalStateException("Can not create registry " + url);
}
REGISTRIES.put(key, registry);
return registry;
} finally {
// Release the lockLOCK.unlock(); }}Copy the codeThe method is simpler, so let’s look directly at the key method createRegistry(URL). CreateRegistry () is an abstract method that calls concrete implementation methods based on the URL, which we analyze using the ZookeeperRegistryFactory class.
public class ZookeeperRegistryFactory extends AbstractRegistryFactory {...public Registry createRegistry(URL url) {
return newZookeeperRegistry(url, zookeeperTransporter); }... }public class ZookeeperRegistry extends FailbackRegistry {...public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
super(url);
if (url.isAnyHost()) {
throw new IllegalStateException("registry address == null");
}
String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
if(! group.startsWith(Constants.PATH_SEPARATOR)) { group = Constants.PATH_SEPARATOR + group; }this.root = group;
zkClient = zookeeperTransporter.connect(url);
zkClient.addStateListener(new StateListener() {
@Override
public void stateChanged(int state) {
if (state == RECONNECTED) {
try {
recover();
} catch(Exception e) { logger.error(e.getMessage(), e); }}}}); }... }Copy the codeThe createRegistry() method of the ZookeeperRegistryFactory class calls the constructor of the ZookeeperRegistry class to create a new ZookeeperRegistry instance and return. The constructor of the ZookeeperRegistry class calls the constructor of the parent FailbackRegistry class before performing subsequent operations. First look at the FailbackRegistry constructor:
public abstract class FailbackRegistry extends AbstractRegistry {...public FailbackRegistry(URL url) {
super(url);
this.retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run(a) {
// Check and connect to the registry
try {
// Delay retry
retry();
} catch (Throwable t) { // Defensive fault tolerance
logger.error("Unexpected error occur at failed retry, cause: "+ t.getMessage(), t); } } }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS); }... }Copy the codeThere is a delayed retry() method in the FailbackRegistry constructor, If the failed set failedRegistered, failedUnregistered, failedSubscribed, failedUnsubscribed, and failedNotified is not null, the retry operation is performed. Continuing with the constructor of the ZookeeperRegistry class:
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {... zkClient = zookeeperTransporter.connect(url); zkClient.addStateListener(new StateListener() {
@Override
public void stateChanged(int state) {
if (state == RECONNECTED) {
try {
recover();
} catch(Exception e) { logger.error(e.getMessage(), e); }}}}); }Copy the codeZookeeperTransporter here. The connect () through the SPI convert the actual call to CuratorZookeeperTransporter. The connect ().
public class CuratorZookeeperTransporter implements ZookeeperTransporter {
@Override
public ZookeeperClient connect(URL url) {
return newCuratorZookeeperClient(url); }}public class CuratorZookeeperClient extends AbstractZookeeperClient<CuratorWatcher> {
private final CuratorFramework client;
public CuratorZookeeperClient(URL url) {
super(url);
try {
int timeout = url.getParameter(Constants.TIMEOUT_KEY, 5000);
CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
.connectString(url.getBackupAddress())
.retryPolicy(new RetryNTimes(1.1000))
.connectionTimeoutMs(timeout);
String authority = url.getAuthority();
if(authority ! =null && authority.length() > 0) {
builder = builder.authorization("digest", authority.getBytes());
}
client = builder.build();
client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
@Override
public void stateChanged(CuratorFramework client, ConnectionState state) {
if (state == ConnectionState.LOST) {
CuratorZookeeperClient.this.stateChanged(StateListener.DISCONNECTED);
} else if (state == ConnectionState.CONNECTED) {
CuratorZookeeperClient.this.stateChanged(StateListener.CONNECTED);
} else if (state == ConnectionState.RECONNECTED) {
CuratorZookeeperClient.this.stateChanged(StateListener.RECONNECTED); }}}); client.start(); }catch (Exception e) {
throw newIllegalStateException(e.getMessage(), e); }}}Copy the codeThe above code creates an instance of the CuratorFramework using the CuratorFrameworkFactory factory class and starts the instance to create a connection to ZooKeeper.
Back to the getRegistry() method in the RegistryProtocol. We see that it eventually creates an instance to ZookeeperRegistry through layers of calls. The ziClient object in this example establishes a connection to ZooKeeper. We know that ZooKeeper is often used as a registry Ok. So now that we’re connected to ZooKeeper, should we write something to ZooKeeper? Keep reading, the show is coming!! ~
Register () Registration method
RegistryProtocol.java
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {...if (register) {
register(registryUrl, registeredProviderUrl);
ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true); }... }public void register(URL registryUrl, URL registedProviderUrl) {
Registry registry = registryFactory.getRegistry(registryUrl);
registry.register(registedProviderUrl);
}
Copy the codeHere the register() method ends up calling the FailbackRegistry class’s register() method (without going into why ！！！！) .
public abstract class FailbackRegistry extends AbstractRegistry {...public void register(URL url) {
super.register(url);
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
// Sending a registration request to the server side
doRegister(url);
} catch (Exception e) {
// ...}}... }public class ZookeeperRegistry extends FailbackRegistry {
protected void doRegister(URL url) {
try {
String str = toUrlPath(url);
zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
} catch (Throwable e) {
throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: "+ e.getMessage(), e); }}}Copy the codeUnderline the point!! DoRegister () method!! This is also a longer call chain. Draw a diagram to summarize:
Summary: The final goal is to create a path node on ZooKeeper that is generated by URL resolution. It looks something like this: Dubbo % % 2 f % 2 f10. 3 a 137.32.54%3 a20880%2 forg. Apache dubbo. Demo. DemoService % 3 fanyhost % 3 dtrue % 26 application % 3 ddemo – provider % 26 Dubbo % 3 d2. 0.2% % 26 generic 3 dfalse % 26 interface % 3 dorg. Apache dubbo. Demo. DemoService % 26 the methods % 3 dsayhello % 26 pid % 3 d4264%26 side %3Dprovider%26timestamp%3D1546848704035
End
Due to space constraints, today’s introduction of so much. As a reminder, we have created a DubboExporter in the registryProtocol.export () method, turned on the Netty server, and also created a temporary node associated with the service on the zooKeeper registry! For the rest of the registryProtocol.export () method, we will have a chance to discuss it later.
Original articles on this BLOG are not allowed to be used for commercial purposes and traditional media without my permission. Network media reprint please indicate the source, otherwise belongs to infringement.