The previous chapter examined the source code exposed by the service, and this chapter continues with the source code referenced by the service. There are two types of reference in Dubbo: the first is direct service connection and the second is registrie-based reference. Direct service connection is generally used in testing scenarios, and online is more registrie-based.
The reference of the service is divided into hungry type and lazy type, which is referred to when invoking afterPropertiesSet method of ReferenceBean. Lazy ReferenceBean is referred to when the service corresponding to the ReferenceBean is injected into another class, that is, when the service is used. Dubbo is lazy referencing by default.
Similarly, the entry to the service reference can be found in the DubboNamespaceHandler, the ReferenceBean class. The entry method is getObject:
public Object getObject(a) throws Exception {
return get();
}
public synchronized T get(a) {
if (destroyed) {
throw new IllegalStateException("Already destroyed!");
}
if (ref == null) {
init();
}
return ref;
}
Copy the codeThe above two methods are relatively simple. If ref is not empty, the initialization method is entered. If the Dubbo source code is below 2.6.5, the init method will not be entered if ref is not null. The solution is to type the breakpoint directly on the init method line. Or set the following in the Settings of IDEA:
0. Process the configuration
Enter the init method, which does a lot of work, basically getting configuration information from various places, as explained below:
private void init(a) {
// Avoid repeated initialization
if (initialized) {
return;
}
initialized = true;
// Check whether the interface name is valid
if (interfaceName == null || interfaceName.length() == 0) {
throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
}
// get consumer's global configuration
//...
// Get the attribute value corresponding to the interface name from the system variable
String resolve = System.getProperty(interfaceName);
String resolveFile = null;
// If not, load it
if (resolve == null || resolve.length() == 0) {
// Get the parse file path from the system variable
resolveFile = System.getProperty("dubbo.resolve.file");
if (resolveFile == null || resolveFile.length() == 0) {
// If not, load from the specified location
File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
if (userResolveFile.exists()) {
// Get the absolute pathresolveFile = userResolveFile.getAbsolutePath(); }}if(resolveFile ! =null && resolveFile.length() > 0) {
Properties properties = new Properties();
FileInputStream fis = null;
try {
fis = new FileInputStream(new File(resolveFile));
// Load the configuration from a file
properties.load(fis);
} catch (IOException e) {
throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);
} finally {
try {
if (null! = fis) fis.close(); }catch(IOException e) { logger.warn(e.getMessage(), e); }}// Get the configuration corresponding to the interfaceresolve = properties.getProperty(interfaceName); }}if(resolve ! =null && resolve.length() > 0) {
url = resolve;
// Print logs.
}
//-------------------------------------------------------------------------------------//
// Get application, registries, and monitor information, or null if none
//...
// Check application validity
checkApplication();
checkStubAndMock(interfaceClass);
//-------------------------------------------------------------------------------------//
// Add side, protocol version, timestamp, process number and other information to map
Map<String, String> map = new HashMap<String, String>();
Map<Object, Object> attributes = new HashMap<Object, Object>();
map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
if (ConfigUtils.getPid() > 0) {
map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
}
// Non-generalization service (what is generalization, finally parse)
if(! isGeneric()) {// Get the version
String revision = Version.getVersion(interfaceClass, version);
if(revision ! =null && revision.length() > 0) {
map.put("revision", revision);
}
// Get the list of methods and add them to the map
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if (methods.length == 0) {
logger.warn("NO method found in service interface " + interfaceClass.getName());
map.put("methods", Constants.ANY_VALUE);
} else {
map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ",")); }}// Add the previously obtained information to the map
map.put(Constants.INTERFACE_KEY, interfaceName);
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);
//-------------------------------------------------------------------------------------//
// Iterate over MethodConfig to handle event notification configuration
//...
//-------------------------------------------------------------------------------------//
// Get the consumer'S IP address
String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
if (hostToRegistry == null || hostToRegistry.length() == 0) {
hostToRegistry = NetUtils.getLocalHost();
} else if (isInvalidLocalHost(hostToRegistry)) {
throw new IllegalArgumentException("Specified invalid registry ip from property:" + Constants.DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
}
map.put(Constants.REGISTER_IP_KEY, hostToRegistry);
// Store property values into the system context, i.e., the various property values obtained earlier
StaticContext.getSystemContext().putAll(attributes);
// Create a proxy object
ref = createProxy(map);
ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
}
Copy the codeThe initial configuration work is about this much, which omitted some code, can be debugged through IDEA. The following is the key.
1. Service references
We enter the createProxy method, which literally creates a proxy object and also calls other methods to build and merge Invoker instances.
private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp"."localhost".0, map);
final boolean isJvmRefer;
if (isInjvm() == null) {
// If the URL is not empty, no local reference is made
if(url ! =null && url.length() > 0) { // if a url is specified, don't do local reference
isJvmRefer = false;
// Check whether a local reference is required according to the protocol, scope, injvm and other parameters of the URL
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
// by default, reference local service if there is
isJvmRefer = true;
} else {
isJvmRefer = false; }}else {
isJvmRefer = isInjvm().booleanValue();
}
// Local reference
if (isJvmRefer) {
// Generate a local reference URL with the protocol injvm
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
// Build InjvmInvoker instance
invoker = refprotocol.refer(interfaceClass, url);
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
// Remote reference
} else {
// If the url is not empty, it may be directly connected to the service.
if(url ! =null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
if(us ! =null && us.length > 0) {
for (String u : us) {
URL url = URL.valueOf(u);
if (url.getPath() == null || url.getPath().length() == 0) {
// Set the fully qualified name of the interface to url path
url = url.setPath(interfaceName);
}
// Check whether the URL protocol is Registry. If so, it indicates that the user wants to use the specified registry instead of the directly connected service
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
} else{ urls.add(ClusterUtils.mergeUrl(url, map)); }}}}else {
// Load the registry URL
List<URL> us = loadRegistries(false);
if(us ! =null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);
if(monitorUrl ! =null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
// Add the refer parameter to the URL and add the URL to the urlsurls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map))); }}// If there is no registry, an exception is thrown
if (urls == null || urls.size() == 0) {
throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config
The main purpose of the createProxy method is to distinguish between a local reference and a remote reference, and then between a direct or a registry, multiple (service) or a single registry. Eventually, an instance of Invoker is built and the generated proxy class is returned. Invoker is at the heart of Dubbo. On the consumer side, Invoker is used to make remote calls that are built from the Protocol implementation class. There are many Protocol implementation classes, mostly protocol-specific, as we saw when analyzing service exposures. As for the creation of Invoker on the consumer side, in order not to affect the smoothness of the article, it will be analyzed at the end.
2. Create an agent
Once Invoker is created, the next step is to generate proxy objects for the service interface. In AbstractProxyFactory abstract class, getProxy mainly gets the list of interfaces, and finally calls an abstract getProxy method. This abstract method is implemented in the JavassistProxyFactory class as follows:
public <T> T getProxy(Invoker
invoker, Class
[] interfaces)
{
// Create proxy objects using newInstance
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
Copy the codeAlthough this Proxy is not in Java, it is encapsulated by Dubbo himself, but it is still implemented based on Java reflection. The InvokerInvocationHandler falls entirely within the Java dynamic proxy, which implements the Invoke method to invoke the target method. You can add extra methods before and after a target method is called, which is how AOP works.
Question parsing
-
How is the consumer Invoker created?
-
-
As you can see from the createProxy method in the source code analysis above, the consumer’s Invoker is built by the refer method. Different protocols go into different Protocol implementation classes and call the corresponding refer methods. Debug source code can be found, the URL protocol is Registry, we enter the corresponding implementation class.
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { / / set protocol header, url = zookeeper: / / 192.168.174.128:2181 / com. Alibaba. Dubbo. Registry. RegistryService... url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY); // Get the registry Registry registry = registryFactory.getRegistry(url); if (RegistryService.class.equals(type)) { return proxyFactory.getInvoker((T) registry, type, url); } // Convert the parameters on the URL to a Map Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY)); // Get the group configuration String group = qs.get(Constants.GROUP_KEY); if(group ! =null && group.length() > 0) { if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) { returndoRefer(getMergeableCluster(), registry, type, url); }}// Continue with the service reference logic return doRefer(cluster, registry, type, url); } Copy the codeNext, enter the doRefer method:
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { // Create a RegistryDirectory instance RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url); // Set up the registry and protocol directory.setRegistry(registry); directory.setProtocol(protocol); // all attributes of REFER_KEY Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters()); // Generate the consumer URL URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.remove(Constants.REGISTER_IP_KEY), 0, type.getName(), parameters); // Register the consumer service, there is a consumer node if(! Constants.ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(Constants.REGISTER_KEY,true)) { registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY, Constants.CHECK_KEY, String.valueOf(false))); } // Subscribe to node data such as providers, Configuration, and Routers directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, Constants.PROVIDERS_CATEGORY + "," + Constants.CONFIGURATORS_CATEGORY + "," + Constants.ROUTERS_CATEGORY)); // There may be multiple service providers in the registry that need to be merged into an Invoker Invoker invoker = cluster.join(directory); ProviderConsumerRegTable.registerConsuemr(invoker, url, subscribeUrl, directory); return invoker; } Copy the codeFinally, we start the consumer and open the Zookeeper client to see if there is consumer information under the consumer node:
-
Need to pay attention to a point, to keep the consumer running, otherwise can’t see.
-
The RegistryProtocol has not enabled the network connection. Why did zK communicate with RegistryProtocol? In fact, the refer method of DubboProtocol is also called in the above method, and the Server is started in DubboProtocol. Let’s look at the refer method for DubboProtocol:
-
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException { optimizeSerialization(url); / / create DubboInvoker DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers); invokers.add(invoker); return invoker; } Copy the code -
This method is relatively simple, and the key is getClients. It gets an instance of a communication client, type ExchangeClient, but ExchangeClient does not have communication capabilities and its underlying default is a Netty client. Let’s look at the getClients method:
-
private ExchangeClient[] getClients(URL url) { // whether to share connection boolean service_share_connect = false; int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0); // if not configured, connection is shared, otherwise, one connection for one service if (connections == 0) { service_share_connect = true; connections = 1; } ExchangeClient[] clients = new ExchangeClient[connections]; for (int i = 0; i < clients.length; i++) { if (service_share_connect) { // Obtain the shared client clients[i] = getSharedClient(url); } else { // Initializes the new clientclients[i] = initClient(url); }}return clients; } Copy the code -
The getSharedClient method also calls the initClient method. Let’s look at the initClient method directly:
-
private ExchangeClient initClient(URL url) { // Get the client type. The default is Netty String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT)); String version = url.getParameter(Constants.DUBBO_VERSION_KEY); booleancompatible = (version ! =null && version.startsWith("1.0.")); // Add codec and heartbeat packet parameters to the URL url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME); url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT)); // BIO is not allowed since it has severe performance issue. if(str ! =null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported client type: " + str + "," + " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), "")); } ExchangeClient client; try { // connection should be lazy if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) { client = new LazyConnectExchangeClient(url, requestHandler); } else { // Create a normal ExchangeClient instanceclient = Exchangers.connect(url, requestHandler); }}catch (RemotingException e) { throw new RpcException("Fail to create remoting client for service(" + url + ")." + e.getMessage(), e); } return client; } Copy the code -
The next step is to simply debug the connect method layer by layer and see the NettyClient constructor.
-
