基于Mysql的Sequence实现方法

2023年 4月 28日 45.1k 0

团队更换新框架。新的业务全部使用新的框架,甚至是新的数据库--Mysql。 这边之前一直是使用oracle,各种订单号、流水号、批次号啥的,都是直接使用oracle的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=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 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 parameterType="java.lang.String" >
delete from t_pub_sequence
where SEQ_NAME = #{seqName,jdbcType=VARCHAR}
</delete>
<insert 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 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 resultMap="BaseResultMap" >
select SEQ_NAME, SEQ_VALUE, MIN_VALUE, MAX_VALUE, STEP
from t_pub_sequence
</select>

<select 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实现方法就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持每日运维。

相关文章

Oracle如何使用授予和撤销权限的语法和示例
Awesome Project: 探索 MatrixOrigin 云原生分布式数据库
下载丨66页PDF,云和恩墨技术通讯(2024年7月刊)
社区版oceanbase安装
Oracle 导出CSV工具-sqluldr2
ETL数据集成丨快速将MySQL数据迁移至Doris数据库

发布评论