时间:2021-02-18 11:20:30 | 栏目: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; } }
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_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; } }