当前位置：首页 > article >正文

Redis 篇- 实战项目中使用 Redis 实现经典功能（异步秒杀商品、点赞功能、共同关注的好友、投喂功能）

article 2024/9/21 1:46:40

🔥博客主页：【小扳_-CSDN博客】
❤感谢大家点赞👍收藏⭐评论✍

文章目录

1.0 使用 Redis 实现异步秒杀

1.1 基于 Lua 脚本判断是否符合条件：库存是否充足、一人一单

1.2 基于 Redis 中的 Stream 实现消息队列

1.3 使用 Java 操作 Redis 实现消息队列

2.0 使用 Redis 实现点赞功能

2.1 使用 Redis 实现点赞排行榜功能

3.0 使用 Redis 实现好友之间的共同关注（共同好友）

4.0 使用 Redis 实现投喂（发布文章）

4.1 使用 Redis 实现收件箱（收邮件）

4.1.1 实现滚动分页查询（查询朋友圈信息）

1.0 使用 Redis 实现异步秒杀

异步秒杀，顾名思义使用不同线程执行不同的任务，在 Redis 实现异步秒杀时，主线程执行操作 Redis 来判断是否符合条件下单，对于操作数据库的任务则交给线程池中的线程来执行。

而且将下单的 id 等信息放入到消息队列中，再由执行数据库操作的时候再来获取 id 消息，这就不用一直等待数据库完成之后，才能迎接下一个线程进行下单操作了。在下单的时候就不用再考虑是否安全问题了，从而实现一人一单的时候就不需要进行上锁处理了，还有解决超卖问题也不需要乐观锁了。

这样数据库就可以在适合的时间段根据消息队列中的消息，来将数据存放到数据库中，而从减轻了数据库的压力，效率还很高。

为了保证在秒杀过程中不会出现线程安全情况，使用 Lua 脚本操作 Redis 命令。

1.1 基于 Lua 脚本判断是否符合条件：库存是否充足、一人一单

实现思路：

Lua 脚本如下：

local shopId = ARGV[1]
local userId = ARGV[2]

-- 商品key
local shopKey = "shop:"+shopId

-- 用户Key
local orderKey = "order:"+shopId

-- 判断库存是否充足
if(tonumber(redis.call('get',shopKey)) <= 0) then
    -- 库存不足，返回1
    return 1

end

-- 判断是否已经下单过了
if(redis.call('sismember', orderKey,userId) == 1) then
    -- 已经下过单了，返回2
    return 2
end
-- 扣库存
redis.call('incrby',shopKey,-1)
-- 下单保存用户
redis.call('sadd', orderKey,userId)
return 0

1.2 基于 Redis 中的 Stream 实现消息队列

使用 Redis 中的 Stream 数据结构，来实现消息队列。

常见的 Redis 命令：

1）发送消息到消息队列中：

XADD key * field string [field string]

将指定的键值对发送到具体的队列中

具体代码如下：

命令执行结果：

2）读取消息队列中的消息：

创建消费者组：

XGROUP CREATE key groupname id|$ MKSTREAM

ID 为 0 的话，从队列中从 0 开始重新读取任务，而 ID 为 $ 的话，从队列最后一个开始读取任务，抛弃之前队列中的任务。

具体代码如下：

消费者组读取消息队列：

XREADGROUP GROUP group consumer COUNT count BLOCK milliseconds STREAMS key ID

消费组从队列中读取消息。若 ID 为 ">" ，则从消息队列最后一个开始，也就是最新的消息开始读取消息；若 ID 为 "0"，则从 pending 中读取消息，也就是读取消息了，但是没有进行确认的消息。

确认消息：

XACK key group id consumer

从消息队列读取出来的消息之后，进行确认，则该消息就不会进入到 pending 状态，否则该消息再没有进行确认下，会来到 pending 状态。

1.3 使用 Java 操作 Redis 实现消息队列

1）发布消息：

可以用 Lua 脚本，在确认完可以运行下单的用户，进行下单，也就是将相关信息放到消息队列中，交由其他线程池来完成读取消息后进行操作数据库。

