背景
在大多数web系统中，不同用户之间的操作并行，但单个用户的操作应当是串行执行的，由于现在的分布式系统大多服务都有多实例，如何保证用户的操作串行化
方案
首先想到的是利用分布式锁，分布式锁有几种不同的实现方式，列举自己用到的两种方式：
1.利用数据库的唯一索引。这种方式比较简单暴力，我在刚接触部门已投产的系统时，系统就存在重复提交申请的问题，利用此方法便可以不修改系统代码的情况下解决问题。
2.利用redis内存数据库。在redis上加入被视为锁的key，根据具体的实现方式，判断key的状态，给相关代码加锁。
题外话：其实系统不管要完成何种业务，不过都是对数据的操作加工，所以数据才是关键，在数据库做限制是很有必要的
实施理论
为每个客户创建一把锁，初始化状态下，锁为未被持有状态，当系统拿到锁后同时将锁置为持有状态
具体实施
在redis中为每个用户创建两个队列，一个为持有队列，一个为未持有队列。初始化时用户存在一个未持有队列，队列中存在一个元素，当用户来获取锁时，阻塞式的将未持有队列中元素弹出并弹入持有队列，反之，当用户释放锁时，便将元素（可以非阻塞式）从持有队列弹出并弹入未持有队列，因为队列中仅存在一个元素，所以在元素弹出后，队列便会消亡，因此需要维护队列长度为1
代码
首先是造轮子
@Component
public class RedisCache {
/* spring redis 模版 */
@Autowired
private RedisTemplate<String, String> redisTemplate;
/* redis key值 操作前缀 */
private String prefixname = "mf_";
/**
* List-阻塞式移除source队列尾元素，放入target队列头部
*
* @param source source队列key
* @param target target队列key
* @param expireTime 阻塞时间
* @return :String
* @author : Sugar
* @since : 2018年11月14日
*/
public String bLPopRPush(String source, String target, int expireTime) {
final byte[] skey = (prefixname + source).getBytes();
final byte[] tkey = (prefixname + target).getBytes();
return redisTemplate.execute((RedisCallback<String>) connection -> {
byte[] bytes = connection.bRPopLPush(expireTime, skey, tkey);
if (null == bytes)
return null;
else
return new String(bytes);
});
}
/**
* List-非阻塞移除source队列尾元素，放入target队列头部,并刷新target存活时间
*
* @param source source队列key
* @param target target队列key
* @param tTL target队列存活时间
* @return :String
* @author : Sugar
* @since : 2018年11月14日
*/
public String rPopLPushEx(String source, String target, long tTL) {
final byte[] skey = (prefixname + source).getBytes();
final byte[] tkey = (prefixname + target).getBytes();
return redisTemplate.execute((RedisCallback<String>) connection -> {
connection.multi();
connection.rPopLPush(skey, tkey);
connection.expire(tkey, tTL);
List<Object> result = connection.exec();
if (null == result)
return null;
if (null == result.get(0))
return null;
return new String((byte[]) result.get(0));
});
}
/**
* 新增key（不存在时成功），并设置存活时间
*
* @param key key
* @param value value
* @param tTL 存活时间
* @return :List<Object>
* @author : Sugar
* @since : 2018年11月14日
*/
public Boolean setNxEx(String key, String value, long tTL) {
final byte[] keyBytes = (prefixname + key).getBytes();
final byte[] valueBytes = value.getBytes();
return redisTemplate.execute((RedisCallback<Boolean>) connection -> {
connection.multi();
connection.setNX(keyBytes, valueBytes);
connection.expire(keyBytes, tTL);
List<Object> result = connection.exec();
if (null == result)
return null;
return (Boolean) result.get(0);
});
}
/**
* 判断任何一个key是否存在
*
* @param keys keys
* @return :List<Object>
* @author : Sugar
* @since : 2018年11月14日
*/
public Boolean exists(String... keys) {
final byte[][] keyBytes = new byte[keys.length][];
for (int i = 0; i < keys.length; i++) {
keyBytes[i] = (prefixname + keys[i]).getBytes();
}
return redisTemplate.execute((RedisCallback<Boolean>) connection -> {
connection.multi();
for (int i = 0; i < keyBytes.length; i++) {
connection.exists(keyBytes[i]);
}
List<Object> result = connection.exec();
if (result == null)
return null;
return result.parallelStream().anyMatch(it -> (Boolean) it);
});
}
/**
* 获取匹配到的keys
*
* @param keys
* @return :String
* @author : Sugar
* @since : 2018年11月23日
*/
public List<String> keys(String keys) {
final byte[] bkey = (prefixname + keys).getBytes();
return redisTemplate.execute((RedisCallback<List<String>>) connection ->
connection.keys(bkey).parallelStream().map(it -> new String(it)).collect(Collectors.toList()));
}
/**
* List-首位置插入一个元素,并设置超时时间
*
* @param key
* @param values
* @param expireTime
* @return :Long
* @author : Sugar
* @since : 2018年11月23日
*/
public Long lPushEx(String key, String values, long expireTime) {
final byte[] bkey = (prefixname + key).getBytes();
final byte[] bvalue = values.getBytes();
Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {
connection.multi();
connection.lPush(bkey, bvalue);
connection.expire(bkey, expireTime);
List<Object> execResult = connection.exec();
if (null == execResult)
return null;
return (Long) execResult.get(0);
});
return result;
}
}
注意，redis并不提供事务回滚机制，按官方（本人英文比较烂，所有看的不是太明白）介绍，如果redis命令语法能加入队列且可执行（watch的key没有变化），那么事务就一定能执行成功，不存在像关系型数据库一样，即使语法正常，也有可能执行失败。
接着是串行化的工具类
public class SerialImpl implements Serial {
@Autowired
private RedisCache redisCache;
final private String runningKey = "serial:running:";
final private Long tTl = 60 * 60L;
final private String readyKey = "serial:ready:";
@Override
public void run(String... strings) throws Exception {
//超时时间取决中服务启动所需时间
if (!redisCache.setNxEx("init:sysStart", "1", 5 * 60))
return;
List<String> keyList = redisCache.keys(runningKey + "*");
keyList.parallelStream().forEach(it -> {
redisCache.rPopLPushEx(it.substring(it.indexOf("_") + 1), readyKey + it.substring(it.indexOf("pro")), tTl);
if (redisCache.exists(it))
redisCache.deleteCache(it);
});
}
@Override
public boolean init(String productNo, String phone) {
String suffix = productNo + ":" + phone;
if (!redisCache.setNxEx("init:" + suffix, "1", 5L))
return false;
if (redisCache.exists(runningKey + suffix, readyKey + suffix)) {
String result = redisCache.bLPopRPush(readyKey + suffix, runningKey + suffix, 30);
if (null == result) {
redisCache.deleteCache("init:" + suffix);
return false;
}
redisCache.deleteCache("init:" + suffix);
return true;
}
Long pushR = redisCache.lPush(runningKey + suffix, "1");
if (null == pushR) {
redisCache.deleteCache("init:" + suffix);
return false;
}
if (1 != pushR) {
redisCache.deleteCache(runningKey + suffix);
redisCache.deleteCache("init:" + suffix);
return false;
}
redisCache.deleteCache("init:" + suffix);
return true;
}
@Override
public boolean entrance(String productNo, String phone) {
String suffix = productNo + ":" + phone;
String result = redisCache.bLPopRPush(readyKey + suffix, runningKey + suffix, 30);
if (null == result)
return false;
return true;
}
@Override
public boolean exit(String productNo, String phone) {
String suffix = productNo + ":" + phone;
String result = redisCache.rPopLPushEx(runningKey + suffix, readyKey + suffix, tTl);
if (null == result)
return false;
return true;
}
}
然后是最后的实施，本人在AOP中切入所有业务Controller,由于本类涉及其他业务代码，故只截取相关代码片段：
boolean serialResult;
serialResult = serial.entrance(custModel.getProductNo(), custModel.getPhoneNo());
if (!serialResult) {
resMap.put("retFdesc", "请稍后重试，或返回首页重试");
resMap.put("retFCode", "19");
resp.setData(resMap);
} else {
try {
resp = ((ResultResponse) point.proceed());
} finally {
serial.exit(custModel.getProductNo(), custModel.getPhoneNo());
}
}
由于初始化是所涉及的接口只有一个，所以单独在业务代码中执行serial.init方法
总结
此次主要是利用redis来实现分布式系统的用户的串行化，如有不对或不足之处，欢迎指正。
网友评论