角色	说明
LeaderContender(竞选者)	需要选主的主体，比如dispatcher、resource manager。当选leader或者回收leader时进行回调
LeaderElectionService(竞选服务)	负责选举的服务。a.关联leaderElectionDriver实现，具体驱动实现可以是zk、k8s等；在启动时初始化驱动类；b.实现LeaderElectionEventHandler调用
LeaderElectionDriver(竞选服务驱动)	选主具体驱动实现。以ZooKeeperLeaderElectionDriver实现为例，其会实现LeaderLatchListener回调接口，在主节点变化和节点变化时得到监听，然后调用 LeaderElectionEventHandler
LeaderElectionEventHandler(竞选服务的事件处理类)	leader竞选回调。被选为Leader会调用LeaderContender的grantLeadership方法进行不同实现主体的后续逻辑，比如DispatcherRunner的grantLeadership后会创建Dispatcher服务，启动DispatcherBootstrap
LeaderRetrievalService(leader监听服务)	监听leader的变化，传递给监听者。实现与LeaderElectionService类似，但作用不同。主要是接收Leader信息然后通知监听者LeaderRetrievalListener，监听者
LeaderRetriever(监听者)	监听leader的监听者，实现了LeaderRetrievalListener接口。具体实现为RpcGatewayRetriever。RpcGatewayRetriever在回调中可以获取到leader的信息，创建akka的连接，生成aop代理类实例
RpcGateway	Rcp网关，实现类通过AOP的方式封装了akka层的细节，可以直接调用实现类方法实现akka通信

一、LeaderContender

其中 LeaderContender 接口主要在 leader 选举中使用，代表了参与leader竞争的角色，
其实现类有

• JobMasterServiceLeadershipRunner
• ResourceManager
• DefaultDispatcherRunner
• WebMonitorEndpoint

该接口中包含了两个重要的方法：

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

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

一个 服务需要进行选举， 在启动时，将自身作为竞争者，传递给了 leaderElectionService。

    @Override

    public void start() throws Exception {

        LOG.debug("Start leadership runner for job {}.", getJobID());

        leaderElectionService.start(this);

    }

二、LeaderElectionService

Leader选举服务 ，以其子类 DefaultLeaderElectionService 为例

2.1、LeaderElectionService

LeaderElectionService主要提供了参与选举竞争的接口调用以及竞争结果的查询接口:

2.2、LeaderElectionEventHandler(竞选服务的事件处理类)

leader竞选回调。被选为Leader会调用LeaderContender的grantLeadership方法进行不同实现主体的后续逻辑，比如DispatcherRunner的grantLeadership后会创建Dispatcher服务，启动DispatcherBootstrap

在开始参加 Leader 时(DefaultleaderElectionService::start )，会通过选举驱动器工厂创建一个 leaderElectionDriver，通过这个Driver工厂类，Flink 将基于 zookeeper 的 CuratorFramework 的细节，与 Flink 本身做了解耦

并将自身作为一个 LeaderElectionEventHandler 传入leaderElectionDriver。

    @Override
    public final void start(LeaderContender contender) throws Exception {
        checkNotNull(contender, "Contender must not be null.");
        Preconditions.checkState(leaderContender == null, "Contender was already set.");

        synchronized (lock) {
            running = true;
            leaderContender = contender;
            
            leaderElectionDriver =
                    leaderElectionDriverFactory.createLeaderElectionDriver(
                            this,
                            new LeaderElectionFatalErrorHandler(),
                            leaderContender.getDescription());
            LOG.info("Starting DefaultLeaderElectionService with {}.", leaderElectionDriver);
        }
    }

三 、LeaderElectionDriver

LeaderElectionDriver 选主具体驱动实现。以ZooKeeperLeaderElectionDriver实现为例，其会实现 LeaderLatchListener 回调接口，在主节点变化和节点变化时得到监听，然后调用 LeaderElectionEventHandler

以其 子类实现 ZooKeeperLeaderElectionDriver 为例

通过工厂创建

	// 通过工厂创建 
    @Override
    public ZooKeeperLeaderElectionDriver createLeaderElectionDriver(
            LeaderElectionEventHandler leaderEventHandler,  // leaderElectionService 对象 作为 leaderEventHandler
            FatalErrorHandler fatalErrorHandler,
            String leaderContenderDescription)
            throws Exception {
        return new ZooKeeperLeaderElectionDriver(
                client, path, leaderEventHandler, fatalErrorHandler, leaderContenderDescription);
    }

