VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > 编程开发 > Java教程 >

  • 来吧，自己动手撸一个分布式ID生成器组件

在经过了众多轮的面试之后，小林终于进入到了一家互联网公司的基础架构组，小林目前在公司有使用到架构组研究到分布式id生成器，前一阵子大概看了下其内部的实现，发现还是存在一些架构设计不合理之处。但是又由于适用于当前的业务场景，所以并没有做过多的优化，这里记录一些相关的技术笔记。

研发背景

在分布式服务中，各种复杂的业务场景需要有一个用于做唯一标识的id，例如订单业务，支付流水，聊天通信等业务场景。尤其是在分库分表场景中，分布式id生成器的使用频率更高。因此分布式id组件的设计应该要能支持以下几个特性：

1.全局唯一特性

这个点比较好理解，这里就不做过多的解释。

2.组件递增特性

可以是每个id都具有递增的特性也可以是支持区间段内具备递增的特性。

3.安全性

有些重要的id如果无意中暴露在了外网环境中，如果没有做过安全防范其实是一件非常危险的事情。例如说订单的id如果只是更具日期加订单数目的格式生成，例如说：2020111100001表示2020年11月11日的第一笔订单，那么如果竞对获取到了

2020111100999这个id，再根据订单的生成时间则大概可以推断出该公司某日生成的订单数目的大致量级。

4.高qps

分布式id生成组件在使用过程中主要是qps偏高，因此在设计起初应该要能支持较高的qps查询，同时对于网络的延迟特性也需要尽可能降低。

5.高可用

由于分布式id生成器是一个需要支持多个服务调用方共同使用的公共服务，一旦出现崩溃后果不堪设想，可能会导致大面积的业务线崩塌，所以在高可用方面需要考虑得尤其重要。

业界常见的分布式id生成方案比对

uuid

java程序中实现uuid的代码：

String result = UUID.randomUUID().toString();
System.out.println(result);

生成的格式如下所示：

b0b2197d-bc8c-4fab-ad73-2b54e11b0869

uuid的格式其实是会被 - 符号划分为五个模块，主要是分为了8-4-4-4-12个字符拼接而成的一段字符串。但是这种字符串的格式对于互联网公司中主推的MySQL数据库并不友好。

尤其是当使用生成的id作为索引的时候，uuid长度过长，大数据量的时候会导致b+树的叶子结点裂变频率加大，而且在进行索引比对的时候需要进行逐个字符比对，性能损耗也较高，应该抛弃该方案。小林查询了一些网上的资料发现uuid的主要组成由以下几个部位：

  • 当前日期和时间
  • 随机数字
  • 机器的MAC地址（能够保证全球范围内机器的唯一特性）

雪花算法

SnowFlake是Twitter公司采用的一种算法，目的是在分布式系统中产生全局唯一且趋势递增的ID。

来吧，自己动手撸一个分布式ID生成器组件

稍微解释一些雪花算法的含义：

第一位通常是0，没有特殊使用含义，因为1通常表示为补码。

中间的41位是用于存储时间，41位的长度足够容纳69年左右的时长。

10bit用于标示机器自身的id，从而表示不通机器自身id的不同。

最后12位bit用于表示某一毫秒内的序列号，12位（bit）可以表示的最大正整数是4096-1=4095，所以也就是说一毫秒内可以同时生成4095个id。

时间戳位置和序列号位置还不能随意调整，应为要保证逐渐递增的特性。

好处

能够保证递增的特性，id具有明确的含义，易懂。

不足点

但是对于机器自身的系统时间有所依赖，一旦机器的系统时间发生了变化，在高并发环境下就有可能会有重复id生成的风险。

有些业务场景希望在id中加入特殊的业务规则名称前缀

例如短信的id：

sms_108678123

奖券的id：

coupon_12908123

需要基于这种算法进行改造，实现支持id注入“基因”的这一特性。

mongodb的主键id设计思路

其实在mongodb里面也有使用到往主键id中注入一些“基因”要素点的这类思路：

mongodb里面没有自增的id。

来吧，自己动手撸一个分布式ID生成器组件

_id是唯一标识的key，value通常我们会设置为objectid对象。

objectid里面包含了时间戳，宿主机的ip，进程号码，自增号

来吧，自己动手撸一个分布式ID生成器组件

