今天看啥
热点:

Curator源码解析(二)初始化和启动分析,curator源码


上一篇文章这里已经列出了Curator的一个使用的例子,这篇文章将详细分析其初始化和启动部分。

测试程序分析

1      初始化和启动

(1) newClient方法返回CuratorFramework接口对象:

 

public static CuratorFramework newClient(String connectString, int sessionTimeoutMs,int connectionTimeoutMs,RetryPolicy retryPolicy)
   {
        return builder().
            connectString(connectString).
            sessionTimeoutMs(sessionTimeoutMs).
           connectionTimeoutMs(connectionTimeoutMs).
            retryPolicy(retryPolicy).
            build();
    }

 

看看builder()方法:

   //返回用于构建CuratorFramework的新的builder对象
   publicstaticBuilder builder()
   {
        return new Builder();
}

 

可以看到这个方法返回一个构建CuratorFramework的Builder。Builder类就在CuratorFrameworkFactory.java文件中。

        //设置连接到的ZooKeeper集群的地址列表
        public BuilderconnectString(String connectString)
        {
            ensembleProvider =newFixedEnsembleProvider(connectString);
            return this;
        }

 

前面方法都是设置当前对象的属性,然后将当前Builder对象返回,设置的属性可以看到包括这些:

private EnsembleProvider   ensembleProvider;
        private int                 sessionTimeoutMs =DEFAULT_SESSION_TIMEOUT_MS;
        private int                 connectionTimeoutMs =DEFAULT_CONNECTION_TIMEOUT_MS;
        private int                 maxCloseWaitMs =DEFAULT_CLOSE_WAIT_MS;
        private RetryPolicy        retryPolicy;
        private ThreadFactory      threadFactory =null;
        private String             namespace;
        private String             authScheme =null;
        private byte[]              authValue =null;
        private byte[]              defaultData =LOCAL_ADDRESS;
        private CompressionProvidercompressionProvider=DEFAULT_COMPRESSION_PROVIDER;
        private ZookeeperFactory   zookeeperFactory =DEFAULT_ZOOKEEPER_FACTORY;
        private ACLProvider        aclProvider =DEFAULT_ACL_PROVIDER;
        private boolean             canBeReadOnly =false;

 

主要看最后一个build()方法:

//使用当前的Builder对象构建一个CuratorFramework接口对象
        public CuratorFramework build()
        {
            return new CuratorFrameworkImpl(this);
    }

 

可以看到创建一个CuratorFrameworkImpl实例,将当前Builder对象传递进去。

CuratorFrameworkImpl类是CuratorFramework接口的实现类,看看其构造函数:

public CuratorFrameworkImpl(CuratorFrameworkFactory.Builder builder)
   {
        ZookeeperFactory localZookeeperFactory= makeZookeeperFactory(builder.getZookeeperFactory());
        this.client =new CuratorZookeeperClient(localZookeeperFactory, builder.getEnsembleProvider(),builder.getSessionTimeoutMs(), builder.getConnectionTimeoutMs(),new Watcher()
        {
            @Override
            public void process(WatchedEvent watchedEvent)
            {
                CuratorEvent event = newCuratorEventImpl(CuratorFrameworkImpl.this, CuratorEventType.WATCHED,watchedEvent.getState().getIntValue(),unfixForNamespace(watchedEvent.getPath()),null,null,null,null,null, watchedEvent,null);
               processEvent(event);
            }
        }, builder.getRetryPolicy(),builder.canBeReadOnly());
 
        listeners = new ListenerContainer<CuratorListener>();
        unhandledErrorListeners =newListenerContainer<UnhandledErrorListener>();
       backgroundOperations = newDelayQueue<OperationAndData<?>>();
        namespace = new NamespaceImpl(this, builder.getNamespace());
        threadFactory = getThreadFactory(builder);
        maxCloseWaitMs = builder.getMaxCloseWaitMs();
        connectionStateManager =new ConnectionStateManager(this,builder.getThreadFactory());
        compressionProvider =builder.getCompressionProvider();
        aclProvider = builder.getAclProvider();
        state = new AtomicReference<CuratorFrameworkState>(CuratorFrameworkState.LATENT);
 
        byte[] builderDefaultData =builder.getDefaultData();
        defaultData = (builderDefaultData !=null) ? Arrays.copyOf(builderDefaultData,builderDefaultData.length):newbyte[0];
 
        if ( builder.getAuthScheme() !=null )
        {
            authInfo.set(new AuthInfo(builder.getAuthScheme(),builder.getAuthValue()));
        }
 
        failedDeleteManager = new FailedDeleteManager(this);
        namespaceFacadeCache =new NamespaceFacadeCache(this);
}

 

