Flink基于zookeeper的高可用实现源码分析

ゝ一世哀愁。 2023-06-27 12:52 9阅读 0赞

![format_png][]

点击箭头处“蓝色字”，关注我们哦！！

--------------------

Flink中JobMaster、ResourceManager、Dispatcher、WebMonitorEndpoint提供了基于zookeeper高可用，涉及到leader选举与监听， leader选举基于zookeeper开源客户端CuratorFramework 的LeaderLatch方式实现，监听则通过NodeCache实现。基于此Flink提供了zookeeper高可用ZooKeeperHaServices, 通过该工具类可以创建LeaderElectionService与LeaderRetrievalService,包含了对应与zookeeper的znode节点分别是

/resource_manager_lock
    /dispatcher_lock
    /job_manager_lock
    /rest_server_lock

LeaderElectionService用于leader选举服务，基于zk的实现类是ZooKeeperLeaderElectionService，LeaderRetrievalService用于监听leader恢复服务，基于zk的实现类是ZooKeeperLeaderRetrievalService。

涉及到两个重要的角色：

LeaderContender接口，在leader选举中使用，代表了参与leader竞争的角色，其实现类有JobManagerRunner、ResourceManager、Dispatcher、WebMonitorEndpoint，该接口中包含了两个重要的方法：

1.  grantLeadership,表示leader竞选成功的回调方法

2.  revokeLeadership,表示由leader变为非leader的回调方法

LeaderRetrievalListener接口，在监听leader恢复中使用，代表的是一个监听者，其实现类有ResourceManagerLeaderListener、JobManagerLeaderListener等，该接口包含一个重要的方法：

1. notifyLeaderAddress,leader发生变化时回调方法

以JobMaster与TaskExecutor之间的交互为例，JobMaster需要通过LeaderElectionService完成leader的选举，而TaskExecutor则需要监听jobMaster leader的变化，通过LeaderRetrievalService完成。

--------------------

**JobMaster选举**

JobManagerRunner负责JobMaster的选举与启动，内部包含了LeaderElectionService对象属性leaderElectionService，也就是ZooKeeperLeaderElectionService的对象，在调用JobManagerRunner的start方法时会调用leaderElectionService的start方法，传入参数this也就是当前的JobManagerRunner对象，

public void start() throws Exception {
        try {
          leaderElectionService.start(this);
        } catch (Exception e) {
          log.error("Could not start the JobManager because the leader election service did not start.", e);
          throw new Exception("Could not start the leader election service.", e);
        }
      }

在ZooKeeperLeaderElectionService 的start方法里面

public void start(LeaderContender contender) throws Exception {
        Preconditions.checkNotNull(contender, "Contender must not be null.");
        Preconditions.checkState(leaderContender == null, "Contender was already set.");
    
    
        LOG.info("Starting ZooKeeperLeaderElectionService {}.", this);
    
    
        synchronized (lock) {
    
    
          client.getUnhandledErrorListenable().addListener(this);
    
    
          leaderContender = contender;
    
    
          leaderLatch.addListener(this);
          leaderLatch.start();
    
    
          cache.getListenable().addListener(this);
          cache.start();
    
    
          client.getConnectionStateListenable().addListener(listener);
    
    
          running = true;
        }
      }

leaderLatch 表示的是CuratorFramework里面参与leader选举的LeaderLatch对象，添加了一个this的listener, 也就是LeaderLatchListener，选举成功会调用isLeader方法，由leader变为非leader调用notLeader方法；并且还要通过NodeCache方式添加了监控当前节点变化的listener，也就是NodeCacheListener，当监听的节点发生变化则调用nodeChanged方法。

若参与者竞选leader成功就会调用isLeader方法

public void isLeader() {
        synchronized (lock) {
          if (running) {
            issuedLeaderSessionID = UUID.randomUUID();
            confirmedLeaderSessionID = null;
    
    
            if (LOG.isDebugEnabled()) {
              LOG.debug(
                "Grant leadership to contender {} with session ID {}.",
                leaderContender.getAddress(),
                issuedLeaderSessionID);
            }
    
    
            leaderContender.grantLeadership(issuedLeaderSessionID);
          } else {
            LOG.debug("Ignoring the grant leadership notification since the service has " +
              "already been stopped.");
          }
        }
      }

在里面会调用leaderContender的grantLeadership方法，也就是JobManagerRunner的grantLeadership，就会执行JobMaster的启动并且执行任务的调度。