在对这几种方案进行了调研之后，于是小林便开始萌生了开发一款专用于自己公司平台的id生成器。

自研主要设计思路

MySQL配置id生成规则，拉取到本地缓存中形成一段本地id，从而降低对于db的访问。

支持集群配置id生成器，能够支持高qps访问和较好的扩容性。

来吧，自己动手撸一个分布式ID生成器组件

配置表如下方所示：

来吧，自己动手撸一个分布式ID生成器组件

建表sql：

1
2
3
4
5
6
7
8
9
10
11
12
13
CREATE TABLE `t_id_builder_config` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `des` varchar(255) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '描述',
  `init_num` bigint(13) DEFAULT NULL COMMENT 'id起步值',
  `current_threshold` bigint(16) DEFAULT NULL COMMENT '当前id所在阶段的阈值',
  `step` int(8) DEFAULT NULL COMMENT 'id递增区间',
  `biz_code` varchar(60) COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT '业务前缀码，如果没有则返回时不携带',
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '乐观锁版本号',
  `is_seq` smallint(2) NOT NULL DEFAULT '0' COMMENT 'id连续递增，0 是  1 不是',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

几个核心设计点：

当同时有多个请求访问mysql获取id配置的时候该如何防止并发问题？

这里我采用了for update的方式加行锁进行读取，同时对于行信息进行更新的时候加入了version版本号信息字段防止更新重复的情况。

假设说更新失败，也会有cas的方式进行重试，重试超过一定次数之后直接中断。

为何不引入redis作为分布式锁来防止并发修改数据库操作？

不希望将该组件变得过于繁杂，减少系统对于第三方的依赖性

假设本地id还没使用完，结果当前服务器宕机了，该如何预防？

每次服务启动都需要更新表的配置，拉去最新的一批id集合到本地，这样就不会出现和之前id冲突的问题了。

本地id集合中如何判断id是否已经使用过？

如果是连续递增型的id，这一步可以忽略，因为本地id每次获取的时候都会比上一个id要大。但是如果是拉取了一段区间的id到本地之后进行随机返回就需要加入bitset作为过滤器了。对于已经使用过的id，则对应bit置为1。如果随机返回的区间id多次都已经被占用了，则超过一定频率之后需要重新拉取id到本地。

来吧，自己动手撸一个分布式ID生成器组件

不通机器的状态表示码该如何设置？

可以通过启动脚本中配置相关参数：

-DidBuilder.index=1001

进行配置，然后通过System.getProperty("idBuilder.index")的方式来获取.

核心代码思路：

接口设计：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
package com.qiyu.tech.id.builder.service;
/**
 * @Author idea
 * @Date created in 11:16 下午 2020/12/17
 */
public interface IdBuilderService {
    /**
     * 根据本地步长度来生成唯一id(区间性递增)
     *
     * @return
     */
    Long unionId(int code);
    /**
     * 对于unionId的算法进行优化(连续性递增)
     *
     * @param code
     * @return
     */
    Long unionSeqId(int code);
    /**
     * 生成包含业务前缀的自增id(区间性递增)
     *
     * @param code
     * @return
     */
    String unionIdStr(int code);
    /**
     * 生成包含业务前缀的自增id(连续性递增)
     *
     * @param code
     * @return
     */
    String unionSeqIdStr(int code);
}

 

具体实现：

复制代码
package com.qiyu.tech.id.builder.service.impl;
import com.qiyu.tech.id.builder.bean.IdBuilderPO;
import com.qiyu.tech.id.builder.bean.LocalSeqId;
import com.qiyu.tech.id.builder.dao.IdBuilderMapper;
import com.qiyu.tech.id.builder.service.IdBuilderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.management.RuntimeErrorException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import static com.qiyu.tech.id.builder.constants.IdTypeConstants.*;
/**
 * @Author idea
 * @Date created in 11:18 下午 2020/12/17
 */
@Service
@Slf4j
public class IdBuilderServiceImpl implements IdBuilderService, InitializingBean {
    private static ConcurrentHashMap<Integer, BitSet> bitSetMap = new ConcurrentHashMap<>();
    private static Map<Integer, IdBuilderPO> idBuilderNotSeqMap;
    private static Map<Integer, IdBuilderPO> idBuilderSeqMap;
    private static Map<Integer, LocalSeqId> localSeqMap;
    private static Map<Integer, Boolean> newBuilderMap;
    private final static Object monitor = new Object();
    @Resource
    private IdBuilderMapper idBuilderMapper;
  
