使用Redis+Lua实现去重顺序任务队列，支持复杂对象去重
2023-07-11 本文已影响0人
liukeless
实现逻辑
实际业务中可能会出现触发一些任务执行，由多个触发点触发，但实际任务只需要再触发点触发之后执行一次就行的场景；多个触发点总共执行一次，减少重复执行次数。
- 通过lua脚本实现组合操作list和set两种数据结构实现保留任务的顺序性和去重的特性；
list中保存的数据结构为：ABS-7#_#{"businessKey":"ABS-7","remark":"1689143338792"}
set中的保存的数据为：ABS-7
通过set完成去重，通过list保证顺序性； lua脚本的方式保证原子性；
_#为特殊分割符号； 需要保证task对象json序列化之后不会出现这个分割符
import com.alibaba.fastjson.JSON;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import java.util.Arrays;
import java.util.function.Function;
@Data
public class RedisQueue<T> {
private static final String SEP = "#_#";
private Class<T> clazz;
public static final String TAKE = "local q = KEYS[1]\n" +
"local q_set = KEYS[1]..\"_set\"\n" +
"local v = redis.call('rpop', q)\n" +
"local items ={}\n" +
"local num = 0 \n" +
"if not v then\n" +
" return v\n" +
"end\n" +
"local temp = string.gsub(v, \"" + SEP + "\", \"#\")\n" +
"for substr in string.gmatch(temp, \"[^#]+\") do\n" +
" items[tostring(num)] = substr\n" +
" num = num + 1\n" +
"end\n" +
"if temp ~= \"\" and items[\"0\"] ~= \"\" and items[\"1\"] ~= \"\" then\n" +
" redis.call('srem', q_set, items[\"0\"] )\n" +
" return items[\"1\"]\n" +
"else \n" +
" redis.call('lpush', q, v)\n" +
" return \"\"\n" +
"end";
private static final String PUT = "local q = KEYS[1]\n" +
"local q_set = KEYS[1]..'_set'\n" +
"if redis.call('sadd', q_set, ARGV[1]) == 1 then\n" +
" redis.call('lpush', q, ARGV[2])\n" +
" return 1\n" +
"else\n" +
" return 0\n" +
"end";
public RedisQueue(String redisQueueName, Function<T, String> func, RedisTemplate<String, String> redisTemplate, Class<T> clazz) {
this.redisQueueName = redisQueueName;
this.func = func;
this.redisTemplate = redisTemplate;
this.clazz = clazz;
}
/**
* redis队列名称
*/
private String redisQueueName;
/**
* 根据此函数获取对象的唯一标识
*/
private Function<T, String> func;
private RedisTemplate redisTemplate;
public boolean addTask(T t) {
String taskKey = func.apply(t);
RedisScript<Long> addScript = RedisScript.of(
PUT, Long.class);
Long result = (Long) redisTemplate.execute(addScript, Arrays.asList(redisQueueName), taskKey, taskKey + SEP + JSON.toJSONString(t));
if (result == 1) {
return true;
} else {
return false;
}
}
public T takeTask() {
RedisScript<String> takeScript = RedisScript.of(
TAKE, String.class);
String result = (String) redisTemplate.execute(takeScript, Arrays.asList(redisQueueName));
if (StringUtils.isBlank(result)) {
return null;
}
return JSON.parseObject(result, clazz);
}
public T takeTask(long t) {
long end = System.currentTimeMillis() + t;
T task = null;
do {
task = takeTask();
if (null != task) {
return task;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
} while (System.currentTimeMillis() < end);
return task;
}
public T takeTaskBlock() {
T task = null;
do {
task = takeTask();
if (null != task) {
return task;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
} while (task == null);
return task;
}
}
import com.alibaba.fastjson.JSON;
import lombok.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.testng.annotations.Test;
import java.io.File;
import java.util.function.Function;
public class RedisQueueTest extends BaseTest {
@Autowired
private StringRedisTemplate redisTemplate;
@Data
public static class TaskVO {
private String businessKey;
private String remark;
}
@Test
public void testAddTask() throws InterruptedException {
RedisQueue<TaskVO> redisQueue = new RedisQueue<TaskVO>("zzz:opt:task:", new Function<TaskVO, String>() {
@Override
public String apply(TaskVO taskVO) {
return taskVO.getBusinessKey();
}
}, redisTemplate, TaskVO.class);
new Thread(new Runnable() {
@Override
public void run() {
while (true) {
try {
TaskVO taskVO = redisQueue.takeTask(5000);
if (taskVO == null) {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
continue;
}
System.out.println(Thread.currentThread().getName() + "消费：" + JSON.toJSONString(taskVO));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
} catch (Exception e) {
e.printStackTrace();
System.out.println(3123);
}
}
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
TaskVO taskVO = new TaskVO();
taskVO.setBusinessKey("ABS-" + i);
taskVO.setRemark(System.currentTimeMillis() + "");
if (i == 9) {
i = 0;
}
try {
Thread.sleep(100);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println(Thread.currentThread().getName() + "开始入队：" + JSON.toJSONString(taskVO));
boolean b = redisQueue.addTask(taskVO);
if (b) {
System.out.println(Thread.currentThread().getName() + "入队成功：" + JSON.toJSONString(taskVO));
}
}
}
}).start();
Thread.sleep(100000000);
}
}
hread-75开始入队：{"businessKey":"ABS-0","remark":"1689143330761"}
Thread-75入队成功：{"businessKey":"ABS-0","remark":"1689143330761"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143330911"}
Thread-74消费：{"businessKey":"ABS-0","remark":"1689143330761"}
Thread-75入队成功：{"businessKey":"ABS-1","remark":"1689143330911"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143338148"}
Thread-75入队成功：{"businessKey":"ABS-2","remark":"1689143338148"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143338269"}
Thread-75入队成功：{"businessKey":"ABS-3","remark":"1689143338269"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143338415"}
Thread-75入队成功：{"businessKey":"ABS-4","remark":"1689143338415"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143338543"}
Thread-75入队成功：{"businessKey":"ABS-5","remark":"1689143338543"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143338662"}
Thread-75入队成功：{"businessKey":"ABS-6","remark":"1689143338662"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143338792"}
Thread-75入队成功：{"businessKey":"ABS-7","remark":"1689143338792"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143338923"}
Thread-75入队成功：{"businessKey":"ABS-8","remark":"1689143338923"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143339045"}
Thread-74消费：{"businessKey":"ABS-1","remark":"1689143330911"}
Thread-75入队成功：{"businessKey":"ABS-9","remark":"1689143339045"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143339174"}
Thread-75入队成功：{"businessKey":"ABS-1","remark":"1689143339174"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143339301"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143339421"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143339553"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143339675"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143339798"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143339931"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143340055"}
Thread-74消费：{"businessKey":"ABS-2","remark":"1689143338148"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143340187"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143340305"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143340429"}
Thread-75入队成功：{"businessKey":"ABS-2","remark":"1689143340429"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143340559"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143340674"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143340806"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143340926"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143341079"}
Thread-74消费：{"businessKey":"ABS-3","remark":"1689143338269"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143341195"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143341326"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143341446"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143341572"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143341702"}
Thread-75入队成功：{"businessKey":"ABS-3","remark":"1689143341702"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143341835"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143341967"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143342096"}
Thread-74消费：{"businessKey":"ABS-4","remark":"1689143338415"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143342217"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143342341"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143342473"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143342605"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143342724"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143342845"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143342971"}
Thread-75入队成功：{"businessKey":"ABS-4","remark":"1689143342971"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143343099"}
Thread-74消费：{"businessKey":"ABS-5","remark":"1689143338543"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143343218"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143343352"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143343481"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143343613"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143343728"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143343852"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143343976"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143344107"}
Thread-74消费：{"businessKey":"ABS-6","remark":"1689143338662"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143344222"}
Thread-75入队成功：{"businessKey":"ABS-5","remark":"1689143344222"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143344345"}
Thread-75入队成功：{"businessKey":"ABS-6","remark":"1689143344345"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143344473"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143344593"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143344734"}
Thread-75开始入队：{"businessKey":"ABS-1","remark":"1689143344861"}
Thread-75开始入队：{"businessKey":"ABS-2","remark":"1689143344986"}
Thread-75开始入队：{"businessKey":"ABS-3","remark":"1689143345122"}
Thread-74消费：{"businessKey":"ABS-7","remark":"1689143338792"}
Thread-75开始入队：{"businessKey":"ABS-4","remark":"1689143345252"}
Thread-75开始入队：{"businessKey":"ABS-5","remark":"1689143345391"}
Thread-75开始入队：{"businessKey":"ABS-6","remark":"1689143345523"}
Thread-75开始入队：{"businessKey":"ABS-7","remark":"1689143345639"}
Thread-75入队成功：{"businessKey":"ABS-7","remark":"1689143345639"}
Thread-75开始入队：{"businessKey":"ABS-8","remark":"1689143345765"}
Thread-75开始入队：{"businessKey":"ABS-9","remark":"1689143345895"}
