美文网首页@IT·互联网
使用Redis+Lua实现去重顺序任务队列，支持复杂对象去重

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

作者: liukeless | 来源:发表于2023-07-11 14:39 被阅读0次

实现逻辑

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

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

相关文章

  • Array集结号

    实现数组去重的几种方法 数组去重一 数组去重二 利用数组indexof+push实现数组去重 数组去重三 利用对象...

  • 数组去重

    使用set进行简单去重 使用reduce进行复杂数组去重

  • js reduce去重用法

    reduce不仅仅可以数据累加，还可以实现去重效果。 重复次数计算 数组去重 数组对象去重，转为数组 对象去重

  • 18,java HashSet 去重复的4种方法

    LinkedHashSet去重去重后保持原有顺序（重复数据只保留一条） HashSet去重方法一去重后顺序打乱（重...

  • HZ数组对象是否重复

    使用背景：批量表单验证 例如：姓名，手机号不可重复 实现思路：根据去重后的属性对象个数对比去重前的 实现步骤：1....

  • iOS面试题整理

    1、数组去重的方法： （1）原来集合操作可以通过valueForKeyPath来实现的，去重可以一行代码实现（顺序...

  • SpringDataMongoDB去重+排序+分页

    一、findDistinct去重 使用mongoTemplate.findDistinct去重，不支持排序，即使你...

  • Python 列表去重

    1、去重不保持原来的顺序: 2、去重保持原顺序(reduce): 3、list元素是字典，去重

  • Java集合去重的几种方法

    一、基础数据类型集合去重： 例：String 方法实现： 二、对象根据指定属性去重

  • List集合去重，排序，对象，equals，

    List元素为基本数据类型，去重 // 使用lambad表达式去重 List元素为对象事时去重，需要重写对象的eq...

网友评论

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

    本文链接：https://www.haomeiwen.com/subject/rfzyudtx.html

    延伸阅读

    深度阅读

    • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

    栏目导航

    热点阅读

    @IT·互联网

    关于我们|服务条款|联系我们|使用Redis+Lua实现去重顺序任务队列，支持复杂对象去重|投稿指南|网站地图|RSS订阅|排版工具|手机版

    提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

    Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

    备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

    本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

    所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！