在检测到选择leader所用的path发生变化就会调用nodeChanged方法，在nodeChanged方法里面判断当前角色是leader就将leader信息写入到zookeeper路径里面去。

--------------------

**TaskExecutor监听**

在TaskExecutor中包含了一个JobLeaderService的对象属性jobLeaderService，在jobLeaderService中维护了所有的jobmaster leader,并且监听它，监听正是通过ZooKeeperLeaderRetrievalService完成。在申请slot时会调用TaskExecutor的requestSlot方法，在该方法里面调用jobLeaderService.add方法，

public void addJob(final JobID jobId, final String defaultTargetAddress) throws Exception {
        Preconditions.checkState(JobLeaderService.State.STARTED == state, "The service is currently not running.");
    
    
        LOG.info("Add job {} for job leader monitoring.", jobId);
    
    
        final LeaderRetrievalService leaderRetrievalService = highAvailabilityServices.getJobManagerLeaderRetriever(
          jobId,
          defaultTargetAddress);
    
    
        JobLeaderService.JobManagerLeaderListener jobManagerLeaderListener = new JobManagerLeaderListener(jobId);
    
    
        final Tuple2<LeaderRetrievalService, JobManagerLeaderListener> oldEntry = jobLeaderServices.put(jobId, Tuple2.of(leaderRetrievalService, jobManagerLeaderListener));
    
    
        if (oldEntry != null) {
          oldEntry.f0.stop();
          oldEntry.f1.stop();
        }
    
    
        leaderRetrievalService.start(jobManagerLeaderListener);
      }

在该方法里面开始对jobmaster leader的监听，创建一个LeaderRetrievalService对象，并且调用其start方法，传入参数是JobManagerLeaderListener对象，也就是LeaderRetrievalListener对象，在ZooKeeperLeaderRetrievalService的start方法里面，通过NodeCache方式添加了监控当前节点变化的listener，也就是NodeCacheListener，当监听的节点发生变化则调用nodeChanged方法，而在该方法里面会调用LeaderRetrievalListener的notifyLeaderAddress方法，也就是JobManagerLeaderListener的notifyLeaderAddress方法，会重新与新的leader建立连接。

![format_png 1][]

往期精彩回顾

![format_png 2][]

[StreamOperator源码简析][StreamOperator]

[StreamTask源码分析][StreamTask]

[Flink心跳机制分析][Flink]

[Flink中延时调用设计与实现][Flink 1]

![format_png 3][]

关注回复Flink

获取更多系列

好看，就要点个"在看"

![format_png 4][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6L1VDNk0xSGYwU1NFcHIxbGJiaWF0aWF4UEpjOHk5SmNPZVJ5Sk9JaHNpYndTeFdQbUhDcUp3eldYOHhUTXo5TVlMcEhLQWtSZlkyZmNNcW92eXhyUjlLcEF3LzY0MA?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9TN3RJdm81eFlNNm9URlR6VnVIaWFIRUJZdll4SzVBd0dITVFPTlNUUlFnUFg1TnBQaWNxalVENnFJRm12a0F4V1ZJOWlhWmlhaWNucHloWTIxVUxtWW5oZVV3LzY0MA?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9TN3RJdm81eFlNNm9URlR6VnVIaWFIRUJZdll4SzVBd0c4SDFjdmMwNzhWWW44Y0ZkMURkMmExbFFlRmpuQVhwd2ZxaG0zNUlGQm56RWpZVEM5dXRjQ0EvNjQw?x-oss-process=image/format,png
[StreamOperator]: https://blog.csdn.net/u013516966/article/details/103379586
[StreamTask]: https://blog.csdn.net/u013516966/article/details/103556098
[Flink]: https://blog.csdn.net/u013516966/article/details/103798056
[Flink 1]: https://blog.csdn.net/u013516966/article/details/102927825
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2pwZy9TN3RJdm81eFlNNjNhR0FoNlZmT0ZJMHBDVnFrVXRkbkJXWUYzVkFaOUVLaGNpYzhWeEtWT1JpY2MzQVFiN2d2ZkVOT1FnQ1dycFZoc2lhdXFhanR0Unc3US82NDA?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9tbWJpei5xcGljLmNuL21tYml6X2dpZi9TN3RJdm81eFlNN01GRVJScTAzTTkwQUVIdTBJRUFSOFpEM21BVzhyZDRJQUZrSGxndEZnaWNCNll1Z014bXdCMk1TVmIyWllMc1kyaWJpYU9pYVdsNHVjUncvNjQw?x-oss-process=image/format,png