源码解析｜registry meta 选主

背景

SOFARgeistry 分为 Session、Data 和 Meta 三个模块。Session 模块用于 client 交互，可以横向扩容，可以承受大量 client 连接请求；Data 是数据存储模块，利用 slot 分片机制来均衡负载，使用备份来解决高可用问题；Meta 是 Session、Data 的注册中心，采用分布式锁来选举 leader，本文详细阐述 Meta 如何选主。

基于 MySQL 的分布式锁

MySQL 表

drop table if exists distribute_lock;
CREATE TABLE distribute_lock (
  id bigint(20) NOT NULL AUTO_INCREMENT primary key,
  data_center varchar(128) NOT NULL,
  lock_name varchar(1024) NOT NULL,
  owner varchar(512) NOT NULL,
  duration bigint(20) NOT NULL,
  term     bigint(20) unsigned NOT NULL DEFAULT 0 COMMENT '任期',
  term_duration    bigint(20) unsigned NOT NULL DEFAULT 0 COMMENT '租期',
  gmt_create timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
  gmt_modified timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
  UNIQUE KEY `uk_data_center_lock` (`data_center`, `lock_name`),
  KEY `idx_lock_owner` (`owner`)
);

表的增改查等操作

public interface DistributeLockMapper {

  /**
   * query by dataCenter and lockName
   *
   * @param dataCenter
   * @param lockName
   * @return
   */
  public DistributeLockDomain queryDistLock(
      @Param("dataCenter") String dataCenter, @Param("lockName") String lockName);

  /**
   * compete lock, it will throw exception if lockName existed
   *
   * @param lock
   */
  public void competeLockOnInsert(DistributeLockDomain lock) throws Exception;

  /**
   * compete lock with cas
   *
   * @param competeLock
   * @return
   */
  public void competeLockOnUpdate(FollowCompeteLockDomain competeLock);

  /** renew lock last update time */
  public void ownerHeartbeat(DistributeLockDomain lock);

  /** force reset owner and duration */
  public void forceRefresh(DistributeLockDomain lock);
}

整体流程

step1：启动时创建锁记录，

<insert id="competeLockOnInsert" parameterType="com.alipay.sofa.registry.jdbc.domain.DistributeLockDomain">
        <![CDATA[
       INSERT /*+ QUERY_TIMEOUT(2000000) */ INTO distribute_lock
       (
           data_center,
           lock_name,
           owner,
           duration,
           gmt_create,
           gmt_modified,
           `term`,
           `term_duration`
       )
       VALUES
       (
           #{dataCenter},
           #{lockName},
           #{owner},
           #{duration},
           NOW(3),
           NOW(3),
           1,
           0
       )
       ON DUPLICATE KEY UPDATE lock_name = #{lockName}
       ]]>
    </insert>

step2：leader 每秒提交心跳，更新表

    <update id="ownerHeartbeat" parameterType="com.alipay.sofa.registry.jdbc.domain.DistributeLockDomain">
        <![CDATA[
          update /*+ QUERY_TIMEOUT(2000000) */ distribute_lock set owner = #{owner}, gmt_modified =  NOW(3), `term_duration` = (`term_duration` + 1)
          where data_center = #{dataCenter} and lock_name = #{lockName} and owner = #{owner} and term = #{term}  and `term_duration` = #{termDuration} and timestampdiff(SECOND, gmt_modified, NOW()) < #{duration}/1000
        ]]>
    </update>

step3：follower 每秒判断锁是否过期，如果过期，则 cas 竞选 leader

  public DistributeLockDomain onFollowWorking(DistributeLockDomain lock, String myself) {
    /** as follow, do compete if lock expire */
    if (lock.expire()) {
      LOG.info("lock expire: {}, meta elector start: {}", lock, myself);
      distributeLockMapper.competeLockOnUpdate(
          new FollowCompeteLockDomain(
              lock.getDataCenter(),
              lock.getLockName(),
              lock.getOwner(),
              lock.getGmtModified(),
              myself,
              lock.getDuration(),
              lock.getTerm(),
              lock.getTermDuration()));
      DistributeLockDomain newLock =
          distributeLockMapper.queryDistLock(lock.getDataCenter(), lock.getLockName());
      LOG.info("elector finish, new lock: {}", lock);
      return newLock;
    }
    return lock;
  }
  public boolean expire() {

    return gmtDbServerTime.getTime() > gmtModified.getTime() + duration;
  }
  <update id="competeLockOnUpdate">
        <!-- update cas with dataCenter,lockName,owner,gmtModified  -->
        <![CDATA[
       update /*+ QUERY_TIMEOUT(2000000) */ distribute_lock set owner = #{newOwner}, gmt_modified =  NOW(3), term = (term + 1) , `term_duration` = 0
       where data_center = #{dataCenter} and lock_name = #{lockName} and owner = #{owner} and term = #{term} and `term_duration` = #{termDuration} and timestampdiff(SECOND, gmt_modified, NOW()) > #{duration}/1000
       ]]>
    </update>

step4：如果 leader 发生切换，通知 xxx

时序图

💡 Tips：输入/画板或点击上方工具栏，选择「画板」、绘制流程图、架构图等各种图形。

类图

主要源代码解析

  public void elect() {
    synchronized (this) {
      if (isObserver) {//如果是Observer，不参与选主
        leaderInfo = doQuery();
      } else {
        leaderInfo = doElect();
      }

      if (amILeader()) {//我是leader
        onIamLeader();//我从follower变成leader，通知xxx
      } else {//我不是leader
        onIamNotLeader();//我从leader变成follower，通知xxx
      }
    }
  }
@Override
  protected LeaderInfo doElect() {
    //1、查询锁
    DistributeLockDomain lock =
        distributeLockMapper.queryDistLock(defaultCommonConfig.getClusterId(tableName()), lockName);
   //2、不存在则创建锁
    /** compete and return leader */
    if (lock == null) {
      return competeLeader(defaultCommonConfig.getClusterId(tableName()));
    }
   //3、判断角色
    ElectorRole role = amILeader(lock.getOwner()) ? ElectorRole.LEADER : ElectorRole.FOLLOWER;
    if (role == ElectorRole.LEADER) {
      lock = onLeaderWorking(lock, myself());//4、提交心跳
    } else {
      lock = onFollowWorking(lock, myself());//5、判断过期与否，如过期，则cas竞争锁
    }
    LeaderInfo result = leaderFrom(lock);//6、锁信息转换为LeaderInfo
    LOG.info("meta role : {}, leaderInfo: {}", role, result);
    return result;
  }

Meta 集群 leader 变更

1、初始化

2、kill 10.177.41.100 节点后

3、kill 10.181.160.18 节点后