    private int idBuilderIndex;
    @Override
    public Long unionId(int code) {
        //考虑到锁升级问题，在高并发场景下使用synchronized要比cas更佳
        synchronized (monitor) {
            IdBuilderPO idBuilderPO = idBuilderNotSeqMap.get(code);
            if (idBuilderPO == null) {
                return null;
            }
            boolean isNew = newBuilderMap.get(code);
            if (isNew) {
                //预防出现id生成器网络中断问题
                IdBuilderPO newIdBuilderPO = this.refreshIdBuilderConfig(idBuilderPO);
                if (newIdBuilderPO == null) {
                    log.error("[unionId] refreshIdBuilderConfig出现异常");
                    return null;
                }
                idBuilderPO.setCurrentThreshold(newIdBuilderPO.getCurrentThreshold());
                newBuilderMap.put(code, false);
            }
            long initNum = idBuilderPO.getCurrentThreshold();
            int step = idBuilderPO.getStep();
            int randomIndex = RandomUtils.nextInt((int) initNum, (int) initNum + step);
            BitSet bitSet = bitSetMap.get(code);
            if (bitSet == null) {
                bitSet = new BitSet();
                bitSetMap.put(code, bitSet);
            }
            Long id;
            int countTime = 0;
            while (true) {
                boolean indexExist = bitSet.get(randomIndex);
                countTime++;
                if (!indexExist) {
                    bitSet.set(randomIndex);
                    id = Long.valueOf(randomIndex);
                    break;
                }
                //如果重试次数大于了空间的0.75则需要重新获取新的id区间 测试之后得出 循环一千万次随机函数，16gb内存条件下，大约耗时在124ms左右
                if (countTime >= step * 0.75) {
                    //扩容需要修改表配置
                    IdBuilderPO newIdBuilderPO = this.updateIdBuilderConfig(idBuilderPO);
                    if (newIdBuilderPO == null) {
                        log.error("重试超过100次没有更新自增id配置成功");
                        return null;
                    }
                    initNum = newIdBuilderPO.getCurrentThreshold();
                    step = newIdBuilderPO.getStep();
                    idBuilderPO.setCurrentThreshold(initNum);
                    bitSet.clear();
                    log.info("[unionId] 扩容IdBuilder，new idBuilderPO is {}",idBuilderPO);
                }
                randomIndex = RandomUtils.nextInt((int) initNum, (int) initNum + step);
            }
            return id;
        }
    }
    @Override
    public Long unionSeqId(int code) {
        synchronized (monitor) {
            LocalSeqId localSeqId = localSeqMap.get(code);
            IdBuilderPO idBuilderPO = idBuilderSeqMap.get(code);
            if (idBuilderPO == null || localSeqId == null) {
                log.error("[unionSeqId] code 参数有误，code is {}", code);
                return null;
            }
            boolean isNew = newBuilderMap.get(code);
            long result = localSeqId.getCurrentId();
            localSeqId.setCurrentId(result + 1);
            if (isNew) {
                //预防出现id生成器网络中断问题
                IdBuilderPO updateIdBuilderPO = this.refreshIdBuilderConfig(idBuilderPO);
                if (updateIdBuilderPO == null) {
                    log.error("[unionSeqId] refreshIdBuilderConfig出现异常");
                    return null;
                }
                newBuilderMap.put(code, false);
                localSeqId.setCurrentId(updateIdBuilderPO.getCurrentThreshold());
                localSeqId.setNextUpdateId(updateIdBuilderPO.getCurrentThreshold() + updateIdBuilderPO.getStep());
            }
            //需要更新本地步长
            if (localSeqId.getCurrentId() >= localSeqId.getNextUpdateId()) {
                IdBuilderPO newIdBuilderPO = this.updateIdBuilderConfig(idBuilderPO);
                if (newIdBuilderPO == null) {
                    log.error("[unionSeqId] updateIdBuilderConfig出现异常");
                    return null;
                }
                idBuilderPO.setCurrentThreshold(newIdBuilderPO.getCurrentThreshold());
                localSeqId.setCurrentId(newIdBuilderPO.getCurrentThreshold());
                localSeqId.setNextUpdateId(newIdBuilderPO.getCurrentThreshold() + newIdBuilderPO.getStep());
                log.info("[unionSeqId] 扩容IdBuilder，new localSeqId is {}",localSeqId);
            }
            return result;
        }
    }
    /**
     * 刷新id生成器的配置
     *
     * @param idBuilderPO
     */
    private IdBuilderPO refreshIdBuilderConfig(IdBuilderPO idBuilderPO) {
        IdBuilderPO updateResult = this.updateIdBuilderConfig(idBuilderPO);
        if (updateResult == null) {
            log.error("更新数据库配置出现异常,idBuilderPO is {}", idBuilderPO);
            throw new RuntimeErrorException(new Error("更新数据库配置出现异常,idBuilderPO is " + idBuilderPO.toString()));
        }
        return updateResult;
    }
    /**
     * 考虑分布式环境下 多个请求同时更新同一行数据的情况
     *
     * @param idBuilderPO
     * @return
     */
    private IdBuilderPO updateIdBuilderConfig(IdBuilderPO idBuilderPO) {
        int updateResult = -1;
        //假设重试过程中出现网络异常，那么使用cas的时候必须要考虑退出情况 极限情况下更新100次
        for (int i = 0; i < 100; i++) {
            IdBuilderPO newIdBuilderPO = idBuilderMapper.selectOneForUpdate(idBuilderPO.getId());
            updateResult = idBuilderMapper.updateCurrentThreshold(newIdBuilderPO.getCurrentThreshold() + newIdBuilderPO.getStep(), newIdBuilderPO.getId(), newIdBuilderPO.getVersion());
            if (updateResult > 0) {
                return newIdBuilderPO;
            }
        }
        return null;
    }
    @Override
    public String unionIdStr(int code) {
        long id = this.unionId(code);
        IdBuilderPO idBuilderPO = idBuilderNotSeqMap.get(code);
        return idBuilderPO.getBizCode() + idBuilderIndex + id;
    }
    @Override
    public String unionSeqIdStr(int code) {
        long id = this.unionSeqId(code);
        IdBuilderPO idBuilderPO = idBuilderSeqMap.get(code);
        return idBuilderPO.getBizCode() + idBuilderIndex + id;
    }
    @Override
    public void afterPropertiesSet() {
        List<IdBuilderPO> idBuilderPOS = idBuilderMapper.selectAll();
        idBuilderNotSeqMap = new ConcurrentHashMap<>(idBuilderPOS.size());
        newBuilderMap = new ConcurrentHashMap<>(idBuilderPOS.size());
        idBuilderSeqMap = new ConcurrentHashMap<>(idBuilderPOS.size());
        localSeqMap = new ConcurrentHashMap<>(0);
        //每次重启到时候，都需要将之前的上一个区间的id全部抛弃，使用新的步长区间
        for (IdBuilderPO idBuilderPO : idBuilderPOS) {
            if (idBuilderPO.getIsSeq() == NEED_SEQ) {
                idBuilderSeqMap.put(idBuilderPO.getId(), idBuilderPO);
                LocalSeqId localSeqId = new LocalSeqId();
                localSeqId.setNextUpdateId(idBuilderPO.getCurrentThreshold() + idBuilderPO.getStep());
                localSeqId.setCurrentId(idBuilderPO.getCurrentThreshold());
                localSeqMap.put(idBuilderPO.getId(), localSeqId);
            } else {
                idBuilderNotSeqMap.put(idBuilderPO.getId(), idBuilderPO);
            }
            newBuilderMap.put(idBuilderPO.getId(), true);
        }
        this.idBuilderIndex= Integer.parseInt(System.getProperty("idBuilder.index"));
    }
}
复制代码

 