CuratorFrameworkImpl主要是对CuratorZookeeperClient的封装,所以我们主要看构造函数中第二句代码是如何构建CuratorZookeeperClient对象的。构造函数中除了参数1和参数5,其他参数都是来自builder对象,参数1是localZookeeperFactory,通过下面方法构造:

ZookeeperFactory localZookeeperFactory = makeZookeeperFactory(builder.getZookeeperFactory());

下面是makeZookeeperFactory()方法的实现代码:

private ZookeeperFactory makeZookeeperFactory(final ZookeeperFactory actualZookeeperFactory)
   {
        return new ZookeeperFactory()
        {
            @Override
            public ZooKeeper newZooKeeper(StringconnectString,intsessionTimeout, Watcher watcher,boolean canBeReadOnly) throws Exception
            {
                ZooKeeper zooKeeper =actualZookeeperFactory.newZooKeeper(connectString, sessionTimeout, watcher,canBeReadOnly);
                AuthInfo auth = authInfo.get();
                if ( auth !=null )
                {
                    zooKeeper.addAuthInfo(auth.scheme, auth.auth);
                }
 
                return zooKeeper;
            }
        };
}

 

传递进来的是builder中定义的ZookeeperFactory对象,实际上就是Curator提供的DefaultZookeeperFactory类,定义如下:

public classDefaultZookeeperFactory implements ZookeeperFactory
{
   @Override
   publicZooKeeper newZooKeeper(String connectString,int sessionTimeout, Watcher watcher,boolean canBeReadOnly) throws Exception
   {
        return new ZooKeeper(connectString, sessionTimeout,watcher, canBeReadOnly);
   }
}

 

仅仅是简单地new出一个原生的ZooKeeper对象,所以传递到CuratorZookeeperClient构造函数中的ZookeeperFactory类的newZooKeeper返回的是原生的ZooKeeper对象。

 

参数5是一个Watcher对象,其中事件响应函数process()又调用了processEvent()方法:

private void processEvent(finalCuratorEvent curatorEvent)
   {
        if ( curatorEvent.getType() ==CuratorEventType.WATCHED )
        {
           validateConnection(curatorEvent.getWatchedEvent().getState());
        }
 
        listeners.forEach(new Function<CuratorListener, Void>()
        {
            @Override
            public Void apply(CuratorListenerlistener)
            {
                try
                {
                    TimeTrace trace = client.startTracer("EventListener");
                   listener.eventReceived(CuratorFrameworkImpl.this, curatorEvent);
                    trace.commit();
                }
                catch ( Exception e )
                {
                    logError("Event listener threw exception", e);
                }
                returnnull;
            }
       });
}

这个watcher什么时候被调用?作用是什么?这个问题稍后再解答。

 

接着深入进去看的CuratorZookeeperClient类的构造函数:

   publicCuratorZookeeperClient(ZookeeperFactory zookeeperFactory, EnsembleProviderensembleProvider,int sessionTimeoutMs,int connectionTimeoutMs, Watcher watcher, RetryPolicy retryPolicy,boolean canBeReadOnly)
   {
        this.connectionTimeoutMs = connectionTimeoutMs;
        state = new ConnectionState(zookeeperFactory,ensembleProvider, sessionTimeoutMs, connectionTimeoutMs, watcher,tracer, canBeReadOnly);
        setRetryPolicy(retryPolicy);
    }

 

