基于Mysql的Sequence实现方法

数据运维 2023-04-28 醒在深海的猫 手机阅读

团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--Mysql。 这边之前一直是使用oracle，各种订单号、流水号、批次号啥的，都是直接使用oracle的sequence提供的数字序列号。

                        <p>团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--Mysql。</p>

这边之前一直是使用oracle，各种订单号、流水号、批次号啥的，都是直接使用oracle的sequence提供的数字序列号。现在数据库更换成Mysql了，显然以前的老方法不能适用了。

需要新写一个：

•分布式场景使用

•满足一定的并发要求

找了一些相关的资料，发现mysql这方面的实现，原理都是一条数据库记录，不断update它的值。然后大部分的实现方案，都用到了函数。

贴一下网上的代码：

基于mysql函数实现

表结构

CREATE TABLE t_sequence ( sequence_name varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT '序列名称' , value int(11) NULL DEFAULT NULL COMMENT '当前值' , PRIMARY KEY (sequence_name) ) ENGINE=InnoDB DEFAULT CHARACTER SET=utf8 COLLATE=utf8_general_ci ROW_FORMAT=COMPACT ;

获取下一个值

CREATE DEFINER = root@localhost FUNCTION nextval(sequence_name varchar(64)) RETURNS int(11) BEGIN declare current integer; set current = 0;

update t_sequence t set t.value = t.value + 1 where t.sequence_name = sequence_name; select t.value into current from t_sequence t where t.sequence_name = sequence_name;

return current; end;

并发场景有可能会出问题，虽然可以在业务层加锁，但分布式场景就无法保证了，然后效率应该也不会高。

自己实现一个，java版

原理：

•读取一条记录，缓存一个数据段，如：0-100，将记录的当前值从0修改为100

•数据库乐观锁更新，允许重试

•读取数据从缓存中读取，用完再读取数据库

不废话，上代码：

基于java实现

表结构

每次update，都是将SEQ_VALUE设置为SEQ_VALUE+STEP

