第三方做网站/武汉新一轮疫情

团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--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=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)
BEGINdeclare 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接口

/*** <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*/@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);}/*** <p>Gets and increment</p>** @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<MysqlSequenceBo> getAll();
}

Mapper

<?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_sequencewhere 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_sequenceset 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, STEPfrom t_pub_sequence</select><select id="getSequence" resultMap="BaseResultMap" >select SEQ_NAME, SEQ_VALUE, MIN_VALUE, MAX_VALUE, STEPfrom t_pub_sequencewhere SEQ_NAME = #{seqName,jdbcType=VARCHAR}</select>
</mapper>

接口实现

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

工厂

工厂只做了两件事

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

@Component
public class MysqlSequenceFactory {private final Lock lock = new ReentrantLock();/**  */private Map<String,MysqlSequenceHolder> holderMap = new ConcurrentHashMap<>();@Autowiredprivate MysqlSequenceDAO msqlSequenceDAO;/** 单个sequence初始化乐观锁更新失败重试次数 */@Value("${seq.init.retry:5}")private int initRetryNum;/** 单个sequence更新序列区间乐观锁更新失败重试次数 */@Value("${seq.get.retry:20}")private int getRetryNum;@PostConstructprivate void init(){//初始化所有sequenceinitAll();}/*** <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();}}

单一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;/*** <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);}// 初始化该sequencetry {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++){//查询boMysqlSequenceBo 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++){//查询boMysqlSequenceBo 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服务 */@Autowiredprivate 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;}}