主要在构造函数的最后两行初始化下面两个成员变量:

private finalConnectionState state;

private finalAtomicReference<RetryPolicy> retryPolicy =newAtomicReference<RetryPolicy>();

 

继续看ConnectionState的构造函数:

ConnectionState(ZookeeperFactoryzookeeperFactory, EnsembleProvider ensembleProvider,int sessionTimeoutMs,int connectionTimeoutMs,Watcher parentWatcher, AtomicReference<TracerDriver> tracer,boolean canBeReadOnly)
   {
        this.ensembleProvider = ensembleProvider;
        this.sessionTimeoutMs = sessionTimeoutMs;
        this.connectionTimeoutMs = connectionTimeoutMs;
        this.tracer = tracer;
        if ( parentWatcher !=null )
        {
            parentWatchers.offer(parentWatcher);
        }
 
        zooKeeper = new HandleHolder(zookeeperFactory,this, ensembleProvider,sessionTimeoutMs, canBeReadOnly);
}

做了两件事情,一个是把传递进来的watcher对象放入parentWatchers容器中,一个是new出HandleHolder对象,注意这里讲当前ConnectionState对象作为watcher参数传递到HandleHolder构造函数。继续看HandleHolder构造函数:

HandleHolder(ZookeeperFactoryzookeeperFactory, Watcher watcher, EnsembleProvider ensembleProvider,int sessionTimeout,boolean canBeReadOnly)
   {
        this.zookeeperFactory = zookeeperFactory;
        this.watcher = watcher;
        this.ensembleProvider = ensembleProvider;
        this.sessionTimeout = sessionTimeout;
        this.canBeReadOnly = canBeReadOnly;
}

 

简单字段赋值,这个类是ZooKeeper对象的持有者,其中包含两个关键函数:

ZooKeeper getZooKeeper() throws Exception
   {
        return (helper !=null) ?helper.getZooKeeper() : null;
}
 
void closeAndReset()throws Exception
   {
        internalClose();
 
        // first helper is synchronized when getZooKeeper is called. Subsequentcalls
        // are not synchronized.
        helper = new Helper()
        {
            private volatile ZooKeeper zooKeeperHandle =null;
            private volatile String connectionString =null;
 
            @Override
            public ZooKeeper getZooKeeper()throws Exception
            {
                synchronized(this)
                {
                    if (zooKeeperHandle ==null)
                    {
                        connectionString =ensembleProvider.getConnectionString();
                        zooKeeperHandle =zookeeperFactory.newZooKeeper(connectionString,sessionTimeout,watcher,canBeReadOnly);
                    }
 
                    helper = newHelper()
                    {
                        @Override
                        public ZooKeepergetZooKeeper()throwsException
                        {
                            returnzooKeeperHandle;
                        }
 
                        @Override
                        public StringgetConnectionString()
                        {
                            returnconnectionString;
                        }
                    };
 
                    returnzooKeeperHandle;
                }
            }
 
            @Override
            public String getConnectionString()
            {
                returnconnectionString;
            }
        };
}

 

可以看到这个类提供了一个getZooKeeper()方法,返回ZooKeeper对象,closeAndReset()方法是对helper对象的初始化,Helper中的getZooKeeper()方法返回的是ZooKeeper对象的单例,保障一个HandleHolder只会持有一个ZooKeeper对象。

总结一下,通过CuratorFrameworkFactory类的newClient()方法将会返回一个实现了CuratorFramework接口的实现类CuratorFrameworkImpl的对象,这个对象中包含一个CuratorZookeeperClient对象,里面又包含一个ConnectionState对象,再里面又包含一个HandleHolder对象,这个对象通过从最外层逐层传递进来的DefaultZookeeperFactory对象获取原生ZooKeeper对象,并以单例进行维护,每一层都有一个getZooKeeper()方法,在外面调用会最终到HandleHolder这里来取得一个ZooKeeper对象。