代码实现：

local shopId = ARGV[1]
local userId = ARGV[2]

-- 商品key
local shopKey = "shop:"+shopId

-- 用户Key
local orderKey = "order:"+shopId
local streamKey = "stream"

-- 判断库存是否充足
if(tonumber(redis.call('get',shopKey)) <= 0) then
    -- 库存不足，返回1
    return 1

end

-- 判断是否已经下单过了
if(redis.call('sismember', orderKey,userId) == 1) then
    -- 已经下过单了，返回2
    return 2
end
-- 扣库存
redis.call('incrby',shopKey,-1)
-- 下单保存用户
redis.call('sadd', orderKey,userId)
-- 将数据保存在消息队列中
redis.call('xadd',streamKey,"*","shopId",shopId,"userId",userId)

return 0

在之前判断是否符合下单的 Lua 脚本中加上往队列中添加数据。

2）从消息队列中获取数据

持续的从消息队列中尝试获取数据。当然，该方法在实战中交给线程池处理。

import cn.hutool.core.bean.BeanUtil;
import com.project.volunteermanagementproject.pojo.StreamObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
@Component
@Slf4j
public class StreamUtil {

    private final StringRedisTemplate stringRedisTemplate;
    public StreamUtil(StringRedisTemplate stringRedisTemplate){
        this.stringRedisTemplate = stringRedisTemplate;
    }


    //实现发送消息
    public RecordId pubStream(StreamObject streamObject){
        Map<String, Object> map = BeanUtil.beanToMap(streamObject);
        return stringRedisTemplate.opsForStream().add("s1", map);
    }

    //实现从消息队列中获取消息
    public void getStream(){
        while (true){
            try {
                List<MapRecord<String, Object, Object>> read = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create("s1", ReadOffset.lastConsumed())
                );

                if (read == null || read.isEmpty()){
                    //如果获取失败，说明没有消息，继续下一次循环
                    continue;
                }
                //解析消息中的消息
                MapRecord<String, Object, Object> entries = read.get(0);
                Map<Object, Object> value = entries.getValue();
                StreamObject streamObject = BeanUtil.fillBeanWithMap(value, new StreamObject(), true);
                //这就拿到了消息队列中的数据了,就可以去使用该对象了
                log.info("成功从消息队列中获取到数据: "+streamObject);
                //这就需要确认消息队列
                stringRedisTemplate.opsForStream().acknowledge("s1", "g1", entries.getId());
            } catch (Exception e) {
                //如果在获取消息过程中出现异常，则需要再次执行该消息任务
                while (true){
                    try {
                        List<MapRecord<String, Object, Object>> read = stringRedisTemplate.opsForStream().read(
                                Consumer.from("g1", "c1"),
                                StreamReadOptions.empty().count(1),
                                StreamOffset.create("s1", ReadOffset.from("0"))
                        );
                        if (read == null || read.size() == 0){
                            break;
                        }
                        MapRecord<String, Object, Object> entries = read.get(0);
                        Map<Object, Object> value = entries.getValue();
                        StreamObject streamObject = BeanUtil.fillBeanWithMap(value, new StreamObject(), true);
                        //重新拿到未确认的数据
                        log.info("再次成功拿到数据: "+streamObject);
                        //再次进行消息确认
                        Long acknowledge = stringRedisTemplate.opsForStream().acknowledge("s1", "g1", entries.getId());
                    } catch (Exception ex) {
                        throw new RuntimeException(ex);
                    }

                }
            }

        }
    }
}

3）手动模拟往消息队列中添加消息进行下单，通过读取消息队列中的方法进行监听：

    @Test
    void text2(){
        //持续接收消息
        StreamUtil streamUtil = new StreamUtil(stringRedisTemplate);
        streamUtil.getStream();
        
    }

发送的消息：

接收的消息：

2.0 使用 Redis 实现点赞功能

