使用Redis+Lua实现去重顺序任务队列，支持复杂对象去重

实现逻辑

实际业务中可能会出现触发一些任务执行，由多个触发点触发，但实际任务只需要再触发点触发之后执行一次就行的场景；多个触发点总共执行一次，减少重复执行次数。

通过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"}

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 222,000评论 6赞 515
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 94,745评论 3赞 399
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 168,561评论 0赞 360
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 59,782评论 1赞 298
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 68,798评论 6赞 397
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 52,394评论 1赞 310
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,952评论 3赞 421
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,852评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 46,409评论 1赞 318
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 38,483评论 3赞 341
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,615评论 1赞 352
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 36,303评论 5赞 350
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,979评论 3赞 334
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,470评论 0赞 24
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,571评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 49,041评论 3赞 377
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,630评论 2赞 359

使用Redis+Lua实现去重顺序任务队列，支持复杂对象去重

实现逻辑

_#为特殊分割符号； 需要保证task对象json序列化之后不会出现这个分割符

推荐阅读更多精彩内容

_#为特殊分割符号；需要保证task对象json序列化之后不会出现这个分割符