数据库层面设计：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
package com.qiyu.tech.id.builder.dao;
import com.baomidou.mybatisplus.mapper.BaseMapper;
import com.qiyu.tech.id.builder.bean.IdBuilderPO;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import org.apache.ibatis.annotations.Update;
import java.util.List;
/**
 * @Author idea
 * @Date created in 10:17 上午 2020/12/17
 */
@Mapper
public interface IdBuilderMapper extends BaseMapper<IdBuilderPO> {
    @Select("select * from t_id_builder_config")
    List<IdBuilderPO> selectAll();
    @Select("select * from t_id_builder_config where id=#{id} limit 1 for update")
    IdBuilderPO selectOneForUpdate(@Param("id") int id);
    @Update("UPDATE t_id_builder_config set current_threshold=#{currentThreshold},version=version+1 where id=#{id} and version=#{version}")
    Integer updateCurrentThreshold(@Param("currentThreshold") long currentThreshold,@Param("id") int id,@Param("version") int version);
}

　　

这里面我只贴出了部分核心代码，http和rpc访问部分其实大同小异，可以更具自己的需要进行额外定制。

下边我贴出关于controller部分的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package com.qiyu.tech.id.builder.controller;
import com.qiyu.tech.id.builder.service.IdBuilderService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
/**
 * @Author idea
 * @Date created in 4:27 下午 2020/12/17
 */