实现点赞思路：每一次点赞，都往数据库中修改一次数据+1，但是在日常的社交软件中，都不会有无限点赞的效果，比如说朋友圈，第一次点赞成功，再点一次则取消点赞。因此，需要解决的是一个人只能点赞一次，或者取消一次点赞。

如何判断当前用户是否已经点赞呢？

当然方法有非常多种，这里介绍的是使用 Redis 解决该方法。

具体思路：使用 Redis 中的 set 集合数据结构，利用 set 的不可重复特性。在点击点赞之前，通过判断集合中之前是否存在该用户 id ，如果存在，则再点击一下为取消点赞；如果之前不存在，则点击一下为点赞成功，之后将该用户 id 放入集合中，方便下一次判断是否点赞。

代码实现：

import cn.hutool.core.util.BooleanUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

@Component
public class ThumbsUpUtil {

    @Autowired
    StringRedisTemplate stringRedisTemplate;
    //当前用户id
    private final Integer currentUserId = 1;
    public Boolean thumbs(Integer userId){
        String key = "userId:"+userId;
        //判断当前用户是否已经点赞了
        Boolean member = stringRedisTemplate.opsForSet().isMember(key, currentUserId.toString());
        if (BooleanUtil.isTrue(member)) {
            //接着之后就可以从数据库中进行更新了

            //当数据库更新成功之后，才判断:如果已经点赞了，则再点一下的结果是取消点赞，将该 currentUserId 从集合中移除
            stringRedisTemplate.opsForSet().remove(key,currentUserId.toString());

            return false;
        }

        //同理，接着之后就可以从数据库中进行更新了

        //当数据库更新成功之后，才判断:如果之前没有点赞，则点一下的结果是点赞成功，将该 currentUserId 添加到集合中
        stringRedisTemplate.opsForSet().add(key,currentUserId.toString());

        return true;
    }
}

2.1 使用 Redis 实现点赞排行榜功能

在点赞完之后，按照点赞时间先后排序，返回 Top5 的用户，如果单单只靠 Redis 中的 set 数据结构可以实现吗？

很显然是不能实现的，因为 set 不具备排序功能。而 SortedSet 可以根据 score 值进行排序。因此，我们将 score 值设置为时间戳，根据时间戳来进行排序，选取前 5 名点赞用户。

那么就需要将之前点赞的代码进行修改，将 set 更换成 SortedSet 。

代码实现：

1）重新使用 SortedSet 实现点赞功能：

    @Autowired
    StringRedisTemplate stringRedisTemplate;
    //当前用户id

    public Boolean thumbs(Integer userId,Integer currentUserId){
        String key = "userId:"+userId;
        //判断当前用户是否已经点赞了
        Double score = stringRedisTemplate.opsForZSet().score(key, currentUserId.toString());
        if (score != null) {
            //接着之后就可以从数据库中进行更新了

            //当数据库更新成功之后，才判断:如果已经点赞了，则再点一下的结果是取消点赞，将该 currentUserId 从集合中移除
            stringRedisTemplate.opsForZSet().remove(key,currentUserId.toString());

            return false;
        }

        //同理，接着之后就可以从数据库中进行更新了

        //当数据库更新成功之后，才判断:如果之前没有点赞，则点一下的结果是点赞成功，将该 currentUserId 添加到集合中
        stringRedisTemplate.opsForZSet().add(key,currentUserId.toString(),System.currentTimeMillis());

        return true;
    }

2）实现点赞排行榜：

    //根据时间来排序集合中的用户 id 前 5 名
    public void userRange(Integer userId){
        String key =  "userId:"+userId;
        Set<String> range = stringRedisTemplate.opsForZSet().range(key, 0, 4);
        //这就获取到点赞前5名的用户id，最后根据用户id来获取用户的其他消息
        log.info("从用户: "+userId+" 中获取到前5名粉丝ID:"+range);
    }

3）代码测试：

