基于Mysql的Sequence实现方法

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

这边之前一直是使用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=InnoDBDEFAULT CHARACTER SET=utf8 COLLATE=utf8_general_ciROW_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='流水号生成表';

/** * <p></p> * @author coderzl * @Title MysqlSequence * @Description 基于mysql数据库实现的序列 * @date 2017/6/6 23:03 */public interface MysqlSequence { /**  * <p>  * 获取指定sequence的序列号  * </p>  * @param seqName sequence名  * @return String 序列号  */ public String nextVal(String seqName);}

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

/** * <p></p> * * @author coderzl * @Title SequenceRange * @Description 序列区间，用于缓存序列 * @date 2017/6/6 22:58 */ @Datapublic 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); } /**  * <p>Gets and increment</p>  *  * @return   */ public long getAndIncrement() {  long currentValue = value.getAndIncrement();  if (currentValue > max) {   over = true;   return -1;  }  return currentValue; }}

对应数据库记录

@Datapublic 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;  }}

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

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<MysqlSequenceBo> getAll();}

<?xml version="1.0" encoding="UTF-8" ?><!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" ><mapper namespace="com.xxxxx.core.sequence.impl.dao.MysqlSequenceDAO" > <resultMap id="BaseResultMap" type="com.xxxxx.core.sequence.impl.MysqlSequenceBo" >  <result column="SEQ_NAME" property="seqName" jdbcType="VARCHAR" />  <result column="SEQ_VALUE" property="seqValue" jdbcType="BIGINT" />  <result column="MIN_VALUE" property="minValue" jdbcType="BIGINT" />  <result column="MAX_VALUE" property="maxValue" jdbcType="BIGINT" />  <result column="STEP" property="step" jdbcType="BIGINT" />  <result column="TM_CREATE" property="tmCreate" jdbcType="TIMESTAMP" />  <result column="TM_SMP" property="tmSmp" jdbcType="TIMESTAMP" /> </resultMap> <delete id="delSequence" parameterType="java.lang.String" >  delete from t_pub_sequence  where SEQ_NAME = #{seqName,jdbcType=VARCHAR} </delete> <insert id="createSequence" parameterType="com.xxxxx.core.sequence.impl.MysqlSequenceBo" >  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()) </insert> <update id="updSequence" parameterType="com.xxxxx.core.sequence.impl.MysqlSequenceBo" >  update t_pub_sequence  set SEQ_VALUE = #{newValue,jdbcType=BIGINT}  where SEQ_NAME = #{seqName,jdbcType=VARCHAR} and SEQ_VALUE = #{oldValue,jdbcType=BIGINT} </update> <select id="getAll" resultMap="BaseResultMap" >  select SEQ_NAME, SEQ_VALUE, MIN_VALUE, MAX_VALUE, STEP  from t_pub_sequence </select> <select id="getSequence" resultMap="BaseResultMap" >  select SEQ_NAME, SEQ_VALUE, MIN_VALUE, MAX_VALUE, STEP  from t_pub_sequence  where SEQ_NAME = #{seqName,jdbcType=VARCHAR} </select></mapper>

@Repository("mysqlSequence")public class MysqlSequenceImpl implements MysqlSequence{ @Autowired private MysqlSequenceFactory mysqlSequenceFactory; /**  * <p>  * 获取指定sequence的序列号  * </p>  *  * @param seqName sequence名  * @return String 序列号  * @author coderzl  */ @Override public String nextVal(String seqName) {  return Objects.toString(mysqlSequenceFactory.getNextVal(seqName)); }}

工厂只做了两件事

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

•获取sequence的下一个值

@Componentpublic class MysqlSequenceFactory { private final Lock lock = new ReentrantLock(); /** */ private Map<String,MysqlSequenceHolder> 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(); } /**  * <p> 加载表中所有sequence，完成初始化 </p>  * @return void  * @author coderzl  */ private void initAll(){  try {   lock.lock();   List<MysqlSequenceBo> 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();  } } /**  * <p> </p>  * @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(); }}

•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; /**  * <p> 构造方法 </p>  * @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(); } /**  * <p> 初始化 </p>  * @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; } /**  * <p> 获取下一个序列号 </p>  * @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; } /**  * <p> 初始化当前这条记录 </p>  * @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"); } /**  * <p> 检查新值是否合法 新的当前值是否在最大最小值之间</p>  * @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; } /**  * <p> 重置sequence当前值 ：当前sequence达到最大值时，重新从最小值开始 </p>  * @Title resetCurrentValue  * @param curBo  * @return long  * @author coderzl  */ private long resetCurrentValue(MysqlSequenceBo curBo){  return curBo.getMinValue(); } /**  * <p> 缓存区间使用完毕时，重新读取数据库记录，缓存新序列段 </p>  * @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拼接方法

@Servicepublic 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()<len){   for(int i=0;i<len-seq.length();i++){    str +="0";    }     }  res =str+seq;  return res;   }}

更多相关文章

随机推荐