| 【Redis企业实战】仿黑马点评项目 | 您所在的位置：网站首页 › 黑马yfis › 【Redis企业实战】仿黑马点评项目 |
【Redis企业实战】仿黑马点评项目
|
目录 一、短信登陆：基于Redis实现共享session实现登录 1、发送短信验证码 2、短信验证码登录、注册 3、校验登陆状态 二、商户查询缓存 1、添加Redis缓存 2、缓存更新策略： 3、缓存穿透 4、缓存雪崩 5、缓存击穿： 三、优惠券秒杀（Redisson分布式锁解决超卖、一人一单，Stream消息队列实现异步秒杀） 1、全局唯一ID 2、超卖问题 3、基于Redis的分布式锁 4、实现Redisson的分布式锁 5、基于Redis的Stream结构作为消息队列，实现异步秒杀下单 四、达人探店 1、发布探店笔记（见Feed流） 2、点赞 3、点赞排行榜 五、好友关注 1、关注和取关 2、共同关注 3、关注推送(Feed流) 六、附近的商户 1、GEO数据结构 2、附近商户搜索 七、用户签到 1、BitMap用法 2、签到功能 3、签到统计 完整版项目代码：dazhongdianping: 仿大众点评 (gitee.com) 一、短信登陆：基于Redis实现共享session实现登录 1、发送短信验证码使用Redis中的String结构，将key:phone,value:code存入Redis @Override public Result sendCode(String phone, HttpSession session) { // 1.校验手机号 if (RegexUtils.isPhoneInvalid(phone)) { // 2.如果不符合，返回错误信息 return Result.fail("手机号格式错误！"); } // 3.符合，生成验证码 String code = RandomUtil.randomNumbers(6); // 4.保存验证码到 session stringRedisTemplate.opsForValue().set(LOGIN_CODE_KEY + phone, code, LOGIN_CODE_TTL, TimeUnit.MINUTES); // 5.发送验证码 log.debug("发送短信验证码成功，验证码：{}", code); // 返回ok return Result.ok(); } 2、短信验证码登录、注册根据手机手机号查询登录用户，用户存在则将查询的用户存入redis，用户不存在创建用户再存入redis，采用保存登录的用户信息，可以使用String结构，以JSON字符串来保存，比较直观。Hash结构可以将对象中的每个字段独立存储，可以针对单个字段做CRUD，并且内存占用更少。 public Result login(LoginFormDTO loginForm, HttpSession session) { // 1.校验手机号 String phone = loginForm.getPhone(); if (RegexUtils.isPhoneInvalid(phone)) { // 2.如果不符合，返回错误信息 return Result.fail("手机号格式错误！"); } // 3.从redis获取验证码并校验 String cacheCode = stringRedisTemplate.opsForValue().get(LOGIN_CODE_KEY + phone); String code = loginForm.getCode(); if (cacheCode == null || !cacheCode.equals(code)) { // 不一致，报错 return Result.fail("验证码错误"); } // 4.一致，根据手机号查询用户 select * from tb_user where phone = ? User user = query().eq("phone", phone).one(); // 5.判断用户是否存在 if (user == null) { // 6.不存在，创建新用户并保存 user = createUserWithPhone(phone); } // 7.保存用户信息到 redis中 // 7.1.随机生成token，作为登录令牌 String token = UUID.randomUUID().toString(true); // 7.2.将User对象转为HashMap存储 UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class); Map userMap = BeanUtil.beanToMap(userDTO, new HashMap(), CopyOptions.create() .setIgnoreNullValue(true) .setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString())); // 7.3.存储 String tokenKey = LOGIN_USER_KEY + token; stringRedisTemplate.opsForHash().putAll(tokenKey, userMap); // 7.4.设置token有效期 stringRedisTemplate.expire(tokenKey, LOGIN_USER_TTL, TimeUnit.MINUTES); // 8.返回token return Result.ok(token); } 3、校验登陆状态为每一个UserDTO创建一个独立线程，使多线程之间资源无法共享。创建二级拦截器，第一级拦截器：1.获取token。2.查询redis用户。3.保存到ThreadLocal，4.刷新token有效期。5.放行。第二级拦截器：1.查询Threadlocal的用户。2.不存在，拦截。3.存在，放行。 UserHolder： public class UserHolder { private static final ThreadLocal tl = new ThreadLocal(); public static void saveUser(UserDTO user){ tl.set(user); } public static UserDTO getUser(){ return tl.get(); } public static void removeUser(){ tl.remove(); } }拦截器： public void addInterceptors(InterceptorRegistry registry) { // 登录拦截器 registry.addInterceptor(new LoginInterceptor()) .excludePathPatterns( "/shop/**", "/voucher/**", "/shop-type/**", "/upload/**", "/blog/hot", "/user/code", "/user/login" ).order(1); // token刷新的拦截器 registry.addInterceptor(new RefreshTokenInterceptor(stringRedisTemplate)).addPathPatterns("/**").order(0); }Redis代替session需要考虑的问题： 选择合适的数据结构选择合适的key选择合适的存储粒度 二、商户查询缓存 1、添加Redis缓存
内存淘汰 超时剔除 主动更新 说明 不用自己维护，利用Redis的内存淘汰机制，当内存不足时自动淘汰部分数据。下次查询时更新缓存。 给缓存数据添加TTL时间，到期后自动删除缓存。下次查询时更新缓存。 编写业务逻辑，在修改数据库的同时，更新缓存。 一致性 差 一般 好 维护成本 无 低 高 缓存更新策略之主动更新 1. Cache Aside Pattern：由缓存的调用者（程序员），在更新数据库的同时更新缓存 2. Read/Write Through Pattern：缓存与数据库整合为一个服务，由服务来维护一致性。调用者调用该服务，无需关心缓存一致性问题 3. Write Behind Caching Pattern：调用者只操作缓存，由其它线程异步的将缓存数据持久化到数据库，保证最终一致。 操作缓存和数据库时有三个问题需要考虑： 1. 删除缓存还是更新缓存？ 更新缓存：每次更新数据库都更新缓存，无效写操作较多删除缓存：更新数据库时让缓存失效，查询时再更新缓存 2. 如何保证缓存与数据库的操作的同时成功或失败？ 单体系统，将缓存与数据库操作放在一个事务分布式系统，利用TCC等分布式事务方案缓存更新策略的最佳实践方案： 低一致性需求：使用Redis自带的内存淘汰机制高一致性需求：主动更新，并以超时剔除作为兜底方案 读操作： 缓存命中则直接返回缓存未命中则查询数据库，并写入缓存，设定超时时间 写操作： 先写数据库，然后再删除缓存要确保数据库与缓存操作的原子性 3、缓存穿透缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。 常见的解决方案有两种： 优点缺点缓存空对象实现简单，维护方便额外内存消耗 可能造成短期不一致 布隆过滤内存占用较少，没有多余key实现复杂 存在误判可能
缓存穿透产生的原因是什么？ 用户请求的数据在缓存中和数据库中都不存在，不断发起这样的请求，给数据库带来巨大压力缓存穿透的解决方案有哪些？ 缓存null值布隆过滤增强id的复杂度，避免被猜测id规律做好数据的基础格式校验加强用户权限校验做好热点参数的限流 public R queryWithPassThrough( String keyPrefix, ID id, Class type, Function dbFallback, Long time, TimeUnit unit){ String key = keyPrefix + id; // 1.从redis查询商铺缓存 String json = stringRedisTemplate.opsForValue().get(key); // 2.判断是否存在 if (StrUtil.isNotBlank(json)) { // 3.存在，直接返回 return JSONUtil.toBean(json, type); } // 判断命中的是否是空值 if (json != null) { // 返回一个错误信息 return null; } // 4.不存在，根据id查询数据库 R r = dbFallback.apply(id); // 5.不存在，返回错误 if (r == null) { // 将空值写入redis stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES); // 返回错误信息 return null; } // 6.存在，写入redis this.set(key, r, time, unit); return r; } 4、缓存雪崩缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。
解决方案： 给不同的Key的TTL添加随机值利用Redis集群提高服务的可用性给缓存业务添加降级限流策略给业务添加多级缓存 5、缓存击穿：缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。 解决方案： 互斥锁逻辑过期解决方案 优点 缺点 互斥锁 • 没有额外的内存消耗 • 保证一致性 • 实现简单 • 线程需要等待，性能受影响 • 可能有死锁风险逻辑过期 • 线程无需等待，性能较好 • 不保证一致性 • 有额外内存消耗 • 实现复杂互斥锁解决缓存击穿实现代码： public R queryWithMutex( String keyPrefix, ID id, Class type, Function dbFallback, Long time, TimeUnit unit) { String key = keyPrefix + id; // 1.从redis查询商铺缓存 String shopJson = stringRedisTemplate.opsForValue().get(key); // 2.判断是否存在 if (StrUtil.isNotBlank(shopJson)) { // 3.存在，直接返回 return JSONUtil.toBean(shopJson, type); } // 判断命中的是否是空值 if (shopJson != null) { // 返回一个错误信息 return null; } // 4.实现缓存重建 // 4.1.获取互斥锁 String lockKey = LOCK_SHOP_KEY + id; R r = null; try { boolean isLock = tryLock(lockKey); // 4.2.判断是否获取成功 if (!isLock) { // 4.3.获取锁失败，休眠并重试 Thread.sleep(50); return queryWithMutex(keyPrefix, id, type, dbFallback, time, unit); } // 4.4.获取锁成功，根据id查询数据库 r = dbFallback.apply(id); // 5.不存在，返回错误 if (r == null) { // 将空值写入redis stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.MINUTES); // 返回错误信息 return null; } // 6.存在，写入redis this.set(key, r, time, unit); } catch (InterruptedException e) { throw new RuntimeException(e); }finally { // 7.释放锁 unlock(lockKey); } // 8.返回 return r; } private boolean tryLock(String key) { Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS); return BooleanUtil.isTrue(flag); } private void unlock(String key) { stringRedisTemplate.delete(key); }逻辑过期解决缓存击穿实现代码： public R queryWithLogicalExpire( String keyPrefix, ID id, Class type, Function dbFallback, Long time, TimeUnit unit) { String key = keyPrefix + id; // 1.从redis查询商铺缓存 String json = stringRedisTemplate.opsForValue().get(key); // 2.判断是否存在 if (StrUtil.isBlank(json)) { // 3.存在，直接返回 return null; } // 4.命中，需要先把json反序列化为对象 RedisData redisData = JSONUtil.toBean(json, RedisData.class); R r = JSONUtil.toBean((JSONObject) redisData.getData(), type); LocalDateTime expireTime = redisData.getExpireTime(); // 5.判断是否过期 if(expireTime.isAfter(LocalDateTime.now())) { // 5.1.未过期，直接返回店铺信息 return r; } // 5.2.已过期，需要缓存重建 // 6.缓存重建 // 6.1.获取互斥锁 String lockKey = LOCK_SHOP_KEY + id; boolean isLock = tryLock(lockKey); // 6.2.判断是否获取锁成功 if (isLock){ // 6.3.成功，开启独立线程，实现缓存重建 CACHE_REBUILD_EXECUTOR.submit(() -> { try { // 查询数据库 R newR = dbFallback.apply(id); // 重建缓存 this.setWithLogicalExpire(key, newR, time, unit); } catch (Exception e) { throw new RuntimeException(e); }finally { // 释放锁 unlock(lockKey); } }); } // 6.4.返回过期的商铺信息 return r; } 三、优惠券秒杀（Redisson分布式锁解决超卖、一人一单，Stream消息队列实现异步秒杀） 1、全局唯一ID全局唯一ID生成策略： UUIDRedis自增snowflake算法数据库自增Redis自增ID策略： 每天一个key，方便统计订单量ID构造是 时间戳 + 计数器 2、超卖问题超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是加锁： 超卖这样的线程安全问题，解决方案有哪些？ 1.悲观锁： 添加同步锁，让线程串行执行，认为线程安全问题一定会发生，因此在操作数据之前先获取锁，确保线程串行执行 优点：简单粗暴缺点：性能一般实现：Synchronized、Lock 2.乐观锁： 不加锁，在更新时判断是否有其它线程在修改。 认为线程安全问题不一定会发生，因此不加锁，只是在更新数据时去判断有没有其它线程对数据做了修改。 优点：性能好缺点：存在成功率低的问题实现：版本号法、CAS法 3、基于Redis的分布式锁一人一单问题：通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。可采用分布式锁。 分布式锁：
基于Redis的分布式锁实现思路： 利用set nx ex获取锁，并设置过期时间，保存线程标示释放锁时先判断线程标示是否与自己一致，一致则删除锁特性： 利用set nx满足互斥性利用set ex保证故障时锁依然能释放，避免死锁，提高安全性利用Redis集群保证高可用和高并发特性基于setnx实现的分布式锁存在下面的问题： 不可重入：同一个线程无法多次获取同一把锁不可重试：获取锁只尝试一次就返回false，没有重试机制超市释放：锁超时释放虽然可以避免死锁，但如果是业务执行耗时较长，也会导致锁释放，存在安全隐患主从一致性：如果Redis提供了主从集群，主从同步存在延迟，当主宕机时，如果从并同步主中的锁数据，则会出现锁实现 4、实现Redisson的分布式锁Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。 Redisson可重入锁原理：
Redisson分布式锁原理： 利用Redisson解决创建订单时一人一单问题实现代码 private void createVoucherOrder(VoucherOrder voucherOrder) { Long userId = voucherOrder.getUserId(); Long voucherId = voucherOrder.getVoucherId(); // 创建锁对象 RLock redisLock = redissonClient.getLock("lock:order:" + userId); // 尝试获取锁 boolean isLock = redisLock.tryLock(); // 判断 if (!isLock) { // 获取锁失败，直接返回失败或者重试 log.error("不允许重复下单！"); return; } try { // 5.1.查询订单 int count = query().eq("user_id", userId).eq("voucher_id", voucherId).count(); // 5.2.判断是否存在 if (count > 0) { // 用户已经购买过了 log.error("不允许重复下单！"); return; } // 6.扣减库存 boolean success = seckillVoucherService.update() .setSql("stock = stock - 1") // set stock = stock - 1 .eq("voucher_id", voucherId).gt("stock", 0) // where id = ? and stock > 0 .update(); if (!success) { // 扣减失败 log.error("库存不足！"); return; } // 7.创建订单 save(voucherOrder); } finally { // 释放锁 redisLock.unlock(); } }总结： 1）不可重入Redis分布式锁： 原理：利用setnx的互斥性；利用ex避免死锁；释放锁时判断线程标示缺陷：不可重入、无法重试、锁超时失效2）可重入的Redis分布式锁： 原理：利用hash结构，记录线程标示和重入次数；利用watchDog延续锁时间；利用信号量控制锁重试等待缺陷：redis宕机引起锁失效问题3）Redisson的multiLock： 原理：多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功缺陷：运维成本高、实现复杂 5、基于Redis的Stream结构作为消息队列，实现异步秒杀下单异步秒杀下单流程图
消息队列（Message Queue），字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色： 消息队列：存储和管理消息，也被称为消息代理（Message Broker）生产者：发送消息到消息队列消费者：从消息队列获取消息并处理消息Redis提供了三种不同的方式来实现消息队列： list结构：基于List结构模拟消息队列PubSub：基本的点对点消息模型Stream：比较完善的消息队列模型List PubSub Stream 消息持久化 支持 不支持 支持 阻塞读取 支持 支持 支持 消息堆积处理 受限于内存空间，可以利用多消费者加快处理 受限于消费者缓冲区 受限于队列长度，可以利用消费者组提高消费速度，减少堆积 消息确认机制 不支持 不支持 支持 消息回溯 不支持 不支持 支持 其余优点利用Redis存储，不受限于JVM内存上限 可以满足消息有序性 采用发布订阅模型，支持多生产、多消费XREADGROUP:可以多消费者抢消息，加快消费速度，没有消息漏读风险 XREAD：有消息漏读风险 缺点无法避免消息丢失 只支持单消费者 无法避免消息丢失 消息堆积有上限，超出时数据丢失 基于Stream的消息队列 消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点： 消息分流：队列中的消息会分流给组内的不同消费者，而不是重复消费，从而加快消息处理的速度 消息标示：消费者组会维护一个标示，记录最后一个被处理的消息，哪怕消费者宕机重启，还会从标示之后读取消息。确保每一个 消息都会被消费 消息确认：消费者获取消息后，消息处于pending状态，并存入一个pending-list。当处理完成后需要通过XACK来确认消息，标记 消息为已处理，才会从pending-list移除。利用Stream结果实现异步秒杀实现代码：（lua+java） -- 1.参数列表 -- 1.1.优惠券id local voucherId = ARGV[1] -- 1.2.用户id local userId = ARGV[2] -- 1.3.订单id local orderId = ARGV[3] -- 2.数据key -- 2.1.库存key local stockKey = 'seckill:stock:' .. voucherId -- 2.2.订单key local orderKey = 'seckill:order:' .. voucherId -- 3.脚本业务 -- 3.1.判断库存是否充足 get stockKey if(tonumber(redis.call('get', stockKey)) List list = stringRedisTemplate.opsForStream().read( Consumer.from("g1", "c1"), StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)), StreamOffset.create("stream.orders", ReadOffset.lastConsumed()) ); // 2.判断订单信息是否为空 if (list == null || list.isEmpty()) { // 如果为null，说明没有消息，继续下一次循环 continue; } // 解析数据 MapRecord record = list.get(0); Map value = record.getValue(); VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 3.创建订单 createVoucherOrder(voucherOrder); // 4.确认消息 XACK stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId()); } catch (Exception e) { log.error("处理订单异常", e); handlePendingList(); } } } private void handlePendingList() { while (true) { try { // 1.获取pending-list中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 0 List list = stringRedisTemplate.opsForStream().read( Consumer.from("g1", "c1"), StreamReadOptions.empty().count(1), StreamOffset.create("stream.orders", ReadOffset.from("0")) ); // 2.判断订单信息是否为空 if (list == null || list.isEmpty()) { // 如果为null，说明没有异常消息，结束循环 break; } // 解析数据 MapRecord record = list.get(0); Map value = record.getValue(); VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true); // 3.创建订单 createVoucherOrder(voucherOrder); // 4.确认消息 XACK stringRedisTemplate.opsForStream().acknowledge("s1", "g1", record.getId()); } catch (Exception e) { log.error("处理订单异常", e); } } } } 四、达人探店 1、发布探店笔记（见Feed流） 2、点赞Redis缓存点赞信息，SortedSet集合 public Result likeBlog(Long id) { // 1.获取登录用户 Long userId = UserHolder.getUser().getId(); // 2.判断当前登录用户是否已经点赞 String key = BLOG_LIKED_KEY + id; Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString()); if (score == null) { // 3.如果未点赞，可以点赞 // 3.1.数据库点赞数 + 1 boolean isSuccess = update().setSql("liked = liked + 1").eq("id", id).update(); // 3.2.保存用户到Redis的set集合 zadd key value score if (isSuccess) { stringRedisTemplate.opsForZSet().add(key, userId.toString(), System.currentTimeMillis()); } } else { // 4.如果已点赞，取消点赞 // 4.1.数据库点赞数 -1 boolean isSuccess = update().setSql("liked = liked - 1").eq("id", id).update(); // 4.2.把用户从Redis的set集合移除 if (isSuccess) { stringRedisTemplate.opsForZSet().remove(key, userId.toString()); } } return Result.ok(); } 3、点赞排行榜获取Top5点赞用户 public Result queryBlogLikes(Long id) { String key = BLOG_LIKED_KEY + id; // 1.查询top5的点赞用户 zrange key 0 4 Set top5 = stringRedisTemplate.opsForZSet().range(key, 0, 4); if (top5 == null || top5.isEmpty()) { return Result.ok(Collections.emptyList()); } // 2.解析出其中的用户id List ids = top5.stream().map(Long::valueOf).collect(Collectors.toList()); String idStr = StrUtil.join(",", ids); // 3.根据用户id查询用户 WHERE id IN ( 5 , 1 ) ORDER BY FIELD(id, 5, 1) List userDTOS = userService.query() .in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list() .stream() .map(user -> BeanUtil.copyProperties(user, UserDTO.class)) .collect(Collectors.toList()); // 4.返回 return Result.ok(userDTOS); } 五、好友关注 1、关注和取关实现代码 public Result follow(Long followUserId, Boolean isFollow) { // 1.获取登录用户 Long userId = UserHolder.getUser().getId(); String key = "follows:" + userId; // 1.判断到底是关注还是取关 if (isFollow) { // 2.关注，新增数据 Follow follow = new Follow(); follow.setUserId(userId); follow.setFollowUserId(followUserId); boolean isSuccess = save(follow); if (isSuccess) { // 把关注用户的id，放入redis的set集合 sadd userId followerUserId stringRedisTemplate.opsForSet().add(key, followUserId.toString()); } } else { // 3.取关，删除 delete from tb_follow where user_id = ? and follow_user_id = ? boolean isSuccess = remove(new QueryWrapper() .eq("user_id", userId).eq("follow_user_id", followUserId)); if (isSuccess) { // 把关注用户的id从Redis集合中移除 stringRedisTemplate.opsForSet().remove(key, followUserId.toString()); } } return Result.ok(); } public Result isFollow(Long followUserId) { // 1.获取登录用户 Long userId = UserHolder.getUser().getId(); // 2.查询是否关注 select count(*) from tb_follow where user_id = ? and follow_user_id = ? Integer count = query().eq("user_id", userId).eq("follow_user_id", followUserId).count(); // 3.判断 return Result.ok(count > 0); } 2、共同关注Set集合的交集功能实现代码 public Result followCommons(Long id) { // 1.获取当前用户 Long userId = UserHolder.getUser().getId(); String key = "follows:" + userId; // 2.求交集 String key2 = "follows:" + id; Set intersect = stringRedisTemplate.opsForSet().intersect(key, key2); if (intersect == null || intersect.isEmpty()) { // 无交集 return Result.ok(Collections.emptyList()); } // 3.解析id集合 List ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList()); // 4.查询用户 List users = userService.listByIds(ids) .stream() .map(user -> BeanUtil.copyProperties(user, UserDTO.class)) .collect(Collectors.toList()); return Result.ok(users); } 3、关注推送(Feed流)Feed流产品有两种常见模式： Timeline ：不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈 优点：信息全面，不会有缺失。并且实现也相对简单缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低 智能排序 ：利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户 优点：投喂用户感兴趣信息，用户粘度很高，容易沉迷缺点：如果算法不精准，可能起到反作用本例中的个人页面，是基于关注的好友来做Feed流，因此采用Timeline的模式。该模式的实现方案有三种： 推模式（也叫写扩散）：和名字一样，就是一种推的方式，发送者发送了一个消息后，立即将这个消息推送给接收者，但是接收者此时不一定在线，那么就需要有一个地方存储这个数据，这个存储的地方我们称为：同步库。推模式也叫写扩散的原因是，一个消息需要发送个多个粉丝，那么这条消息就会复制多份，写放大，所以也叫写扩散。这种模式下，对同步库的要求就是写入能力极强和稳定。读取的时候因为消息已经发到接收者的收件箱了，只需要读一次自己的收件箱即可，读请求的量极小，所以对读的QPS需求不大。归纳下，推模式中对同步库的要求只有一个：写入能力强。 拉模式（也叫读扩散）：这种是一种拉的方式，发送者发送了一条消息后，这条消息不会立即推送给粉丝，而是写入自己的发件箱，当粉丝上线后再去自己关注者的发件箱里面去读取，一条消息的写入只有一次，但是读取最多会和粉丝数一样，读会放大，所以也叫读扩散。拉模式的读写比例刚好和写扩散相反，那么对系统的要求是：读取能力强。另外这里还有一个误区，很多人在最开始设计feed流系统时，首先想到的是拉模式，因为这种和用户的使用体感是一样的，但是在系统设计上这种方式有不少痛点，最大的是每个粉丝需要记录自己上次读到了关注者的哪条消息，如果有1000个关注者，那么这个人需要记录1000个位置信息，这个量和关注量成正比的，远比用户数要大的多，这里要特别注意，虽然在产品前期数据量少的时候这种方式可以应付，但是量大了后就会事倍功半，得不偿失，切记切记。 推拉结合模式：推模式在单向关系中，因为存在大V，那么一条消息可能会扩散几百万次，但是这些用户中可能有一半多是僵尸，永远不会上线，那么就存在资源浪费。而拉模式下，在系统架构上会很复杂，同时需要记录的位置信息是天量，不好解决，尤其是用户量多了后会成为第一个故障点。基于此，所以有了推拉结合模式，大部分用户的消息都是写扩散，只有大V是读扩散，这样既控制了资源浪费，又减少了系统设计复杂度。但是整体设计复杂度还是要比推模式复杂。 归纳如下： 如果产品中是双向关系，那么就采用推模式。 如果产品中是单向关系，且用户数少于1000万，那么也采用推模式，足够了。 如果产品是单向关系，单用户数大于1000万，那么采用推拉结合模式，这时候可以从推模式演进过来，不需要额外重新推翻重做。 永远不要只用拉模式。 如果是一个初创企业，先用推模式，快速把系统设计出来，然后让产品去验证、迭代，等客户数大幅上涨到1000万后，再考虑升级为推拉集合模式。 拉模式 推模式 推拉结合 写比例 低 高 中 读比例 高 低 中 用户读取延迟 高 低 低 实现难度 复杂 简单 很复杂 使用场景 很少使用 用户量少、没有大V 过千万的用户量，有大V 本项目采用推模式：用SortedSet保存博客id，以及时间戳推送给粉丝收件箱实现代码 public Result saveBlog(Blog blog) { // 1.获取登录用户 UserDTO user = UserHolder.getUser(); blog.setUserId(user.getId()); // 2.保存探店笔记 boolean isSuccess = save(blog); if(!isSuccess){ return Result.fail("新增笔记失败!"); } // 3.查询笔记作者的所有粉丝 select * from tb_follow where follow_user_id = ? List follows = followService.query().eq("follow_user_id", user.getId()).list(); // 4.推送笔记id给所有粉丝 for (Follow follow : follows) { // 4.1.获取粉丝id Long userId = follow.getUserId(); // 4.2.推送 String key = FEED_KEY + userId; stringRedisTemplate.opsForZSet().add(key, blog.getId().toString(), System.currentTimeMillis()); } // 5.返回id return Result.ok(blog.getId()); }粉丝收件箱接收实现代码 public Result queryBlogOfFollow(Long max, Integer offset) { // 1.获取当前用户 Long userId = UserHolder.getUser().getId(); // 2.查询收件箱 ZREVRANGEBYSCORE key Max Min LIMIT offset count String key = FEED_KEY + userId; Set typedTuples = stringRedisTemplate.opsForZSet() .reverseRangeByScoreWithScores(key, 0, max, offset, 2); // 3.非空判断 if (typedTuples == null || typedTuples.isEmpty()) { return Result.ok(); } // 4.解析数据：blogId、minTime（时间戳）、offset List ids = new ArrayList(typedTuples.size()); long minTime = 0; // 2 int os = 1; // 2 for (ZSetOperations.TypedTuple tuple : typedTuples) { // 5 4 4 2 2 // 4.1.获取id ids.add(Long.valueOf(tuple.getValue())); // 4.2.获取分数(时间戳） long time = tuple.getScore().longValue(); if(time == minTime){ os++; }else{ minTime = time; os = 1; } } // 5.根据id查询blog String idStr = StrUtil.join(",", ids); List blogs = query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list(); for (Blog blog : blogs) { // 5.1.查询blog有关的用户 queryBlogUser(blog); // 5.2.查询blog是否被点赞 isBlogLiked(blog); } // 6.封装并返回 ScrollResult r = new ScrollResult(); r.setList(blogs); r.setOffset(os); r.setMinTime(minTime); return Result.ok(r); } 六、附近的商户 1、GEO数据结构GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。常见的命令有： GEOADD：添加一个地理空间信息，包含：经度（longitude）、纬度（latitude）、值（member） GEODIST：计算指定的两个点之间的距离并返回 GEOHASH：将指定member的坐标转为hash字符串形式并返回 GEOPOS：返回指定member的坐标 GEORADIUS：指定圆心、半径，找到该圆内包含的所有member，并按照与圆心之间的距离排序后返回。6.2以后已废弃 GEOSEARCH：在指定范围内搜索member，并按照与指定点之间的距离排序后返回。范围可以是圆形或矩形。6.2.新功能 GEOSEARCHSTORE：与GEOSEARCH功能一致，不过可以把结果存储到一个指定的key。 6.2.新功能 2、附近商户搜索按照商户类型做分组，类型相同的商户作为同一组，以typeId为key存入同一个GEO集合中即可 public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) { // 1.判断是否需要根据坐标查询 if (x == null || y == null) { // 不需要坐标查询，按数据库查询 Page page = query() .eq("type_id", typeId) .page(new Page(current, SystemConstants.DEFAULT_PAGE_SIZE)); // 返回数据 return Result.ok(page.getRecords()); } // 2.计算分页参数 int from = (current - 1) * SystemConstants.DEFAULT_PAGE_SIZE; int end = current * SystemConstants.DEFAULT_PAGE_SIZE; // 3.查询redis、按照距离排序、分页。结果：shopId、distance String key = SHOP_GEO_KEY + typeId; GeoResults results = stringRedisTemplate.opsForGeo() // GEOSEARCH key BYLONLAT x y BYRADIUS 10 WITHDISTANCE .search( key, GeoReference.fromCoordinate(x, y), new Distance(5000), RedisGeoCommands.GeoSearchCommandArgs.newGeoSearchArgs().includeDistance().limit(end) ); // 4.解析出id if (results == null) { return Result.ok(Collections.emptyList()); } List list = results.getContent(); if (list.size() { // 4.2.获取店铺id String shopIdStr = result.getContent().getName(); ids.add(Long.valueOf(shopIdStr)); // 4.3.获取距离 Distance distance = result.getDistance(); distanceMap.put(shopIdStr, distance); }); // 5.根据id查询Shop String idStr = StrUtil.join(",", ids); List shops = query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list(); for (Shop shop : shops) { shop.setDistance(distanceMap.get(shop.getId().toString()).getValue()); } // 6.返回 return Result.ok(shops); } 七、用户签到 1、BitMap用法Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是 2^32个bit位。 BitMap的操作命令有： SETBIT：向指定位置（offset）存入一个0或1 GETBIT ：获取指定位置（offset）的bit值 BITCOUNT ：统计BitMap中值为1的bit位的数量 BITFIELD ：操作（查询、修改、自增）BitMap中bit数组中的指定位置（offset）的值 2、签到功能实现代码： public Result sign() { // 1.获取当前登录用户 Long userId = UserHolder.getUser().getId(); // 2.获取日期 LocalDateTime now = LocalDateTime.now(); // 3.拼接key String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM")); String key = USER_SIGN_KEY + userId + keySuffix; // 4.获取今天是本月的第几天 int dayOfMonth = now.getDayOfMonth(); // 5.写入Redis SETBIT key offset 1 stringRedisTemplate.opsForValue().setBit(key, dayOfMonth - 1, true); return Result.ok(); } 3、签到统计实现代码： public Result signCount() { // 1.获取当前登录用户 Long userId = UserHolder.getUser().getId(); // 2.获取日期 LocalDateTime now = LocalDateTime.now(); // 3.拼接key String keySuffix = now.format(DateTimeFormatter.ofPattern(":yyyyMM")); String key = USER_SIGN_KEY + userId + keySuffix; // 4.获取今天是本月的第几天 int dayOfMonth = now.getDayOfMonth(); // 5.获取本月截止今天为止的所有的签到记录，返回的是一个十进制的数字 BITFIELD sign:5:202203 GET u14 0 List result = stringRedisTemplate.opsForValue().bitField( key, BitFieldSubCommands.create() .get(BitFieldSubCommands.BitFieldType.unsigned(dayOfMonth)).valueAt(0) ); if (result == null || result.isEmpty()) { // 没有任何签到结果 return Result.ok(0); } Long num = result.get(0); if (num == null || num == 0) { return Result.ok(0); } // 6.循环遍历 int count = 0; while (true) { // 6.1.让这个数字与1做与运算，得到数字的最后一个bit位 // 判断这个bit位是否为0 if ((num & 1) == 0) { // 如果为0，说明未签到，结束 break; }else { // 如果不为0，说明已签到，计数器+1 count++; } // 把数字右移一位，抛弃最后一个bit位，继续下一个bit位 num >>>= 1; } return Result.ok(count); } |
【本文地址】
公司简介
联系我们