团队更换新框架。新的业务全部使用新的框架，甚至是新的数据库--Mysql。
这边之前一直是使用oracle，各种订单号、流水号、批次号啥的，都是直接使用oracle的sequence提供的数字序列号。现在数据库更换成Mysql了，显然以前的老方法不能适用了。
需要新写一个：
•分布式场景使用
•满足一定的并发要求
找了一些相关的资料，发现mysql这方面的实现，原理都是一条数据库记录，不断update它的值。然后大部分的实现方案，都用到了函数。
贴一下网上的代码：
基于mysql函数实现
表结构
|
1
2
3
4
5
6
7
8
9
|
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=InnoDBDEFAULT CHARACTER SET=utf8 COLLATE=utf8_general_ciROW_FORMAT=COMPACT; |
获取下一个值
|
1
2
3
4
5
6
7
8
9
10
11
|
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
|
1
2
3
4
5
6
7
8
9
10
|
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接口
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
/** * <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区间
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
/** * <p></p> * * @author coderzl * @Title SequenceRange * @Description 序列区间，用于缓存序列 * @date 2017/6/6 22:58 */ @Datapublic 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
对应数据库记录
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
|
@Datapublic 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
增删改查，其实就用到了改和查
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
<?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> |
接口实现
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
@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的下一个值
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
@Componentpublic 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() 获取下一个值
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
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拼接方法
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
@Servicepublic 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实现方法就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持服务器之家。
原文链接：http://www.cnblogs.com/coderzl/p/7489886.html
