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 遍歷的用法
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
相關文章