这里面HandleHolderZooKeeper对象的持有者,外层封装的ConnectionState类是核心,管理ZooKeeper的连接状态,响应ZooKeeperwatch回调事件。这个回调函数是创建HandleHolder对象时将自己传递进去注册的。

初始化这一步骤注册的真正的原生ZooKeeper对象的watcher响应事件是ConnectionState类中的process()函数,我们看看这个函数:

@Override
   publicvoidprocess(WatchedEvent event)
   {
       //逐个调用parentWatchers容器中的Watcher的process函数
        for ( Watcher parentWatcher :parentWatchers )
        {
            TimeTrace timeTrace = new TimeTrace("connection-state-parent-process",tracer.get());
            parentWatcher.process(event);
            timeTrace.commit();
        }
 
        boolean wasConnected =isConnected.get();
        boolean newIsConnected = wasConnected;
        if ( event.getType() ==Watcher.Event.EventType.None )
        {
            newIsConnected =checkState(event.getState(), wasConnected);
        }
 
        //若当前连接状态不为false,则真正设置isConnected = true
        if ( newIsConnected != wasConnected )
        {
            isConnected.set(newIsConnected);
            connectionStartMs = System.currentTimeMillis();
        }
   }

 

这个watch事件响应函数主要做两件事:

(1)将parentWatchers容器中的所有Watcher都调用一次;

(2)检查并更新ConnectionState类中维护的ZooKeeper的连接状态isConnected

 

那么parentWatchers容器中有哪些Watcher呢,目前只有CuratorFrameworkImpl构造函数中初始化CuratorZookeeperClient对象时传递进去的Watcher,如下所示:

this.client=newCuratorZookeeperClient(localZookeeperFactory,builder.getEnsembleProvider(), builder.getSessionTimeoutMs(),builder.getConnectionTimeoutMs(),new Watcher()
        {
            @Override
            public void process(WatchedEvent watchedEvent)
            {
                CuratorEvent event = newCuratorEventImpl(CuratorFrameworkImpl.this, CuratorEventType.WATCHED,watchedEvent.getState().getIntValue(),unfixForNamespace(watchedEvent.getPath()),null,null,null,null,null, watchedEvent,null);
                processEvent(event);
            }
        }, builder.getRetryPolicy(),builder.canBeReadOnly());

 

其中的processEvent()函数实际上是将CuratorListener列表中的所有事件响应函数全部调用一次,这个和异步执行ZooKeeper操作相关,具体不介绍了,详细可以参考这个例子:

public staticvoid     setDataAsync(CuratorFramework client,String path,byte[]payload)throwsException
   {
        // this is one method of getting event/async notifications
        CuratorListener listener =newCuratorListener()
        {
            @Override
            public void eventReceived(CuratorFramework client,CuratorEvent event) throws Exception
            {
                // examine event for details
            }
        };
       client.getCuratorListenable().addListener(listener);
 
        // set data for the given node asynchronously. The completion notification
        // is done via the CuratorListener.
       client.setData().inBackground().forPath(path, payload);
}

 

(2)CuratorFramework 的start()方法启动:


CuratorFramework的start方法会调用CuratorZookeeperClient对象的start方法,内部又调用ConnectionState的start方法,最后ConnectionState的start方法调用一个reset方法:

private synchronizedvoidreset() throwsException
{
    isConnected.set(false);
        connectionStartMs = System.currentTimeMillis();
        zooKeeper.closeAndReset();
        zooKeeper.getZooKeeper();  // initiateconnection
}

 

主要是最后两句代码,调用HandleHolder类对象zooKeeper的closeAndReset方法是为了实例化获取ZooKeeper对象的Helper对象,调用一次getZooKeeper方法是为了先第一次实例化好ZooKeeper对象,提高之后调用访问接口时的性能。




www.bkjia.comtruehttp://www.bkjia.com/yjs/937894.htmlTechArticleCurator源码解析(二)初始化和启动分析,curator源码 上一篇文章这里已经列出了Curator的一个使用的例子,这篇文章将详细分析其初始化和启动...

相关文章

相关搜索: 笔记本电脑

帮客评论

视觉看点