ZooKeeperLeaderElectionDriver 封装了 CuratorFramework 作为选举框架

    /**
     * Creates a ZooKeeperLeaderElectionDriver object.
     *
     * @param client Client which is connected to the ZooKeeper quorum
     * @param path ZooKeeper node path for the leader election
     * @param leaderElectionEventHandler Event handler for processing leader change events
     * @param fatalErrorHandler Fatal error handler
     * @param leaderContenderDescription Leader contender description
     */
    public ZooKeeperLeaderElectionDriver(
            CuratorFramework client,
            String path,
            LeaderElectionEventHandler leaderElectionEventHandler, // DefaultLeaderElectionService
            FatalErrorHandler fatalErrorHandler,
            String leaderContenderDescription)
            throws Exception {
        checkNotNull(path);
        this.client = checkNotNull(client);
        this.connectionInformationPath = ZooKeeperUtils.generateConnectionInformationPath(path);
        this.leaderElectionEventHandler = checkNotNull(leaderElectionEventHandler);
        this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
        this.leaderContenderDescription = checkNotNull(leaderContenderDescription);
        leaderLatchPath = ZooKeeperUtils.generateLeaderLatchPath(path);
        leaderLatch = new LeaderLatch(client, leaderLatchPath);
        this.cache =
                ZooKeeperUtils.createTreeCache(
                        client,
                        connectionInformationPath,
                        this::retrieveLeaderInformationFromZooKeeper);

        running = true;
		// 启动选举 
        leaderLatch.addListener(this); // 添加  LeaderLatchListener
        leaderLatch.start();

        cache.start();

        client.getConnectionStateListenable().addListener(listener);
    }

// org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.leader.LeaderLatch#start
    public void start() throws Exception {
        Preconditions.checkState(this.state.compareAndSet(LeaderLatch.State.LATENT, LeaderLatch.State.STARTED), "Cannot be started more than once");
        this.startTask.set(AfterConnectionEstablished.execute(this.client, new Runnable() {
            public void run() {
                try {
                    LeaderLatch.this.internalStart();
                } finally {
                    LeaderLatch.this.startTask.set((Object)null);
                }

            }
        }));
    }

3.1、LeaderLatchListener

public interface LeaderLatchListener {
	// 选举成功回调方法
    void isLeader();
	
    void notLeader();
}

回调实现

    @Override
    public void isLeader() {
    	// org.apache.flink.runtime.leaderelection.DefaultLeaderElectionService#start 中  
    	// DefaultLeaderElectionService 将 自己作为 一个 leaderElectionEventHandler
    	// 所以此处选举成功， 通过回调 通知 DefaultLeaderElectionService 选举leader
        leaderElectionEventHandler.onGrantLeadership(UUID.randomUUID());
    }

    @Override
    public void notLeader() {
        leaderElectionEventHandler.onRevokeLeadership();
    }

DefaultLeaderElectionService 响应选举成功，回调 leaderContender.grantLeadership：

    @Override
    @GuardedBy("lock")
    public void onGrantLeadership(UUID newLeaderSessionId) {
        synchronized (lock) {
            if (running) {
                issuedLeaderSessionID = newLeaderSessionId;
                clearConfirmedLeaderInformation();

                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Grant leadership to contender {} with session ID {}.",
                            leaderContender.getDescription(),
                            issuedLeaderSessionID);
                }
				// 主要逻辑 leaderContender， 
                leaderContender.grantLeadership(issuedLeaderSessionID);
            } else {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(
                            "Ignoring the grant leadership notification since the {} has "
                                    + "already been closed.",
                            leaderElectionDriver);
                }
            }
        }
    }

以 WebMonitorEndpoint 为例

    @Override
    public void grantLeadership(final UUID leaderSessionID) {
        log.info(
                "{} was granted leadership with leaderSessionID={}",
                getRestBaseUrl(),
                leaderSessionID);
        leaderElectionService.confirmLeadership(leaderSessionID, getRestBaseUrl());
    }

参考：
https://www.modb.pro/db/107324
https://blog.csdn.net/qq_44836294/article/details/108022739