Redis的集合操作
实话说，Redis提供的集合操作是我选择它成为内存数据库的一个主要理由，它弥补了传统关系型数据库在这方面带来的复杂度，使得只需要简单的一个命令就可以完成一个复杂SQL任务，并且交、并、差操作在实际的业务场景中应用非常广泛，比如快速检索出具备一系列标签属性的一个集合，本篇文章将主要介绍对于求交集操作结果缓存的设计方案。
Redis命令
对于set类型，提供了sinter、sinterstore进行交集操作，对于sortedset，提供了zinter、zinterstore进行交集操作。命令十分便捷，对于不保存结果的方法sinter和zinter只需要输入待求交集的集合的key的数组就可以得到求交集后的结果集合，对于保存结果的方法，你可以将求交集后的结果集合保存到你指定key的一个集合中。
设计方案
目标
计算给定集合群的交集集合，并对计算过程中所产生的中间结果集合进行缓存，从而达到下次计算给定集合群的一些子群集时，先查询是否存在交集key，如果存在直接读取，避免重复计算。
原始方法（Only）
以下我们以Redis Java客户端库Jedis进行举例，Jedis提供了与Redis命令的一致接口方法，实现了交集操作，如下代码：
public Set<String> inter(String[] keys, String keycached){
int size = keys.length;
Jedis jedis = new Jedis("127.0.0.1");
if(size < 2){
try{
return jedis.smembers(keys[0]);
} finally {
jedis.close();
}
}
try{
jedis.sinterstore(keycached, keys);
return jedis.smembers(keycached);
} finally {
jedis.close();
}
}
原始方法的问题在于只对最终的交集key进行了缓存，简洁方便，但每次变更给定集合群时，都需要重新在此计算。
原始方法上的改造方案（All）
在原始方法上进行改造，我们可以在计算过程中依次增加计算集合群的集合数量，比如给定的集合群key{A，B，C，D}，我们先计算A、B，保存一个{A，B}的交集结果，再依次计算A、B、C和A、B、C、D并对结果进行保存。
显然，这是个糟糕的方案，但确实完成了我们设定的目标，参考代码如下：
private Set<String> interByAll(String... keys){
Jedis jedis = new Jedis("127.0.0.1");
Set<String> value = null;
int interNum = 2;
for(int i = 0; i < (keys.length - 1); i++){
String keystored = "";
String[] keyintered = new String[interNum];
for(int j = 0; j < interNum; j++){
keystored += (keys[j] + "&");
keyintered[j] = keys[j];
}
if(jedis.sinterstore(keystored, keyintered) == 0){
jedis.sadd(keystored, "nocache");
}
if(interNum == keys.length){
value = jedis.smembers(keystored);
}
interNum++;
}
jedis.close();
return value;
}
递归方案（Recursive）
根据上面糟糕的设计方案，你应该改进实现一种递归方案，递归方案的好处是你每次只求一对集合的交集，逐步完成对整个给定集合群的交集计算，计算过程如下图所示：
private boolean isEnd = false;
private Set<String> value = null;
private Set<String> getKeysWithInner(String[] keys, String srckey, Jedis jedis, int i){ String key = null; if(srckey == null){//表示为第一次求交集
srckey = keys[i++];
key = keys[i++]; } else {
key = keys[i++];
} String keystored = srckey + "&" + key;//生成缓存key if(jedis.sinterstore(keystored, srckey, key) == 0){
jedis.sadd(keystored, "nocache");
} if(i == keys.length){ //当与最后一个key求交集后，返回结果，并跳出递归调用
value = jedis.smembers(keystored);
isEnd = true;
} while(!isEnd){//递归调用，一对key集合求交集
getKeysWithInner(keys, keystored, jedis, i);
} return value; } public Set<String> interByRecursive(String... keys){
int size = keys.length;
Jedis jedis = new Jedis("127.0.0.1");
if(size < 2){
try{
return jedis.smembers(keys[0]);
} finally {
jedis.close();
}
} try{ return getKeysWithInner(keys, null, jedis, 0);
} finally {
isEnd = false;
jedis.close();
} }
该方案的优势不仅仅是对计算过程进行了缓存，而且，每次都只是完成一对集合的计算，计算量显著降低。
Fork-Join方案（Fork-Join）
一写递归方案，你就可以直接想到使用Fork-Join框架进行改造，并使其并行化，计算过程如下图所示：
InterTask类：
public class InterTask extends RecursiveTask<String>{
private String[] keys = null;
private static final int THRESHOLD = 3;
public InterTask(String[] keys){
this.keys = keys;
}
private static String genKeyinnered(String... keys){
StringBuilder sb = new StringBuilder();
for(String key : keys){
sb.append(key);
sb.append("&");
}
return sb.toString().substring(0, sb.length() - 1);
}
@Override
protected String compute() {
int size = keys.length;
if(size < THRESHOLD){//当keys数组中元素数为2个或1个时，计算交集，并退出递归
if(size < 2){
return keys[0];
}
Jedis jedis = new Jedis("127.0.0.1");
try{
jedis.sinterstore(genKeyinnered(keys), keys[0], keys[1]);
} finally {
jedis.close();
}
return genKeyinnered(keys);
} else {
//取keys数组的中值进行分治算法
String[] leftkeys = new String[size / 2];
String[] rightkeys = new String[size - (size / 2)];
//按中值拆分keys数组
for(int i = 0; i < size; i++){
if(i < leftkeys.length){
leftkeys[i] = keys[i];
} else {
rightkeys[i - leftkeys.length] = keys[i];
}
}
InterTask lefttask = new InterTask(leftkeys);
InterTask righttask = new InterTask(rightkeys);
lefttask.fork();
righttask.fork();
//取得从递归中返回的一对存储交集结果的key
String left = lefttask.join();
String right = righttask.join();
Jedis jedis = new Jedis("127.0.0.1");
try{
jedis.sinterstore(left + "&" + right, left, right);
} finally {
jedis.close();
}
return left + "&" + right;
}
}
}
这里运用了最基础的分治算法思想，逐步将一个大的给定集合拆解为若干个成对的集合进行交集计算。
调用方法：
public Set<String> interByForkJoin(String... keys){
Set<String> value = null;
Jedis jedis = new Jedis("127.0.0.1");
InterTask task = new InterTask(keys);
ForkJoinPool forkJoinPool = new ForkJoinPool();
Future<String> result = forkJoinPool.submit(task);
try {
String key = result.get();
value = jedis.smembers(key);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
} finally {
jedis.close();
}
return value;
}
测试
测试数据准备
我们这里准备100个集合，每个给定集合包含1000000个元素，参考如下代码：
Jedis jedis = new Jedis("127.0.0.1");
jedis.flushAll();
String token = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ";
Random rand = new Random();
String[] keys = new String[100];
for(int i = 0; i < 100; i++){
keys[i] = "ct" + i;
String[] values = new String[1000000];
for(int j = 0; j < 1000000; j++){
StringBuilder sb = new StringBuilder();
for(int k = 0;k < 2; k++){
sb.append(token.charAt(rand.nextInt(62)));
}
values[j] = sb.toString();
}
jedis.sadd(keys[i], values);
}
jedis.close();
测试结果
我们分别对25、50、75、100的给定集合进行交集计算，测试结果如下：
我们可以清楚的看到，All方案是多么的糟糕，剔除All方案的结果：
总体来说Only方案和Recursive方案不分伯仲，但在相对较小的给定合集计算场景下，Recursive存在优势，而且其进行了计算过程结果的缓存。
对于Fork-Join方案表示比较遗憾，当然这里可以采用另外一种更优的分解算法完成并行过程，但是就Redis本身作为通过单线程epoll模型实现的异步IO来说，可能客户端的并行计算在服务端仍然被串行化处理，另外，分治算法拆分数组的时间损耗也不能忽略。