dubbo源码：consumer中reference引用生成
前面咱们已经看了dubbo-config在解析了配置中dubbo:reference标签以后会在容器中建立一个referenceBean，在系统中调用远程服务方法时会先从容器中取出该referenceBean实例。咱们看其源码发现ReferenceBean除了实现InitializingBean接口还实现了FactoryBean接口，因此在从容器中获取实例时会调用该类实现方法getObject()。java
public class ReferenceBean<T> extends ReferenceConfig<T>
implements FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
private static final long serialVersionUID = 213195494150089726L;
private transient ApplicationContext applicationContext;
public ReferenceBean() {
super();
}
public ReferenceBean(Reference reference) {
super(reference);
}
public void setApplicationContext(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
SpringExtensionFactory.addApplicationContext(applicationContext);
}
//从IOC中获取实例时调用
public Object getObject() throws Exception {
return get();
}
public Class<?> getObjectType() {
return getInterfaceClass();
}
@Parameter(excluded = true)
public boolean isSingleton() {
return true;
}
@SuppressWarnings({ "unchecked" })
public void afterPropertiesSet() throws Exception {
。。。
}
}
咱们再getObject()方法中看到调用了其父类实现的同步方法get()在get()方法中进行了init初始化操做。该初始化操做就是咱们接下来要看的主要内容。spring
public synchronized T get() {
if (destroyed){
throw new IllegalStateException("Already destroyed!");
}
if (ref == null) { //ref代理接口的引用
init();
}
return ref;
}
private void init() {
if (initialized) {
return;
}
initialized = true;
if (interfaceName == null || interfaceName.length() == 0) {
throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
}
// 获取消费者全局配置
checkDefault();
appendProperties(this);//
//判断标签配置接口是否为泛化应用接口
if (getGeneric() == null && getConsumer() != null) {
setGeneric(getConsumer().getGeneric());
}
//加载接口class
if (ProtocolUtils.isGeneric(getGeneric())) {
interfaceClass = GenericService.class;
} else {
try {
interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
.getContextClassLoader());
} catch (ClassNotFoundException e) {
throw new IllegalStateException(e.getMessage(), e);
}
checkInterfaceAndMethods(interfaceClass, methods);//检查接口以及配置方法均为接口方法
}
//判断该类是否配置直连提供者，也能够经过文件方式配置dubbo2.0以上版本自动加载${user.home}/dubbo-resolve.properties文件，不须要配置。
String resolve = System.getProperty(interfaceName);
String resolveFile = null;
if (resolve == null || resolve.length() == 0) {
resolveFile = System.getProperty("dubbo.resolve.file");
if (resolveFile == null || resolveFile.length() == 0) {
File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
if (userResolveFile.exists()) {
resolveFile = userResolveFile.getAbsolutePath();
}
}
if (resolveFile != null && resolveFile.length() > 0) {
Properties properties = new Properties();
FileInputStream fis = null;
try {
fis = new FileInputStream(new File(resolveFile));
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);
}
}
resolve = properties.getProperty(interfaceName);
}
}
if (resolve != null && resolve.length() > 0) {
url = resolve;
if (logger.isWarnEnabled()) {
if (resolveFile != null && resolveFile.length() > 0) {
logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");
} else {
logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");
}
}
}
//加载consumer、module、registries、monitor、application等配置属性
if (consumer != null) {
if (application == null) {
application = consumer.getApplication();
}
if (module == null) {
module = consumer.getModule();
}
if (registries == null) {
registries = consumer.getRegistries();
}
if (monitor == null) {
monitor = consumer.getMonitor();
}
}
if (module != null) {
if (registries == null) {
registries = module.getRegistries();
}
if (monitor == null) {
monitor = module.getMonitor();
}
}
if (application != null) {
if (registries == null) {
registries = application.getRegistries();
}
if (monitor == null) {
monitor = application.getMonitor();
}
}
checkApplication();//检查application配置
checkStubAndMock(interfaceClass);//检查该接口引用是否为local、stub、mock等配置
//装载配置信息
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()));
}
if (! isGeneric()) {
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put("revision", revision);
}
//将引用的接口类转化成包装类,而后再获取接口中方法名称
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)), ","));//拼接方法名称用“,”分割
}
}
//组装属性配置
map.put(Constants.INTERFACE_KEY, interfaceName);
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, consumer, Constants.DEFAULT_KEY);
appendParameters(map, this);
String prifix = StringUtils.getServiceKey(map);
if (methods != null && methods.size() > 0) {
for (MethodConfig method : methods) {
appendParameters(map, method, method.getName());
String retryKey = method.getName() + ".retry";
if (map.containsKey(retryKey)) {
String retryValue = map.remove(retryKey);
if ("false".equals(retryValue)) {
map.put(method.getName() + ".retries", "0");
}
}
appendAttributes(attributes, method, prifix + "." + method.getName());
checkAndConvertImplicitConfig(method, map, attributes);
}
}
//attributes经过系统context进行存储.（dubbo内部维护的一个ConcurrentHashMap用于存储属性）
StaticContext.getSystemContext().putAll(attributes);
//根据配置信息建立代理类
ref = createProxy(map);
}
初始化中首先进行了属性的组装，最后进行接口代理引用的建立，这里提供了引用建立的入口咱们继续往下看（createProxy()方法）。bootstrap
private T createProxy(Map<String, String> map) {
URL tmpUrl = new URL("temp", "localhost", 0, map);
final boolean isJvmRefer;
if (isInjvm() == null) {//优先从JVM内获取引用实例，是否配置了本地调用<dubbo:protocol name="injvm" />
if (url != null && url.length() > 0) { //指定URL值的状况下，不作本地引用
isJvmRefer = false;
} else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {//本地有服务暴露使用本地调用
//默认状况下若是本地有服务暴露，则引用本地服务.
isJvmRefer = true;
} else {
isJvmRefer = false;
}
} else {
isJvmRefer = isInjvm().booleanValue();
}
if (isJvmRefer) {//本地调用状况
URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
invoker = refprotocol.refer(interfaceClass, url);//生成了invoker，它表明一个可执行体，可向它发起invoke调用
if (logger.isInfoEnabled()) {
logger.info("Using injvm service " + interfaceClass.getName());
}
} else {
//非本地调用，用户配置url，直连调用
if (url != null && url.length() > 0) { // 用户指定URL，指定的URL多是对点对直连地址，也多是注册中心URL
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) {
url = url.setPath(interfaceName);
}
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 { // 经过注册中心配置拼装URL（注册中心serviceURL）
List<URL> us = loadRegistries(false);
if (us != null && us.size() > 0) {
for (URL u : us) {
URL monitorUrl = loadMonitor(u);//监控中心url
if (monitorUrl != null) {
map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
}
urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));//注册中心url拼上refer（接口引用ref相关信息）
}
}
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 <dubbo:registry address=\"...\" /> to your spring config.");
}
}
if (urls.size() == 1) {//只有一个注册中心时
invoker = refprotocol.refer(interfaceClass, urls.get(0));
} else {//多注册中心时产生多个invoker
List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
URL registryURL = null;
for (URL url : urls) {
invokers.add(refprotocol.refer(interfaceClass, url));//咱们经过注册中心生成了invoker，它表明一个可执行体，可向它发起invoke调用
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
registryURL = url; // 用了最后一个registry url,也就是最后一个注册中心
}
}
//多注册中心时先对注册中心进行路由，而后再对注册中心每一个服务器进行路由
if (registryURL != null) { // 有 注册中心协议的URL
// 对有注册中心的Cluster 只用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
} else { // 不是 注册中心的URL
invoker = cluster.join(new StaticDirectory(invokers));
}
}
}
Boolean c = check;
if (c == null && consumer != null) {
c = consumer.isCheck();
}
if (c == null) {
c = true; // default true
}
if (c && ! invoker.isAvailable()) {
throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
}
if (logger.isInfoEnabled()) {
logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
}
// 建立服务代理
return (T) proxyFactory.getProxy(invoker);
}
在createProxy()的代码中，首先是判断是否能够进行本地调用，由于本地调用要比远程调用效率高不少，因此若是能够进行本地调用就优先进行本地调用。若是是非本地调用，就根据注册中心生成invoker对象（可执行体能够发起invoke调用），每一个注册中心生成一个invoker对象，若是是多个注册中心，就对生成的invokers作路由（只适用AvailableCluster）最后选出一个invoker建立代理。这里咱们要详细看的代码是invoker的生成过程（refprotocol.refer(interfaceClass, urls.get(0));）和经过invoker建立代理的过程（(T) proxyFactory.getProxy(invoker);）。Protocol接口有多个实现类支持多种协议这里咱们就拿dubbo默认协议为例来看。服务器
Protocol中ref()方法执行顺序:网络
ProtocolFilterWrapper.refer()-->ProtocolListenerWrapper.refer()-->RegistryProtocol.ref()这个过程进行客户端到zookeeper的注册和订阅，而后还会执行到ProtocolFilterWrapper.refer()-->ProtocolListenerWrapper.refer()-->DubboProtocol.refer()在这个过程当中会进行invoker过滤器的建立最后包装成InvokerChain调用链。这里咱们暂时先看DubboProtocol中的实现。app
//引用远程服务
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
// create rpc invoker.
DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
invokers.add(invoker);
return invoker;
}
//获取连接客户端
private ExchangeClient[] getClients(URL url){
//是否共享链接
boolean service_share_connect = false;
int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
//若是connections不配置，则共享链接，不然每服务每链接
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){
clients[i] = getSharedClient(url);
} else {
clients[i] = initClient(url);//若是不共享连接，会每次都建立一个新的client
}
}
return clients;
}
/**
*获取共享链接
*/
private ExchangeClient getSharedClient(URL url){
String key = url.getAddress();
ReferenceCountExchangeClient client = referenceClientMap.get(key);//map中获取
if ( client != null ){
if ( !client.isClosed()){
client.incrementAndGetCount();
return client;
} else {
// logger.warn(new IllegalStateException("client is closed,but stay in clientmap .client :"+ client));
referenceClientMap.remove(key);//已经关闭的连接从map中删除
}
}
ExchangeClient exchagneclient = initClient(url);//若是从map中获取的连接为null，则建立新连接
//ReferenceCountExchangeClient是exchagneclient的包装类，包装了一个计数器
client = new ReferenceCountExchangeClient(exchagneclient, ghostClientMap);
referenceClientMap.put(key, client);//新建以后都会存入map
ghostClientMap.remove(key);
return client;
}
/**
* 建立新链接.
*/
private ExchangeClient initClient(URL url) {
// client type setting.
//client属性值，为空时获取server值，若是server也是空就是用默认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);
//兼容1.0版本codec
boolean compatible = (version != null && version.startsWith("1.0."));
url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
//默认开启heartbeat，60s
url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
// BIO存在严重性能问题，暂时不容许使用
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 {
//设置链接应该是lazy的
if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){
client = new LazyConnectExchangeClient(url ,requestHandler);
} else {
client = Exchangers.connect(url ,requestHandler);//建立连接，Exchangers工具类里面进行操做的是HeaderExchanger
}
} catch (RemotingException e) {
throw new RpcException("Fail to create remoting client for service(" + url
+ "): " + e.getMessage(), e);
}
return client;
}
invoker中包含了连接客户端client。消费者端和生产者端就是经过这个client进行信息交换的。咱们深刻看一下client的建立过程。Exchangers是一个工具包装类里面执行的是HeaderExchanger的connect()方法jvm
public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
if (handler == null) {
throw new IllegalArgumentException("handler == null");
}
url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
//HeaderExchanger信息交换层
return getExchanger(url).connect(url, handler);
}
HeaderExchanger的connect()方法socket
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
//DecodeHandler编码解码处理器，
//Transporters是工具包装类，里面真正执行connect操做的是NettyTransporter
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
Transporters也是一个工具包装类里面执行的是NettyTransporter中connect()方法tcp
public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
if (url == null) {
throw new IllegalArgumentException("url == null");
}
ChannelHandler handler;
if (handlers == null || handlers.length == 0) {
handler = new ChannelHandlerAdapter();
} else if (handlers.length == 1) {
handler = handlers[0];
} else {
handler = new ChannelHandlerDispatcher(handlers);
}
//NettyTransporter网络传输层
return getTransporter().connect(url, handler);
}
NettyTransporter的connect()方法ide
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
//建立nettyClient客户端
return new NettyClient(url, listener);
}
NettyClient中直接调用父类的构造方法，在父类构造方法中又调用了子类中实现的doOpen()方法和doConnect()方法。
protected void doOpen() throws Throwable {
NettyHelper.setNettyLoggerFactory();
bootstrap = new ClientBootstrap(channelFactory);
// config
// @see org.jboss.netty.channel.socket.SocketChannelConfig
bootstrap.setOption("keepAlive", true);
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("connectTimeoutMillis", getTimeout());
final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("decoder", adapter.getDecoder());
pipeline.addLast("encoder", adapter.getEncoder());
pipeline.addLast("handler", nettyHandler);
return pipeline;
}
});
}
protected void doConnect() throws Throwable {
long start = System.currentTimeMillis();
ChannelFuture future = bootstrap.connect(getConnectAddress());
try{
boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
if (ret && future.isSuccess()) {
Channel newChannel = future.getChannel();
newChannel.setInterestOps(Channel.OP_READ_WRITE);
try {
// 关闭旧的链接
Channel oldChannel = NettyClient.this.channel; // copy reference
if (oldChannel != null) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
}
oldChannel.close();
} finally {
NettyChannel.removeChannelIfDisconnected(oldChannel);
}
}
} finally {
if (NettyClient.this.isClosed()) {
try {
if (logger.isInfoEnabled()) {
logger.info("Close new netty channel " + newChannel + ", because the client closed.");
}
newChannel.close();
} finally {
NettyClient.this.channel = null;
NettyChannel.removeChannelIfDisconnected(newChannel);
}
} else {
NettyClient.this.channel = newChannel;
}
}
} else if (future.getCause() != null) {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + ", error message is:" + future.getCause().getMessage(), future.getCause());
} else {
throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
+ getRemoteAddress() + " client-side timeout "
+ getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
+ NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
}
}finally{
if (! isConnected()) {
future.cancel();
}
}
}
相关文章
- 1. dubbo源码—service reference
- 2. dubbo的consumer代理的生成方式源码
- 3. dubbo源码分析三：consumer注册及生成代理对象
- 4. dubbo源码学习笔记----Provider和Consumer
- 5. Dubbo源码分析（二）：Dubbo之消费端（Consumer）
- 6. 读Kafka Consumer源码
- 7. dubbo标签的service与provider、reference与consumer的区别
- 8. Dubbo 源码分析 - 服务引用
- 9. dubbo服务源码——引用服务
- 10. dubbo源码分析02：服务引用
- 更多相关文章...
- • Eclipse 生成jar包 - Eclipse 教程
- • 现实生活中的 XML - XML 教程
- • Scala 中文乱码解决
- • C# 中 foreach 遍历的用法
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. Window下Ribbit MQ安装
- 2. Linux下Redis安装及集群搭建
- 3. shiny搭建网站填坑战略
- 4. Mysql8.0.22安装与配置详细教程
- 5. Hadoop安装及配置
- 6. Python爬虫初学笔记
- 7. 部署LVS-Keepalived高可用集群
- 8. keepalived+mysql高可用集群
- 9. jenkins 公钥配置
- 10. HA实用详解
欢迎关注本站公众号,获取更多信息
相关文章