【问题标题】：Poping the value from Redis List in multi-threaded environment returns the duplicate value to few threads在多线程环境中从 Redis 列表中弹出值会将重复值返回给少数线程
【发布时间】：2018-08-15 17:34:50
【问题描述】：

数据库中有一些记录在我通过一些复杂的数据库查询获得的 cId 和 mId 之间具有关联。用例是获取cId&mId组合对应的dataId的完整列表，并推送到redis。我们从输入 csv 文件到我的批处理作业中获得 cId 和 mId 组合。文件中会有多条记录对应相同的组合批处理作业配置有 10 个并行线程和一个读取记录的线程。我们想要的是当一个线程从文件中读取特定组合时，我们从数据库中获取所有记录并将它们上传到 redis 以解决 2 个问题： 1. 数据库命中 2. 并发问题，即两个线程不应该根据 cId & mId 组合从 db 中获取相同的记录

private static Long DEFAULT_REDIS_OBJECTID = 0L;


        public DataObject getDataObject(C cId, M mId) {
                String redisListKey = new StringBuilder().append(LIST-).append(cId)
                        .append("-").append(mId).toString();
    if (BooleanUtils.isFalse(redisTemplate.hasKey(redisListKey))) {
                pushRedisData(cId, mId, redisListKey);
            }
        }
    Long dataId = redisTemplate.opsForList().leftPop(redisListKey);
            if (Objects.isNull(dataId ) || 0L.equals(dataId )) {
//    Again creating the key in order to make sure another thread request for this should not shouldn't go in the db as we know we dont have data in db for this combination         
redisTemplate.opsForList().leftPush(redisListKey, 0L);
                //create a new dataObject and return it
            } else {
                //Get the dataObject based on dataId and return it
            }


    public synchronized void pushRedisData(Long cId, Long mId, String redisListKey) {
            if (BooleanUtils.isFalse(redisTemplate.hasKey(redisListKey))) {
                List<Long> dataToPush = dataService.getDataIdListFromCIdAndMIdCombination(cId,
                        mId);
                if (CollectionUtils.isNotEmpty(dataToPush )) {
                    redisTemplate.opsForList().leftPushAll(redisListKey, dataToPush );
                    redisTemplate.expire(redisListKey, 5, TimeUnit.HOURS);
                } else {
                    if (redisTemplate.opsForList().size(redisListKey) == 0) {
                        redisTemplate.opsForList().leftPush(redisListKey, DEFAULT_REDIS_ITEMID);
                    }

                }

我已经做了同步的方法将记录推送到redis，这样只有一个线程可以将数据发布到redis对应的key，其他线程只是从redis中弹出数据。如果在数据库中没有找到任何 cId 和 mId 组合的记录，那么我将在 redis 上创建默认值为 0 的键，这样具有这种组合的线程不应该进行数据库调用。

问题：当我在文件中执行包含 1000 条记录的批处理作业并且配置了 10 个线程来处理这些记录时，我看到 3-5 个线程获得了已分配给其他线程的重复 dataId线程和对象正在处理导致具有重复 dataId 的线程的 Stalestate 异常。我还发现这个问题是在前几条记录的工作开始阶段遇到的。

【问题讨论】：

标签： redis spring-batch

【解决方案1】：

TL;DR

由于同步泄漏，您遇到了竞争条件。

说明

您尝试同步的方式是泄漏的，它允许多个线程执行相同的工作。这是你已经注意到的。没有原子块可以保护两个并发线程不执行相同的语句——这里不说锁。

如果您使用相同的对象实例，引入synchronized 方法可能会在本地解决问题，但这不是解决方案。它会阻塞其他并发线程，阻止这些线程继续进行。

您为什么不通过例如使用 Redis 的属性进行同步？一套？

您可以使用 Redis 集来确保只有单个线程/进程能够通过将特定元素添加到 Redis 集中来处理该元素。 Redis 将响应添加该元素是否成功。此信息将帮助您通过检查响应来解决竞争。如果可以将元素添加到集合中，则当前线程是第一个命中您的dataId 的线程，并且该线程可能会继续进行昂贵的工作（数据库提取，...）。

当向同步集中添加元素失败时，您就知道某个其他进程已经在做昂贵的工作，您可以继续从缓存中查找数据。您需要注意，虽然另一个线程可能已经赢得了非阻塞同步，但另一个第一个线程不一定完成填充缓存。然后你可以：

轮询缓存直到值存在。
暂停（您无法确定其他进程是否处于活动状态），然后重复工作以继续执行您的实际任务。

想想更糟糕的情况：等待缓存值并阻止导入进度或多次查询昂贵的数据源。您可以针对这两种情况进行优化，但您需要自行决定。

【讨论】：