实现逻辑

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

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