西安手机网站制作公司,wordpress菜单里想在添加一个菜单,做学校网站的目的,wordpress 适配 手机引言#xff1a;
代码在 https://gitee.com/lbmb/mb-live-app 中 【mb-live-id-generate-provider】 模块里面 如果喜欢 希望大家给给star 项目还在持续更新中。
背景介绍
项目整体架构是 基于springboot 3.0 开发 rpc 调用采用 dubbo 注册配置中心 使用 nacos 采用shardin…引言
代码在 https://gitee.com/lbmb/mb-live-app 中 【mb-live-id-generate-provider】 模块里面 如果喜欢 希望大家给给star 项目还在持续更新中。
背景介绍
项目整体架构是 基于springboot 3.0 开发 rpc 调用采用 dubbo 注册配置中心 使用 nacos 采用sharding-jdbc 来实现分库分表。 基于以上情况 我想生成分布式id。再根据生成的分布式id 存到不同的表中 例如 id 1000 存在 user01表 id 1001 存到 user02表然后sharding-jdbc会根据我们
基础成长
可以学习到多线程、线程池的使用和设计分布式id器的优化策略预加载、类似hashmap扩容
首先我们需要设计一张id策略表
CREATE TABLE t_id_generate_config (id int NOT NULL AUTO_INCREMENT COMMENT 主键 id,remark varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT 描述,next_threshold bigint DEFAULT NULL COMMENT 当前 id 所在阶段的阈\n值,init_num bigint DEFAULT NULL COMMENT 初始化值,current_start bigint DEFAULT NULL COMMENT 当前 id 所在阶段的开始\n值,step int DEFAULT NULL COMMENT id 递增区间,is_seq tinyint DEFAULT NULL COMMENT 是否有序0 无序1 有序,id_prefix varchar(60) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci DEFAULT NULL COMMENT 业务前缀码如果没有则返回\n时不携带,version int NOT NULL DEFAULT 0 COMMENT 乐观锁版本号,create_time datetime DEFAULT CURRENT_TIMESTAMP COMMENT 创建时\n间,update_time datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 更新时间,PRIMARY KEY (id)
) ENGINEInnoDB AUTO_INCREMENT8 DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci;INSERT INTO t_id_generate_config (id, remark,
next_threshold, init_num, current_start, step, is_seq,id_prefix, version, create_time, update_time)
VALUES
(1, 用户 id 生成策略, 10050, 10000, 10000, 50, 0,
user_id, 0, 2023-05-23 12:38:21, 2023-05-23 23:31:45);定义全局变量 变量解析
localSeqIdBOMap 缓存中可分配的分布式id有序idlocalUnSeqIdBOMap 缓存中可分配的分布式id无序idSEQ_ID 1; 判断是否为有序id 的操作扩容 存取 等threadPoolExecutor 移步线程池用来异步动态扩容缓存的可分配id 池semaphoreMap 信号量存放map 防止多线程环境下 多次重复触发异步扩容线程池。参考 ConcurrentHashMap 的扩容 实现ConcurrentHashMap 。UNDATE_RATE动态扩容阀值
private static final Logger LOGGER LoggerFactory.getLogger(IdGenerateService.class);private static MapInteger, LocalSeqIdBO localSeqIdBOMap new ConcurrentHashMapInteger, LocalSeqIdBO();private static MapInteger, LocalUnSeqIdBO localUnSeqIdBOMap new ConcurrentHashMapInteger, LocalUnSeqIdBO();private static final Integer SEQ_ID 1;private static ThreadPoolExecutor threadPoolExecutor new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new ArrayBlockingQueue(1000),new ThreadFactory() {Overridepublic Thread newThread(Runnable r) {Thread thread new Thread(r);thread.setName(id-generate-thread- ThreadLocalRandom.current().nextInt(1000));return null;}});/*** 使用Semaphore 信号量来防止多线程并发 多次刷新id段*/private static MapInteger, Semaphore semaphoreMap new ConcurrentHashMap();/*** id段刷新优化 阈值为0.75 达到百分之75 执行异步任务创建 优化*/private static final float UNDATE_RATE 0.75f;有序 id 生成器 /*** 有序id生成器** param id* return*/Overridepublic Long geSeqId(Integer id) {if (id null) {LOGGER.error([geSeqId] id is error,id is{}, id);return null;}LocalSeqIdBO localSeqIdBO localSeqIdBOMap.get(id);if (localSeqIdBO null) {LOGGER.error([geSeqId] localSeqIdBO is null,id is{}, id);return null;}/*** 异步 预执行刷新id段*/this.refreshLocalSeqId(localSeqIdBO);long andIncrement localSeqIdBO.getCurrentNum().getAndIncrement();if (andIncrement localSeqIdBO.getNextThreshold()) {LOGGER.error([geSeqId] id is over limit,id is{}, id);return null;}// 获取当前id 直增return andIncrement;}代码解读 从数据库读取到对应的方案有序id 和无序id 方案 会有当前 可用的 id段 开始值 和 结束值 以及步长等信息 LocalSeqIdBO localSeqIdBO localSeqIdBOMap.get(id); 预扩容例如当前 可用id段是 1000-1500 判断 1000-1500 的id 被使用超过了 百分之75 就动态将id池 进行扩容
this.refreshLocalSeqId(localSeqIdBO);取出当前已使用的id 最大值 并且进行1 long andIncrement localSeqIdBO.getCurrentNum().getAndIncrement();
// 优化逻辑 如果当前 已经用的id 1 后 超过了当前id池的最大值 则不会生成id。例 当前id池最大是 1500 但是取出的当前已用的id 为 1500 则加一后是1501 超过了id池最大值1500 则不会生成id 继而下一次操作 会扩容id池 if (andIncrement localSeqIdBO.getNextThreshold()) { LOGGER.error(“[geSeqId] id is over limit,id is{}”, id); return null; }
异步刷新本地 id池 /*** 刷新本地有序的id段** param localSeqIdBO*/private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {// 当前 id字段区间值long step localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();/*** 使用Semaphore 信号量来防止多线程并发 多次刷新id段* 防止没扩容完成的时候过多线程进入到 if里面*/if (localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart() step * UNDATE_RATE) {Semaphore semaphore semaphoreMap.get(localSeqIdBO.getId());if (semaphore null) {LOGGER.error(semaphore is null ,id is{}, localSeqIdBO.getId());return;}boolean acquireStatus semaphore.tryAcquire();if (acquireStatus) {// 异步进行同步id字段的操作LOGGER.info(尝试开始进行同步id段的同步操作);threadPoolExecutor.execute(new Runnable() {Overridepublic void run() {try {IdGeneratePO idGeneratePO mapper.selectById(localSeqIdBO.getId());tryUpdateMysqlRecord(idGeneratePO);// 释放semaphore资源} catch (Exception e) {LOGGER.error([refreshLocalSeqId] error is {}, e);} finally {semaphoreMap.get(localSeqIdBO.getId()).release();LOGGER.info(有序id段同步完成id is {}, localSeqIdBO.getId());}}});}}}初次落第一批id数据到id池
spring 容器启动的时候 在初始化Bean后 会回调这个方法 //spring 启动的时候 bean 初始化的时候会回调这里Overridepublic void afterPropertiesSet() throws Exception {ListIdGeneratePO idGeneratePOList mapper.selectAll();for (IdGeneratePO idGeneratePO : idGeneratePOList) {tryUpdateMysqlRecord(idGeneratePO);semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));}}更新数据库里面的 id字段占用位置信息 并且尝试将 已更新的id 段写入到缓存 /*** 更新mysql里面的分布式id的配置信息占用对应id段** param idGeneratePO*/private void tryUpdateMysqlRecord(IdGeneratePO idGeneratePO) {int updateResult mapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());if (updateResult 0) {localIdBoHandler(idGeneratePO);return;}for (int i 0; i 3; i) {IdGeneratePO newIdGeneratePO mapper.selectById(idGeneratePO.getId());updateResult mapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());if (updateResult 0) {localIdBoHandler(idGeneratePO);
// LocalSeqIdBO localSeqIdBO new LocalSeqIdBO();
// AtomicLong atomicLong new AtomicLong(idGeneratePO.getCurrentStart());
// localSeqIdBO.setId(idGeneratePO.getId() );
// localSeqIdBO.setCurrentNum(atomicLong );
// localSeqIdBO.setCurrentStart(idGeneratePO.getCurrentStart() );
// localSeqIdBO.setNextThreshold(idGeneratePO.getNextThreshold() );
// localSeqIdBO.setCurrentNum(atomicLong );
// localSeqIdBOMap.put(localSeqIdBO.getId(),localSeqIdBO);return;}}throw new RuntimeException(表id字段占用失败, 竞争过于激烈 id is : idGeneratePO.getId());}将更新的id段 实际落到缓存
/*** 专门处理如何将id对象放入本地缓存中** param idGeneratePO*/private void localIdBoHandler(IdGeneratePO idGeneratePO) {long currentStart idGeneratePO.getCurrentStart();long nextThreshold idGeneratePO.getNextThreshold();long currentNum currentStart;// 判断数据库取出来的id配置是有序还是无序 1 有序 非 1 无序if (idGeneratePO.getIsSeq() SEQ_ID) {// 有序存储LocalSeqIdBO localSeqIdBO new LocalSeqIdBO();AtomicLong atomicLong new AtomicLong(currentStart);localSeqIdBO.setId(idGeneratePO.getId());localSeqIdBO.setCurrentStart(currentStart);localSeqIdBO.setNextThreshold(nextThreshold);localSeqIdBO.setCurrentNum(atomicLong);localSeqIdBOMap.put(localSeqIdBO.getId(), localSeqIdBO);} else {LocalUnSeqIdBO localUnSeqIdBO new LocalUnSeqIdBO();localUnSeqIdBO.setId(idGeneratePO.getId());localUnSeqIdBO.setCurrentStart(currentStart);localUnSeqIdBO.setNextThreshold(nextThreshold);long begin idGeneratePO.getCurrentStart();long end idGeneratePO.getNextThreshold();ConcurrentLinkedQueue idQueue new ConcurrentLinkedQueue();ArrayListLong idList new ArrayList();for (long i begin; i end; i) {idList.add(i);}// 无序操作将有序集合打乱Collections.shuffle(idList);idQueue.addAll(idList);localUnSeqIdBO.setIdQueue(idQueue);localUnSeqIdBOMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);}}mapper 内容
Mapper
public interface IdGenerateMapper extends BaseMapperIdGeneratePO {// Update(update t_id_generate_config set next_threshold next_threshold step,current_startcurrent_start step , version version 1 where id #{id} and version #{version})
// int updateNewIdCountAndVersion(Param(id) int id, Param(version) int version);Update(update t_id_generate_config set next_thresholdnext_thresholdstep, current_startcurrent_startstep,versionversion1 where id #{id} and version#{version})int updateNewIdCountAndVersion(Param(id) int id, Param(version) int version);Select(select * from t_id_generate_config)ListIdGeneratePO selectAll();
}