@RestController
@RequestMapping(value = "id-builder")
public class IdBuilderController {
    @Resource
    private IdBuilderService idBuilderService;
  
    @GetMapping("increase-id")
    public Long increaseId(int code){
        long result = idBuilderService.unionId(code);
        System.out.println(result);
        return result;
    }
    @GetMapping("increase-seq-id")
    public Long increaseSeqId(int code){
        long result = idBuilderService.unionSeqId(code);
        System.out.println(result);
        return result;
    }
    @GetMapping("increase-seq-id-str")
    public String unionSeqIdStr(int code){
        String result = idBuilderService.unionSeqIdStr(code);
        System.out.println(result);
        return result;
    }
    @GetMapping("increase-id-str")
    public String unionIdStr(int code){
        String result = idBuilderService.unionIdStr(code);
        System.out.println(result);
        return result;
    }
}

　　

application.yml配置文件

1
2
3
4
5
6
7
8
mybatis-plus:
  configuration:
    map-underscore-to-camel-case: true
server:
  port: 8082
  tomcat:
    max-threads: 500
    max-connections: 5000

　　

注意需要结合实际机器配置nginx的并发线程数目和tomcat的并发访问参数。

启动类：

ps：这里面的db访问配置是采用了自己封装的一个db工具，其实本质和SpringBoot直接配置jdbc是一样的，可以忽略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
package com.qiyu.tech.id.builder;
import com.qiyu.datasource.annotation.AppDataSource;
import com.qiyu.datasource.enums.DatasourceConfigEnum;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * @Author idea
 * @Date created in 11:16 下午 2020/12/17
 */
@SpringBootApplication(scanBasePackages = "com.qiyu.*")
@AppDataSource(datasourceType = {DatasourceConfigEnum.PROD_DB},defaultType = DatasourceConfigEnum.PROD_DB)
public class IdBuilderApplication {
    public static void main(String[] args) {
        SpringApplication.run(IdBuilderApplication.class,args);
        System.out.println("========== IdBuilderApplication started! =========");
    }
}

　　

测试环节：

通过将服务打包部署在机器上边，同时运行多个服务，通过nginx配置负载均衡，请求到不通的机器上边。

下边是我自己进行压测的一些相关配置参数：

来吧，自己动手撸一个分布式ID生成器组件

压测启动后，后台控制台会打印相关系列参数：

来吧，自己动手撸一个分布式ID生成器组件

当我们需要扩增机器的时候，新加的机器不会对原有发号令机器的id产生影响，可以支持较好的扩容。

每次拉取的本地id段应该设计在多次较好？

这里我们先将本地id段简称为segment。

按照一些过往经验的参考，通常是希望id发号器能够经量减少对于MySQL的访问次数，同时也需要结合实际部门的运维能力进行把控。

假设说我们MySQL是采用了1主2从的方式搭建，当某一从节点挂了，切换新的从节点时候需要消耗大约1分钟时长，那么我们的segment至少需要设计为高峰期QPS * 60 * 1 * 4 ，期间考需要额外考虑一些其他因素，例如网络新的节点切换之后带来的一些网络抖动问题等等，这能够保证即使MySQL出现了故障，本地的segment也可以暂时支撑一段时间。

设计待完善点：

该系统的设计不足点在于，当本地id即将用光的时候需要进行数据库查询，因此这个关键点会拖慢系统的响应时长，所以这里可以采用异步更新配置拉取id的思路进行完善。也就是说当本地id列表剩余只有15%可以使用的时候，便可以进行开启一个异步线程去拉取id列表了。

来吧，自己动手撸一个分布式ID生成器组件

 


来源：https://www.cnblogs.com/javazhiyin/p/14172051.html


相关教程