添加不同用户 ID 到集合中：

    @Autowired
    ThumbsUpUtil thumbsUpUtil;
    @Test
    void text3(){
        //模拟添加数据
        thumbsUpUtil.thumbs(1,2);
        thumbsUpUtil.thumbs(1,3);
        thumbsUpUtil.thumbs(1,4);
        thumbsUpUtil.thumbs(1,5);
        thumbsUpUtil.thumbs(1,6);
        thumbsUpUtil.thumbs(1,7);
    }

Top5 排行榜：

    @Test
    void text4(){
        //获取用户ID为1的集合中的前五名用户
        thumbsUpUtil.userRange(1);
    }

运行结果：

前 5 名的用户 ID 是根据时间戳来进行排序的，所以输出结果是没有问题的。

测试相同用户点赞多次：

    @Test
    void text5(){
        //现在用户2想继续对用户1点赞
        thumbsUpUtil.thumbs(1,2);
    }

运行结果：

此时用户 ID 为 2 已经被移除了，这就是说明取消点赞了。

现在用户 1 中粉丝前 5 名的 ID 为：

需要注意的是：当将根据用户 ID 从数据库中获取数据的时候，需要按照传进来的 ID 顺序来得到最终的结果。因为在根据用户 ID 进行批量查询的是用 in(用户 ID ) ，这样返回的结果会按照用户 ID 从大到小进行返回，因此，通过 ORDER BY FIELD(用户 ID 顺序) 命令，才会按照指定的用户 ID 顺序返回数据。

3.0 使用 Redis 实现好友之间的共同关注（共同好友）

用户之间的关注，可以直接用一张数据库表来进行关联，而对于用户与用户之间的共同用户该用什么的方法实现呢？

可以使用 Redis 中的 set 数据结构来实现，利用 set 的不可重复和通过两个集合求得的交集,从而来获取共同的好友。因为不在乎用户顺序，所以不需要用到 SortedSet 数据结构。

代码如下：

准备了两个集合：

求该两个集合的交集：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Component
public class CommonUser {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    public void getCommonUser(String key1,String key2){
        Set<String> common = stringRedisTemplate.opsForSet().intersect(key1, key2);
        if (common != null){
            List<Object> collect = common.stream().map(Long::valueOf).collect(Collectors.toList());
            System.out.println(collect);
        }
    }
}

    @Autowired
    CommonUser commonUser;
    @Test
    void text6(){
        String k1 = "userId:1";
        String k2 = "userId:2";
        commonUser.getCommonUser(k1,k2);
    }

运行结果：

4.0 使用 Redis 实现投喂（发布文章）

关注推送也叫做 Feed 流，直译为投喂。为用户持续的提供"沉浸式"的体验，通过无限下拉刷新获取新的信息。

Feed 流有两种常见的模式：

1）Timeline：不做内容筛选，简单的按照内容发布时间排序，常用与好友或关注。例如朋友圈。

优点：信息全面，不会缺失。并且实现也相对简单。

缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低。

该模式实现的方案有三种：拉模式、推模式、推拉结合。

2）智能排序：利用智能算法屏蔽违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户。

优点：投喂用户感兴趣，用户黏度很高，容易沉迷。

缺点：如果算法不精确，可能起到反作用。

4.1 使用 Redis 实现收件箱（收邮件）

用户将文章推送给自己的粉丝，首先可以根据用户之间的关系表来查询用户的粉丝。获取到粉丝 ID 之后，循环将文章逐个推送给粉丝收件箱中，当前不是将整个文章推送到收件箱中，而是推送的是文章 ID ，文章则保存在数据库中，粉丝可以根据文章 ID 来查询文章内容。

对于粉丝的收件箱可以用 Redis 来实现，由于根据发送的时间来将文章进行排列，所以使用 SortedSet 实现收件箱。key 设置为粉丝 ID ，每一个粉丝都会有一个收件箱；value 设置为文章 ID；score 设置为当前时间戳。

只要是博主发送文章的时候，将文章保存到数据库中，且将文章 ID 发送出去即可。