CREATE TABLE t_pub_sequence ( SEQ_NAME varchar(128) CHARACTER SET utf8 NOT NULL COMMENT '序列名称', SEQ_VALUE bigint(20) NOT NULL COMMENT '目前序列值', MIN_VALUE bigint(20) NOT NULL COMMENT '最小值', MAX_VALUE bigint(20) NOT NULL COMMENT '最大值', STEP bigint(20) NOT NULL COMMENT '每次取值的数量', TM_CREATE datetime NOT NULL COMMENT '创建时间', TM_SMP datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '修改时间', PRIMARY KEY (SEQ_NAME) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='流水号生成表';

sequence接口

/**

  • @author coderzl
  • @Title MysqlSequence
  • @Description 基于mysql数据库实现的序列
  • @date 2017/6/6 23:03 */ public interface MysqlSequence { /**

    • 获取指定sequence的序列号

    • @param seqName sequence名
    • @return String 序列号 */ public String nextVal(String seqName); }

序列区间

用于本地缓存一段序列，从min到max区间

/**

  • @author coderzl
  • @Title SequenceRange
  • @Description 序列区间，用于缓存序列

  • @date 2017/6/6 22:58 */ @Data public class SequenceRange { private final long min; private final long max; /* / private final AtomicLong value; /* 是否超限 / private volatile boolean over = false;

    /**

    • 构造.
    • @param min
    • @param max */ public SequenceRange(long min, long max) { this.min = min; this.max = max; this.value = new AtomicLong(min); }

    /**

    • Gets and increment

    • @return */ public long getAndIncrement() { long currentValue = value.getAndIncrement(); if (currentValue > max) { over = true; return -1; }

    return currentValue; }

}

BO

对应数据库记录

@Data public class MysqlSequenceBo { /**

  • seq名 */ private String seqName; /**
  • 当前值 */ private Long seqValue; /**
  • 最小值 */ private Long minValue; /**
  • 最大值 */ private Long maxValue; /**

  • 每次取值的数量 */ private Long step; /* / private Date tmCreate; /* / private Date tmSmp;

    public boolean validate(){ //一些简单的校验。如当前值必须在最大最小值之间。step值不能大于max与min的差 if (StringUtil.isBlank(seqName) || minValue < 0 || maxValue <= 0 || step <= 0 || minValue >= maxValue || maxValue - minValue <= step ||seqValue < minValue || seqValue > maxValue ) { return false; } return true; } }

DAO

增删改查，其实就用到了改和查

public interface MysqlSequenceDAO { /**

  • */ public int createSequence(MysqlSequenceBo bo);

    public int updSequence(@Param("seqName") String seqName, @Param("oldValue") long oldValue ,@Param("newValue") long newValue);

    public int delSequence(@Param("seqName") String seqName);

    public MysqlSequenceBo getSequence(@Param("seqName") String seqName);

    public List getAll(); }

Mapper

http://mybatis.org/dtd/mybatis-3-mapper.dtd" > delete from t_pub_sequence where SEQ_NAME = #{seqName,jdbcType=VARCHAR} insert into t_pub_sequence (SEQ_NAME,SEQ_VALUE,MIN_VALUE,MAX_VALUE,STEP,TM_CREATE) values (#{seqName,jdbcType=VARCHAR}, #{seqValue,jdbcType=BIGINT}, #{minValue,jdbcType=BIGINT}, #{maxValue,jdbcType=BIGINT}, #{step,jdbcType=BIGINT}, now()) update t_pub_sequence set SEQ_VALUE = #{newValue,jdbcType=BIGINT} where SEQ_NAME = #{seqName,jdbcType=VARCHAR} and SEQ_VALUE = #{oldValue,jdbcType=BIGINT}

接口实现

@Repository("mysqlSequence") public class MysqlSequenceImpl implements MysqlSequence{

@Autowired private MysqlSequenceFactory mysqlSequenceFactory; /**

  • 获取指定sequence的序列号

  • @param seqName sequence名
  • @return String 序列号
  • @author coderzl */ @Override public String nextVal(String seqName) { return Objects.toString(mysqlSequenceFactory.getNextVal(seqName)); } }

工厂

工厂只做了两件事

•服务启动的时候，初始化数据库中所有sequence【完成序列区间缓存】

•获取sequence的下一个值

@Component public class MysqlSequenceFactory {

private final Lock lock = new ReentrantLock();

/* / private Map holderMap = new ConcurrentHashMap<>();

@Autowired private MysqlSequenceDAO msqlSequenceDAO; /* 单个sequence初始化乐观锁更新失败重试次数 / @Value("${seq.init.retry:5}") private int initRetryNum; /* 单个sequence更新序列区间乐观锁更新失败重试次数 / @Value("${seq.get.retry:20}") private int getRetryNum;

@PostConstruct private void init(){ //初始化所有sequence initAll(); }

/**

  • 加载表中所有sequence，完成初始化

  • @return void

  • @author coderzl */ private void initAll(){ try { lock.lock(); List boList = msqlSequenceDAO.getAll(); if (boList == null) { throw new IllegalArgumentException("The sequenceRecord is null!"); } for (MysqlSequenceBo bo : boList) { MysqlSequenceHolder holder = new MysqlSequenceHolder(msqlSequenceDAO, bo,initRetryNum,getRetryNum); holder.init(); holderMap.put(bo.getSeqName(), holder); } }finally { lock.unlock(); } }

    /**

  • @param seqName
  • @return long
  • @author coderzl */ public long getNextVal(String seqName){ MysqlSequenceHolder holder = holderMap.get(seqName); if (holder == null) { try { lock.lock(); holder = holderMap.get(seqName); if (holder != null){ return holder.getNextVal(); } MysqlSequenceBo bo = msqlSequenceDAO.getSequence(seqName); holder = new MysqlSequenceHolder(msqlSequenceDAO, bo,initRetryNum,getRetryNum); holder.init(); holderMap.put(seqName, holder); }finally { lock.unlock(); } } return holder.getNextVal(); }

}

单一sequence的Holder

•init() 初始化 其中包括参数校验，数据库记录更新，创建序列区间

•getNextVal() 获取下一个值

public class MysqlSequenceHolder {

private final Lock lock = new ReentrantLock();

/* seqName / private String seqName;

/* sequenceDao / private MysqlSequenceDAO sequenceDAO;

private MysqlSequenceBo sequenceBo; / */ private SequenceRange sequenceRange; /* 是否初始化 / private volatile boolean isInitialize = false; / sequence初始化重试次数 */ private int initRetryNum; /* sequence获取重试次数 / private int getRetryNum;

/**

  • 构造方法

  • @Title MysqlSequenceHolder
  • @param sequenceDAO
  • @param sequenceBo
  • @param initRetryNum 初始化时，数据库更新失败后重试次数
  • @param getRetryNum 获取nextVal时，数据库更新失败后重试次数
  • @return

  • @author coderzl */ public MysqlSequenceHolder(MysqlSequenceDAO sequenceDAO, MysqlSequenceBo sequenceBo,int initRetryNum,int getRetryNum) { this.sequenceDAO = sequenceDAO; this.sequenceBo = sequenceBo; this.initRetryNum = initRetryNum; this.getRetryNum = getRetryNum; if(sequenceBo != null) this.seqName = sequenceBo.getSeqName(); }

    /**

  • 初始化

  • @Title init
  • @param
  • @return void

  • @author coderzl */ public void init(){ if (isInitialize == true) { throw new SequenceException("[" + seqName + "] the MysqlSequenceHolder has inited"); } if (sequenceDAO == null) { throw new SequenceException("[" + seqName + "] the sequenceDao is null"); } if (seqName == null || seqName.trim().length() == 0) { throw new SequenceException("[" + seqName + "] the sequenceName is null"); } if (sequenceBo == null) { throw new SequenceException("[" + seqName + "] the sequenceBo is null"); } if (!sequenceBo.validate()){ throw new SequenceException("[" + seqName + "] the sequenceBo validate fail. BO:"+sequenceBo); } // 初始化该sequence try { initSequenceRecord(sequenceBo); } catch (SequenceException e) { throw e; } isInitialize = true; }

    /**

  • 获取下一个序列号

  • @Title getNextVal
  • @param
  • @return long

  • @author coderzl */ public long getNextVal(){ if(isInitialize == false){ throw new SequenceException("[" + seqName + "] the MysqlSequenceHolder not inited"); } if(sequenceRange == null){ throw new SequenceException("[" + seqName + "] the sequenceRange is null"); } long curValue = sequenceRange.getAndIncrement();

    if(curValue == -1){ try{ lock.lock(); curValue = sequenceRange.getAndIncrement(); if(curValue != -1){ return curValue; } sequenceRange = retryRange(); curValue = sequenceRange.getAndIncrement(); }finally { lock.unlock(); } } return curValue; }

    /**

  • 初始化当前这条记录

  • @Title initSequenceRecord
  • @Description
  • @param sequenceBo
  • @return void

  • @author coderzl */ private void initSequenceRecord(MysqlSequenceBo sequenceBo){ //在限定次数内，乐观锁更新数据库记录 for(int i = 1; i < initRetryNum; i++){ //查询bo MysqlSequenceBo curBo = sequenceDAO.getSequence(sequenceBo.getSeqName()); if(curBo == null){ throw new SequenceException("[" + seqName + "] the current sequenceBo is null"); } if (!curBo.validate()){ throw new SequenceException("[" + seqName + "] the current sequenceBo validate fail"); } //改变当前值 long newValue = curBo.getSeqValue()+curBo.getStep(); //检查当前值 if(!checkCurrentValue(newValue,curBo)){ newValue = resetCurrentValue(curBo); } int result = sequenceDAO.updSequence(sequenceBo.getSeqName(),curBo.getSeqValue(),newValue); if(result > 0){ sequenceRange = new SequenceRange(curBo.getSeqValue(),newValue - 1); curBo.setSeqValue(newValue); this.sequenceBo = curBo; return; }else{ continue; } } //限定次数内，更新失败，抛出异常 throw new SequenceException("[" + seqName + "] sequenceBo update error"); }

    /**

  • 检查新值是否合法 新的当前值是否在最大最小值之间

  • @param curValue
  • @param curBo
  • @return boolean

  • @author coderzl */ private boolean checkCurrentValue(long curValue,MysqlSequenceBo curBo){ if(curValue > curBo.getMinValue() && curValue <= curBo.getMaxValue()){ return true; } return false; }

    /**

  • 重置sequence当前值 ：当前sequence达到最大值时，重新从最小值开始

  • @Title resetCurrentValue
  • @param curBo
  • @return long

  • @author coderzl */ private long resetCurrentValue(MysqlSequenceBo curBo){ return curBo.getMinValue(); }

    /**

  • 缓存区间使用完毕时，重新读取数据库记录，缓存新序列段

  • @Title retryRange
  • @param SequenceRange

  • @author coderzl */ private SequenceRange retryRange(){ for(int i = 1; i < getRetryNum; i++){ //查询bo MysqlSequenceBo curBo = sequenceDAO.getSequence(sequenceBo.getSeqName()); if(curBo == null){ throw new SequenceException("[" + seqName + "] the current sequenceBo is null"); } if (!curBo.validate()){ throw new SequenceException("[" + seqName + "] the current sequenceBo validate fail"); } //改变当前值 long newValue = curBo.getSeqValue()+curBo.getStep(); //检查当前值 if(!checkCurrentValue(newValue,curBo)){ newValue = resetCurrentValue(curBo); } int result = sequenceDAO.updSequence(sequenceBo.getSeqName(),curBo.getSeqValue(),newValue); if(result > 0){ sequenceRange = new SequenceRange(curBo.getSeqValue(),newValue - 1); curBo.setSeqValue(newValue); this.sequenceBo = curBo; return sequenceRange; }else{ continue; } } throw new SequenceException("[" + seqName + "] sequenceBo update error");

    } }

总结

•当服务重启或异常的时候，会丢失当前服务所缓存且未用完的序列

•分布式场景，多个服务同时初始化，或者重新获取sequence时，乐观锁会保证彼此不冲突。A服务获取0-99，B服务会获取100-199，以此类推

•当该sequence获取较为频繁时，增大step值，能提升性能。但同时服务异常时，损失的序列也较多

•修改数据库里sequence的一些属性值，比如step，max等，再下一次从数据库获取时，会启用新的参数

•sequence只是提供了有限个序列号（最多max-min个），达到max后，会循环从头开始。

•由于sequence会循环，所以达到max后，再获取，就不会唯一。建议使用sequence来做业务流水号时，拼接时间。如：20170612235101+序列号

业务id拼接方法

@Service public class JrnGeneratorService { private static final String SEQ_NAME = "T_SEQ_TEST";

/* sequence服务 / @Autowired private MySqlSequence mySqlSequence;

public String generateJrn() { try { String sequence = mySqlSequence.getNextValue(SEQ_NAME); sequence = leftPadding(sequence,8); Calendar calendar = Calendar.getInstance(); SimpleDateFormat sDateFormat = new SimpleDateFormat("yyyyMMddHHmmss"); String nowdate = sDateFormat.format(calendar.getTime()); nowdate.substring(4, nowdate.length()); String jrn = nowdate + sequence + RandomUtil.getFixedLengthRandom(6);//10位时间＋8位序列 + 6位随机数＝24位流水号 return jrn; } catch (Exception e) { //TODO } }

private String leftPadding(String seq,int len){ String res =""; String str =""; if(seq.length() } res =str+seq; return res;

}

}

以上这篇基于Mysql的Sequence实现方法就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持每日运维。