zoukankan      html  css  js  c++  java
  • 基于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=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;
    		
    	}
    
    }
    
    


    作者:coderzl
    出处:https://www.cnblogs.com/coderzl/
    本文采用知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议进行许可,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接。

  • 相关阅读:
    Android安全-代码安全1-ProGuard混淆处理
    快速提高Android开发效率的Web工具
    Android采用ListView实现数据列表显示2-使用SimpleAdapter进行数据绑定
    Java中Enum方法toString与ordinal方法
    视一:Web前端开发之HTML+CSS
    100-days: thirty-seven
    css 权重
    100-days: thirty-six
    100-days: thirty-five
    100-days: thirty-four
  • 原文地址:https://www.cnblogs.com/coderzl/p/7489886.html
Copyright © 2011-2022 走看看