网站关键词优化软件,小红书推广模式,wordpress如何上传主题,设计制作一个企业类型网站文章目录 本系列前言号段模式双buffer优化biz优化动态step源码走读 雪花算法怎么设置workerId解决时钟回拨源码走读 总结 本系列
漫谈分布式唯一ID分布式唯一ID生成#xff08;二#xff09;#xff1a;leaf#xff08;本文#xff09;分布式唯一ID生成#xff08;三二leaf本文分布式唯一ID生成三uid-generator分布式唯一ID生成四tinyid
前言
本文将介绍leaf号段模式和雪花算法模式的设计原理并走读源码 源码地址https://github.com/Meituan-Dianping/Leaf leaf提供了 leaf server业务只管调leaf server的接口获取IDleaf serve内部根据号段或雪花算法生成ID而不是业务服务自己去请求数据库生成id或自己根据雪花算法生成id 号段模式
在使用db自增主键的基础上从每次获取ID都要读写一次db改成一次获取一批ID
各个业务的记录通过 biz_tag 区分每个业务的ID上次分配到哪了在一张表中用一条记录表示
表结构如下
CREATE TABLE leaf_alloc (biz_tag varchar(128) NOT NULL DEFAULT , -- your biz unique namemax_id bigint(20) NOT NULL DEFAULT 1,step int(11) NOT NULL,description varchar(256) DEFAULT NULL,update_time timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (biz_tag)
) ENGINEInnoDB;重要字段为
biz_tag标识业务max_id目前分配到的最大值-1也是 下一个号段的起始值step每次分配号段的长度
如果step1000当这1000个ID消耗完后才会读写一次DB对DB的压力降为原来的 1/1000
当缓存中没有ID时需要从db获取号段在事务中执行如下2条sql
UPDATE leaf_alloc SET max_id max_id step WHERE biz_tag #{tag}SELECT biz_tag, max_id, step FROM leaf_alloc WHERE biz_tag #{tag}然后加载到本地
N个server执行上述操作对外提供http接口用于生成id整体架构如下图所示 优点
Leaf服务可以很方便的线性扩展例如按照biz_tag分库分表ID是趋势递增的64位数字满足上述数据库存储的类型要求容灾性高Leaf服务内部有号段缓存即使DB宕机短时间内仍能正常对外提供服务易接入可自定义初始max_id的值方便业务从原有的ID方式上迁移过来
缺点 业务上不够安全ID近似于严格递增会泄露发号数量的信息 TP999显著比其他TP值大当号段使用完之后还是会hang在更新数据库的I/O上 以step1000为例99.9%的请求分配ID都非常快0.1%的请求会比较慢读写一次db平均5ms如果恰好遇到db抖动耗时能到几秒 单点问题DB如果宕机会造成整个系统不可用 网络IO开销大client每获取一个id都要对leaf server发起一次http调用 双buffer优化
针对TP999大的问题Leaf号段模式做了一些优化在内存中维护两个号段在当前号段消费到一定百分比时就 异步去db加载下一个号段 到内存中 这样当前号段用完后就能马上切换到下一个号段 biz优化
对于每个请求都需要校验参数中的biz是否合法。如果每次都去db查下biz在leaf_alloc表是否存在性能开销大且没必要
leaf在实例启动时将全量biz都查出来放到本地缓存中之后每隔60s都会刷新一次这样校验biz是否合法都用本地map判断性能极高
缺点是最多延迟1分钟才新增的biz才生效
也就牺牲一点一致性换取超高的性能 动态step
如果每次获取号段的长度step是固定的但流量不是固定的如果流量增加 10 倍每个号段很快就被用完了仍然有可能导致数据库压力较大
同时也降低了可用性例如本来能在DB不可用的情况下维持10分钟正常工作那么如果流量增加10倍就只能维持1分钟正常工作了
因此leaf中每次从db加载号段时加载多少ID并不是固定的
如果qps高就可以一次多加载点减少调db的次数如果qps低可以一次少加载点。否则在缓存中的号段迟迟消耗不完的情况下会导致更新DB的新号段与前一次下发的号段ID跨度过大
leaf的策略为每次更新buffer时动态维护step当需要从db加载号段时计算距离上次从db加载过去了多久
小于15mins获取的号段长度翻倍1530mins获取的号段长度和上次一样大于30mins获取的号段长度减半 源码走读
初始化
public boolean init() {// 将所有biz加载到内存updateCacheFromDb();initOK true;// 后台每1s刷新一次内存中的bizupdateCacheFromDbAtEveryMinute();return initOK;
}获取ID流程如下
public Result get(final String key) {if (!initOK) {return new Result(EXCEPTION_ID_IDCACHE_INIT_FALSE, Status.EXCEPTION);}// 校验biz是否合法if (cache.containsKey(key)) {SegmentBuffer buffer cache.get(key);if (!buffer.isInitOk()) {synchronized (buffer) {if (!buffer.isInitOk()) {try {// 号段未初始化从db加载号段updateSegmentFromDb(key, buffer.getCurrent());buffer.setInitOk(true);} catch (Exception e) {}}}}// 从号段获取IDreturn getIdFromSegmentBuffer(cache.get(key));}return new Result(EXCEPTION_ID_KEY_NOT_EXISTS, Status.EXCEPTION);
}getIdFromSegmentBuffer方法
public Result getIdFromSegmentBuffer(final SegmentBuffer buffer) {while (true) {buffer.rLock().lock();try {final Segment segment buffer.getCurrent();// 如果当前号段已经用了10%异步去加载下一个号段if (!buffer.isNextReady() (segment.getIdle() 0.9 * segment.getStep()) buffer.getThreadRunning().compareAndSet(false, true)) {service.execute(new Runnable() {Overridepublic void run() {// 加载下一个号段});}// 当前号段还没用完从当前号段获取long value segment.getValue().getAndIncrement();if (value segment.getMax()) {return new Result(value, Status.SUCCESS);}} finally {buffer.rLock().unlock();}// 到这说明当前号段用完了waitAndSleep(buffer);buffer.wLock().lock();try {// 再次检查当前号段因为可能别的线程加载了final Segment segment buffer.getCurrent();long value segment.getValue().getAndIncrement();if (value segment.getMax()) {return new Result(value, Status.SUCCESS);}// 切换到下一个号段重新执行while循环获取if (buffer.isNextReady()) {buffer.switchPos();buffer.setNextReady(false);} else {// 两个号段都不可用报错return new Result(EXCEPTION_ID_TWO_SEGMENTS_ARE_NULL, Status.EXCEPTION);}} finally {buffer.wLock().unlock();}}}最后看下怎么从db加载号段
public void updateSegmentFromDb(String key, Segment segment) {SegmentBuffer buffer segment.getBuffer();LeafAlloc leafAlloc;// ...// 动态调整下一次的steplong duration System.currentTimeMillis() - buffer.getUpdateTimestamp();int nextStep buffer.getStep();if (duration SEGMENT_DURATION) {if (nextStep * 2 MAX_STEP) {} else {nextStep nextStep * 2;}} else if (duration SEGMENT_DURATION * 2) {} else {nextStep nextStep / 2 buffer.getMinStep() ? nextStep / 2 : nextStep;}LeafAlloc temp new LeafAlloc();temp.setKey(key);temp.setStep(nextStep);// 从db获取下一个号段leafAlloc dao.updateMaxIdByCustomStepAndGetLeafAlloc(temp);buffer.setUpdateTimestamp(System.currentTimeMillis());buffer.setStep(nextStep);buffer.setMinStep(leafAlloc.getStep());// 加载到内存中long value leafAlloc.getMaxId() - buffer.getStep();segment.getValue().set(value);segment.setMax(leafAlloc.getMaxId());segment.setStep(buffer.getStep());
}内存中Segment结构主要有以下字段
value下一个要分配的IDmax当前号段的最大边界
每次从Segment中分配ID时返回value的值即可并把value 雪花算法
号段模式的ID很接近严格递增如果在订单场景可以根据ID猜到一天的订单量。此时就可以用雪花算法模式
leaf在每一位的分配和标准snowflake一致 最高位符号位为0 接下来41位毫秒级时间戳 存储当前时间距离2010年某一天的差值 接下来10位workerId 最后12位每一毫秒内的序列号
每到新的毫秒时每一毫秒内的序列号不是从0开始而是从100以内的一个随机数开始
为啥这么设计试想如果每一秒都从0开始在qps低的情况下每一毫秒只产生1个id那么最末尾永远是0。如果对ID取模分表就会永远在第0号表造成数据分布不均 怎么设置workerId
对于workerID的分配当服务集群数量较小的情况下完全可以手动配置。如果服务规模较大动手配置成本太高。于是leaf用zookeeper 自动获取workerId流程如下 以自己的 ip:port为key去zk建立持久顺序节点以zk生成的 自增序号为workerId 创建的节点最后两级路径为/forever/ip:port-序列号 如果zk中已经有自己ip:port的节点就 复用 其workerId 怎么判断拉取/forever下所有节点每个节点的格式为ipport-序列号判断每个节点中-前面的ipport是不是等于自己如果等于取-后面的序列号作为workerId只有leaf server会创建zk节点因此节点数量可控为啥可以复用不会在同一时刻有相同ip:port的两个实例因此复用一定不会发生冲突
这种workerId分配策略能保证唯一性吗能
如果 ip:port 不同在zk中一定是两个不同的序列号因此不会冲突一个集群中不可能同时存在ip:port相同的两个机器
每个leaf server的ip:port最好手动指定或者部署在ip不会变化的环境中
高可用workerId会存到本地文件这样遇到极端情况leaf server服务重启且zk也宕机时也不影响使用 解决时钟回拨
雪花算法严格依赖时间如果发生了时钟回拨就可能导致ID重复因此需要监测是否发生了时钟回拨并处理在服务启动和运行时都会检测
在服务启动时检测时间是否回退
leaf server运行时每隔3s会上自己的当前时间到zk节点中启动时校验当前时间不能小于 zk中最近一次上报的时间 官方文档还提到如果是第一次启动还会和其他leaf server校准时间。但源码中没找到这部分应该是不需要做这个校准已删除 在运行时检测时间是否回退 全局维护了上次获取ID时的时间戳lastTimestamp 如果当前时间 now lastTimestamp说明发生了时钟回拨 回拨了超过5ms返回报错回拨了5ms内sleep一会直到赶上上次时间
如果zk宕机导致定时上报没有成功同时又发生了时钟回拨且leaf server宕机。此时leaf server启动时可能产生和之前重复的ID。因此需要做好监控告警zk的高可用
如果3s内没上报leaf server宕机了然后时钟回退了2s此时根据zk的时间检测不出来发生了时钟回退也会造成ID重复。解决方法就是等一段时间才重启机器保证等待的时间比回拨的时间长就行 源码走读
初始化
public boolean init() {try {CuratorFramework curator createWithOptions(connectionString, new RetryUntilElapsed(1000, 4), 10000, 6000);curator.start();Stat stat curator.checkExists().forPath(PATH_FOREVER);if (stat null) {//不存在根节点,机器第一次启动,创建/snowflake/ip:port-000000000,并上传数据zk_AddressNode createNode(curator);//worker id 默认是0存到本地文件updateLocalWorkerID(workerID);//每3s上报本机时间给forever节点ScheduledUploadData(curator, zk_AddressNode);return true;} else {MapString, Integer nodeMap Maps.newHashMap();//ip:port-00001MapString, String realNode Maps.newHashMap();//ip:port-(ipport-000001)//存在根节点,先检查是否有属于自己的节点ListString keys curator.getChildren().forPath(PATH_FOREVER);for (String key : keys) {String[] nodeKey key.split(-);realNode.put(nodeKey[0], key);nodeMap.put(nodeKey[0], Integer.parseInt(nodeKey[1]));}Integer workerid nodeMap.get(listenAddress);if (workerid ! null) {//有自己的节点,zk_AddressNodeip:portzk_AddressNode PATH_FOREVER / realNode.get(listenAddress);workerID workerid;//启动worder时使用会使用// 当前时间不能小于 zk中最近一次上报的时间if (!checkInitTimeStamp(curator, zk_AddressNode)) {throw new CheckLastTimeException(init timestamp check error,forever node timestamp gt this node time);}// 每3s上报时间doService(curator);// 将workerId写到本地updateLocalWorkerID(workerID);} else {//新启动的节点,创建持久节点 ,不用check时间String newNode createNode(curator);zk_AddressNode newNode;String[] nodeKey newNode.split(-);workerID Integer.parseInt(nodeKey[1]);doService(curator);updateLocalWorkerID(workerID);}}} catch (Exception e) {// zk不可用从本地文件加载workerIdtry {Properties properties new Properties();properties.load(new FileInputStream(new File(PROP_PATH.replace({port}, port ))));workerID Integer.valueOf(properties.getProperty(workerID));} catch (Exception e1) {return false;}}return true;}获取ID
public synchronized Result get(String key) {long timestamp timeGen();// 发生了时钟回拨if (timestamp lastTimestamp) {long offset lastTimestamp - timestamp;// 回拨了5ms内sleep一会if (offset 5) {try {wait(offset 1);timestamp timeGen();if (timestamp lastTimestamp) {return new Result(-1, Status.EXCEPTION);}} catch (InterruptedException e) {return new Result(-2, Status.EXCEPTION);}// 回拨了超过5ms返回报错} else {return new Result(-3, Status.EXCEPTION);}}// 和上次在同一毫秒if (lastTimestamp timestamp) {sequence (sequence 1) sequenceMask;if (sequence 0) {//seq 为0表示是当前ms已经超过4096个ID了// 需要sleep一会下一毫秒时间开始对seq做随机sequence RANDOM.nextInt(100);timestamp tilNextMillis(lastTimestamp);}} else {//如果是新的ms, 对seq做随机sequence RANDOM.nextInt(100);}lastTimestamp timestamp;long id ((timestamp - twepoch) timestampLeftShift) | (workerId workerIdShift) | sequence;return new Result(id, Status.SUCCESS);}总结
leaf提供两种分布式ID生成策略 号段模式 每次从db获取一批ID而不是一个ID减少调DB的频率用双buffer解决TP999耗时高的问题在内存判断参数biz是否合法提高校验性能使用动态step解决突发流量造成对db压力仍然大的问题 雪花算法模式 配合ZK做到动态获取workerId解决海量机器的 workId 维护问题也能保证正确性同时不会有两个leaf server拥有相同的workerId在服务启动时和运行时都校验是否发生了时钟回拨。不过服务启动时的校验有时会失效最好sleep一段时间再重启这段时间要大于时钟回拨的时间
