点击下载
本文文档

当前位置：首页 - 科技 - 知识百科 - 正文

基于Mysql的Sequence实现方法

来源：动视网责编：小采时间：2020-11-09 20:30:43

基于Mysql的Sequence实现方法

基于Mysql的Sequence实现方法:团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--Mysql。这边之前一直是使用oracle，各种订单号、流水号、批次号啥的，都是直接使用oracle的sequence提供的数字序列号。现在数据库更换成Mysql了，显然以前的老方法不能适用了。需要新写一个

推荐度：

点击下载本文 文档为doc格式

导读基于Mysql的Sequence实现方法:团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--Mysql。这边之前一直是使用oracle，各种订单号、流水号、批次号啥的，都是直接使用oracle的sequence提供的数字序列号。现在数据库更换成Mysql了，显然以前的老方法不能适用了。需要新写一个

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

/**
 * <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;
 }

}

对应数据库记录

@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_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的下一个值

@Component
public 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();
 }

}

单一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);
 }
 // 初始化该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拼接方法

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

}

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