代码实现：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class InboxUtil {

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    //实现发布文章的时候，将文章推送给好友
    //前缀默认已经实现了:获取到了当前用户的全部好友、将文章已经保存在数据库中
    public void send(Integer articleId, List<Integer> listId,Integer userId){
        String key = "feed:";
        //将文章ID推送到好友收件箱中
        for (Integer id : listId) {
            stringRedisTemplate.opsForZSet().add(key+id, articleId.toString(),System.currentTimeMillis());
        }
    }
}

模拟发布文章，且将文章 ID 推送给好友的收件箱：

代码如下：

    @Autowired
    InboxUtil inboxUtil;
    //实现将文章ID推送给好友
    @Test
    void text7(){
        List<Integer> list = new ArrayList<>(10);
        for (int i = 1; i <= 10; i++) {
            list.add(i);
        }
        Integer articleId = 100;
        inboxUtil.send(articleId,list);

    }

运行结果：

现在用户 1 - 10 收件箱中都有一篇文件的 ID 。

4.1.1 实现滚动分页查询（查询朋友圈信息）

用户查询自己的收件箱，比如像朋友圈，按照时间来排序好友发布的信息，而查看好友的朋友圈的时候，从上至下，来查看好友的消息。

实现思路：

从用户集合中查询信息也不难，在实现的时候不就是直接去查看当前用户的收件箱，其实不然，需要考虑的细节还挺多的。

查看消息的时候，按照时间来查看，从最新的时间到最旧的时间的顺序来进行查看，比如说，朋友圈，打开朋友圈最先看到的是最新的好友信息，

因此使用一下命令来进行查看：

ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]

1）max：最大值，要区分第一次查询和其他查询：第一次查询的时候 max 最大值为当前时间戳；其他几次查询的时候 max 最大值为上一次的时间戳最小值。

2）min：最小值，不需要区分第一次查询和其他查询，可以将其设置为 0 ，将 min 固定为 0 即可。

3）offset：偏移量，需要区分第一次查询和其他查询：第一次查询 offset 设置为 0，偏移量为 0 时，表示当前时间戳可以大于等于此次要查询的时间戳，所以就可以拿到文章 ID ；其他查询 offset 设置为 1 ，一般来说没有问题，但是考虑到特殊情况，万一查询到的在相同的时间戳内发布了多篇文章呢？因此，在其他查询 offset 设置为最小时间戳有多少个。

4）count：每次查询多少个 ID ，一般根据业务制定。

代码实现：

import com.project.volunteermanagementproject.pojo.ScrollQueryDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@Slf4j
public class ScrollQuery {

    @Autowired
    StringRedisTemplate stringRedisTemplate;
    //读取收件箱
    public ScrollQueryDTO read(int id,long max,long offset){
        //用户ID的收件箱
        String key = "feed:"+ id;
        Set<ZSetOperations.TypedTuple<String>> tuples = stringRedisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, 0, max, offset, 3);

        if (tuples == null || tuples.isEmpty()){
            return null;
        }
        List<Integer> articleIds = new ArrayList<>(tuples.size());
        long time = 0;
        long count = 1;
        for (ZSetOperations.TypedTuple<String> tuple : tuples) {
            //获取文章id

            articleIds.add(Integer.valueOf(tuple.getValue()));
            //获取时间戳
            long l = tuple.getScore().longValue();

            if (time == l){
                count++;
            }else {
                time = l;
                count = 1;
            }
        }
        log.info("当前用户查看的文章ID为:"+articleIds);
        log.info("下次的最大时间戳为:"+time+","+"下次的偏移量为:"+count);
        return new ScrollQueryDTO(time, count);
    }
}

测试代码：

    @Autowired
    ScrollQuery scrollQuery;
    @Test
    void text8() throws InterruptedException {
        //当第一次查询的时候，max设置为当前时间戳，而偏移量为0
        int id = 1;
        long max= System.currentTimeMillis();
        long offset = 0;
        while (true){
            ScrollQueryDTO read = scrollQuery.read(id, max, offset);
            if (read == null){
                return;
            }
            max = read.getMax();
            offset = read.getCount();
            Thread.sleep(1000);
        }
    }

运行结果：