tomcat的运行原理：
当用户发起请求时，会访问tomcat注册的端口，任何程序想要运行，都需要有一个线程对当前端口号进行监听，当用户和tomcat连接连接时，那会由监听线程创建socket连接，socket都是成对出现的，用户通过socket像互相传递数据，当tomcat端的socket接受到数据后，此时监听线程会从tomcat的线程池中取出一个线程执行用户请求，在我们的服务部署到tomcat后，线程会找到用户想要访问的工程，然后用这个线程转发到工程中的controller，service，dao中，并且访问对应的DB，在用户执行完请求后，再统一返回，再找到tomcat端的socket，再将数据写回到用户端的socket，完成请求和响应
由此我们可以得知 每个用户其实对应都是去找tomcat线程池中的一个线程来完成工作的， 使用完成后再进行回收，所以在每个用户去访问我们的工程时，我们可以使用threadlocal来做到线程隔离，每个线程操作自己的一份数据
在threadLocal中，无论是他的put方法和他的get方法， 都是先从获得当前用户的线程，然后从线程中取出线程的成员变量map，只要线程不一样，map就不一样，所以可以通过这种方式来做到线程隔离。
短信验证码登录
//发送验证码
@Override
public Result sendCode(String phone, HttpSession session) {
if (RegexUtils.isPhoneInvalid(phone)) {
return Result.fail("手机号码格式错误");
}
String code = RandomUtil.randomNumbers(6);
//保存到session
session.setAttribute("code",code);
log.debug("验证成功，发从验证码：{}",code);
return Result.ok();
}
//登录功能
@Override
public Result login(LoginFormDTO loginForm, HttpSession session) {
String phone = loginForm.getPhone();
if (RegexUtils.isPhoneInvalid(phone)) {
return Result.fail("手机号码格式错误");
}
Object cacheCode = session.getAttribute("code");
String code = loginForm.getCode();
if(cacheCode == null || !cacheCode.equals(code)){
return Result.fail("验证码错误");
}
//一致，根据手机号查询用户
User user = query().eq("phone", phone).one();
if(user == null){
//不存在，则创建
user = createUserWithPhone(phone);
}
//保存用户信息到session中
session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));//注意BeanUtil的包位置
return Result.ok(token);
}
登录
public class LoginInterceptor implements HandlerInterceptor {
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
HttpSession session = request.getSession();
// 1、判断用户是否存在
User user = (User) session.getAttribute("user");
if (Objects.isNull(user)){
// 用户不存在，直接拦截
response.setStatus(401);
return false;
}
// 2、用户存在，则将用户信息保存到ThreadLocal中，方便后续逻辑处理
// 比如：方便获取和使用用户信息，session获取用户信息是具有侵入性的
ThreadLocalUtls.saveUser(user);
return HandlerInterceptor.super.preHandle(request, response, handler);
}
}
将自定义的添加到SpringMVC的列表中:
@Configuration
public class WebMvcConfig implements WebMvcConfigurer {
@Override
public void addInterceptors(InterceptorRegistry registry) {
// 添加登录
registry.addInterceptor(new LoginInterceptor())
// 设置放行请求
.excludePathPatterns(
"/user/code",
"/user/login",
"/blog/hot",
"/shop/**",
"/shop-type/**",
"/upload/**",
"/voucher/**"
);
}
}
session共享问题：
每个tomcat中都有一份属于自己的session,用户第一次访问一台tomcat，并且把自己的信息存放到第一台服务器的session中，但是第二次这个用户访问到了第二台tomcat，那么在第二台服务器上，肯定没有第一台服务器存放的session，所以此时整个登录拦截功能就会出现问题。早期的方案是session拷贝，但是每当任意一台服务器的session修改时，都会同步给其他的Tomcat服务器的session
但是这种方案具有两个大问题
解决办法：我们可以使用Redis实现，因为Redis数据本身就是共享的
整体访问流程：
//发送验证码
@Override
public Result sendCode(String phone, HttpSession session) {
if (RegexUtils.isPhoneInvalid(phone)) {
return Result.fail("手机号码格式错误");
}
String code = RandomUtil.randomNumbers(6);
stringRedisTemplate.opsForValue().set("login:code" + phone,code,2, TimeUnit.MINUTES);
log.debug("验证成功，发从验证码：{}",code);
return Result.ok();
}
//登录
public Result login(LoginFormDTO loginForm, HttpSession session) {
String phone = loginForm.getPhone();
if (RegexUtils.isPhoneInvalid(phone)) {
return Result.fail("手机号码格式错误");
}
//从redis中获取验证码并校验
String cacheCode = stringRedisTemplate.opsForValue().get("login:code" + phone);
String code = loginForm.getCode();
if(cacheCode == null || !cacheCode.equals(code)){
return Result.fail("验证码错误");
}
//一致，根据手机号查询用户
User user = query().eq("phone", phone).one();
if(user == null){
//不存在，则创建
user = createUserWithPhone(phone);
}
//保存用户信息到redis
String token = UUID.randomUUID().toString();
UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
//！！！因为使用的是stringRedisTemplate，map的值必须为String类型
Map<String, Object> userMap = BeanUtil.beanToMap(userDTO,new HashMap<>(), CopyOptions.create()
.setIgnoreNullValue(true) //忽略空的值，在转换过程中不包括空值的字段
.setFieldValueEditor((fieldName,fieldValue) -> fieldValue.toString())); //修改字段值：将字段值转为字符串
stringRedisTemplate.opsForHash().putAll("login:token" + token,userMap);
stringRedisTemplate.expire("login:token" + token,30,TimeUnit.MINUTES);
return Result.ok(token);
}
token刷新问题
将token刷新等功能放在一个中确实也可以实现token刷新，但当用户访问了不需要拦截的接口不会生效，token也就无法刷新。所以需要加一个拦截所有路径
刷新token
//拦截一切路径的请求，并刷新token的有效期，即使用户不存在也放行
public class RefreshTokenInterceptor implements HandlerInterceptor {
/**当前类RefreshTokenInterceptor没有交给spring管理，不能使用依赖注入（@Autowired）
!!!只能用构造方法方式注入属性，将来由使用者构造对象传入stringRedisTemplate实例**/
private StringRedisTemplate stringRedisTemplate;
public RefreshTokenInterceptor(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
//获取请求头中的token authorization为前段token名字
String token = request.getHeader("authorization");
if (StrUtil.isBlank(token)) {
return true;
}
//基于token获取redis中的用户 get是获取单个键值对；entries获得所有键值对
Map<Object, Object> userMap = stringRedisTemplate.opsForHash().entries("login:token" + token);
//判断用户是否存在
if(userMap.isEmpty()){
return true;
}
//将Map转为DTO对象
UserDTO userDTO = BeanUtil.fillBeanWithMap(userMap, new UserDTO(), false);
UserHolder.saveUser(userDTO);
stringRedisTemplate.expire("login:token" + token,30, TimeUnit.MINUTES);
return true;
}
登录校验
public class LoginInterceptor implements HandlerInterceptor {
/**
* 前置，用于判断用户是否登录
*/
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
// 判断当前用户是否已登录
if (ThreadLocalUtls.getUser() == null){
// 当前用户未登录，直接拦截
response.setStatus(401);
return false;
}
// 用户存在，直接放行
return true;
}
}
配置
@Configuration
public class MvcConfig implements WebMvcConfigurer {
@Resource
private StringRedisTemplate stringRedisTemplate;
@Override
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);
}
}
/**
* 根据id查询商铺数据
*/
@Override
public Result queryById(Long id) {
String key = CACHE_SHOP_KEY + id;
// 1、从Redis中查询店铺数据
String shopJson = stringRedisTemplate.opsForValue().get(key);
Shop shop = null;
if (StrUtil.isNotBlank(shopJson)) {
shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
// 2、 缓存未命中，从数据库中查询店铺数据
shop = this.getById(id);
if (Objects.isNull(shop)) {
return Result.fail("店铺不存在");
}
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop));
return Result.ok(shop);
}
缓存更新问题
常见的缓存更新策略：
内存淘汰：利用Redis的内存淘汰机制实现缓存更新，Redis的内存淘汰机制是当Redis发现内存不足时，会根据一定的策略自动淘汰部分数据
超时剔除：手动给缓存数据添加TTL，到期后Redis自动删除缓存
主动更新：手动编码实现缓存更新，在修改数据库的同时更新缓存
双写方案：人工编码方式，缓存调用者在更新完数据库后再去更新缓存。使用困难，灵活度高。
读取（Read）：当需要读取数据时，首先检查缓存是否存在该数据。如果缓存中存在，直接返回缓存中的数据。如果缓存中不存在，则从底层数据存储（如数据库）中获取数据，并将数据存储到缓存中，以便以后的读取操作可以更快地访问该数据。
写入（Write）：当进行数据写入操作时，首先更新底层数据存储中的数据。然后，根据具体情况，可以选择直接更新缓存中的数据（使缓存与底层数据存储保持同步），或者是简单地将缓存中与修改数据相关的条目标记为无效状态（缓存失效），以便下一次读取时重新加载最新数据
读写穿透方案（Read/Write Through Pattern）：将读取和写入操作首先在缓存中执行，然后再传播到数据存储
读取穿透（Read Through）：当进行读取请求时，首先检查缓存。如果所请求的数据在缓存中找到，直接返回数据。如果缓存中没有找到数据，则将请求转发给数据存储以获取数据。获取到的数据随后存储在缓存中，然后返回给调用者。
写入穿透（Write Through）：当进行写入请求时，首先将数据写入缓存。缓存立即将写操作传播到数据存储，确保缓存和数据存储之间的数据保持一致。这样保证了后续的读取请求从缓存中返回更新后的数据。
写回方案（Write Behind Caching Pattern）：调用者只操作缓存，其他线程去异步处理数据库，实现最终一致
主动更新策略中三种方案的比较：
双写方案 和 读写穿透方案 在写入数据时都会直接更新缓存，以保持缓存和底层数据存储的一致性。而 写回方案 延迟了缓存的更新操作，将数据先放入缓存队列，然后再进行批量或异步写入。
读写穿透方案 和 写回方案 相比，写回方案 具有更高的写入性能，因为它通过批量和异步操作减少了频繁的写入操作。但是 写回方案 带来了数据一致性的考虑，需要确保缓存和底层数据存储在某个时间点上保持一致，而 读写穿透方案 将数据库和缓存整合为一个服务，由服务来维护缓存与数据库的一致性，调用者无需关心数据一致性问题，降低了系统的可维护性，但是实现困难
主动更新策略中三种方案的应用场景：双写方案 较适用于读多写少的场景，数据的一致性由应用程序主动管理
读写穿透方案 适用于数据实时性要求较高、对一致性要求严格的场景
写回方案 适用于追求写入性能的场景，对数据的实时性要求相对较低、可靠性也相对低
更新策略的应用场景：对于低一致性需求，可以使用内存淘汰机制。例如店铺类型数据的查询缓存
对于高一致性需求，可以采用主动更新策略，并以超时剔除作为兜底方案。例如店铺详情数据查询的缓存
缓存主动更新策略的实现
最终实现逻辑：根据id查询店铺时，如果缓存未命中，则查询数据库，将数据库结果写入缓存，并设置超时时间。根据id修改店铺时，先修改数据库，再删除缓存
public Result queryById(Long id) {
String key = CACHE_SHOP_KEY + id;
String shopJson = stringRedisTemplate.opsForValue().get(key);
Shop shop = null;
// 判断缓存是否命中
if (StrUtil.isNotBlank(shopJson)) {
shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
// 缓存未命中，从数据库中查询店铺数据
shop = this.getById(id);
// 判断数据库是否存在店铺数据
if (Objects.isNull(shop)) {
return Result.fail("店铺不存在");
}
// 数据库中存在，重建缓存，并返回店铺数据
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop), CACHE_SHOP_TTL, TimeUnit.MINUTES);
return Result.ok(shop);
}
/**
* 更新商铺数据（更新时，更新数据库，删除缓存）
*/
@Transactional
@Override
public Result updateShop(Shop shop) {
// 参数校验, 略
// 1、更新数据库中的店铺数据
boolean f = this.updateById(shop);
if (!f){
// 缓存更新失败，抛出异常，事务回滚
throw new RuntimeException("数据库更新失败");
}
// 2、删除缓存
f = stringRedisTemplate.delete(CACHE_SHOP_KEY + shop.getId());
if (!f){
// 缓存删除失败，抛出异常，事务回滚
throw new RuntimeException("缓存删除失败");
}
return Result.ok();
}
缓存穿透 ：缓存穿透是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。
解决办法：
方案一：
/**
* 根据id查询商铺数据
*/
@Override
public Result queryById(Long id) {
String key = CACHE_SHOP_KEY + id;
String shopJson = stringRedisTemplate.opsForValue().get(key);
Shop shop = null;
// 判断缓存是否命中
if (StrUtil.isNotBlank(shopJson)) {
shop = JSONUtil.toBean(shopJson, Shop.class);
return Result.ok(shop);
}
// 缓存未命中，判断缓存中查询的数据是否是空字符串(isNotBlank把null和空字符串给排除了)
if (Objects.nonNull(shopJson)){
// 当前数据是空字符串（说明该数据是之前缓存的空对象），直接返回失败信息
return Result.fail("店铺不存在");
}
// 当前数据是null，则从数据库中查询店铺数据
shop = this.getById(id);
// 判断数据库是否存在店铺数据
if (Objects.isNull(shop)) {
// 数据库中不存在，缓存空对象（解决缓存穿透），返回失败信息
stringRedisTemplate.opsForValue().set(key, "", CACHE_NULL_TTL, TimeUnit.SECONDS);
return Result.fail("店铺不存在");
}
// 4.2 数据库中存在，重建缓存，并返回店铺数据
stringRedisTemplate.opsForValue().set(key, JSONUtil.toJsonStr(shop),
CACHE_SHOP_TTL, TimeUnit.MINUTES);
return Result.ok(shop);
}
缓存穿透的解决方案有哪些：缓存null值、布隆过滤、增强id的复杂度，避免被猜测id规律、做好数据的基础格式校验、加强用户权限校验、做好热点参数的限流。
缓存雪崩是指在同一时段大量的缓存key同时失效或者Redis服务宕机，导致大量请求到达数据库，带来巨大压力。
解决方案：
缓存击穿问题也叫热点Key问题，就是一个被高并发访问并且缓存重建业务较复杂的key突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。
常见的解决方案有两种：
互斥锁方案：
操作锁代码
核心思路就是利用redis的setnx方法来表示获取锁，该方法含义是redis中如果没有这个key，则插入成功，返回1，在stringRedisTemplate中返回true， 如果有这个key则插入失败，则返回0，在stringRedisTemplate返回false，我们可以通过true，或者是false，来表示是否有线程成功插入key，成功插入的key的线程我们认为他就是获得到锁的线程。
/获取锁函数 setIfAbsent为setnx
private boolean tryLock(String key){
//为防止程序异常，锁获取成功后迟迟没有释放，导致后续无发获取锁，给锁增加有效期
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent(key, "1", 10, TimeUnit.SECONDS);
//此处返回的是0或1，spring帮我们转换成啦对象结果，我们需要转换为基本类型，直接返回flag会自动拆箱可能导致空指针异常
//对象结果：flag为Boolean，而不是具体的boolean，而Boolean的值可以为null，此处将false和null都处理为null
return BooleanUtil.isTrue(flag);
}
//释放锁函数
private void unlock(String key){
stringRedisTemplate.delete(key);
}
业务代码：
//缓存击穿设置互斥锁的解决方案 （利用redis String类型的setnx的特性判断是否可以获取锁）
public Shop queryWithMutex(Long id){
String key = CACHE_SHOP_KEY + id;
//从redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//判断是否命中
if(StrUtil.isNotBlank(shopJson)){
//命中直接返回
Shop shop = JSONUtil.toBean(shopJson,Shop.class);
return shop;
}
//判断命中的是否空值
if(shopJson != null){
return null;
}
String lockKey = "lock:shop:" + id;
Shop shop = null;
try {
boolean islock = tryLock(lockKey);
if(!islock){
Thread.sleep(50);
return queryWithMutex(id);
}
//获取锁成功，根据id查询数据库
shop = getById(id);
//mysql中也不存在，返回错误
if(shop == null){
//写入空值
stringRedisTemplate.opsForValue().set(key,"",2,TimeUnit.MINUTES);
return null;
}
//存在，写入redis
stringRedisTemplate.opsForValue().set(key,JSONUtil.toJsonStr(shop),10, TimeUnit.MINUTES);
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
unlock(lockKey);
}
return shop;
}
逻辑过期方案：
当用户开始查询redis时，判断是否命中，如果没有命中则直接返回空数据，不查询数据库，而一旦命中后，将value取出，判断value中的过期时间是否满足，如果没有过期，则直接返回redis中的数据，如果过期，则在开启线程后直接返回之前的数据，线程去重构数据，重构完成后释放互斥锁。
//缓存击穿设置逻辑过期的解决方案
public Shop queryWithLogicalExpire(Long id){
String key = CACHE_SHOP_KEY + id;
//从redis查询缓存
String shopJson = stringRedisTemplate.opsForValue().get(key);
//判断是否命中
if(StrUtil.isBlank(shopJson)){
return null;
}
//存在，把json反序列化为对象
RedisData redisData = JSONUtil.toBean(shopJson, RedisData.class);
JSONObject data = (JSONObject) redisData.getData();//因为使用的是stringRedisTemplate，这里获得的对象为JSONObject
Shop shop = JSONUtil.toBean(data,Shop.class);
LocalDateTime expireTime = redisData.getExpireTime();
//判断是否过期
if(expireTime.isAfter(LocalDateTime.now())){
//未过期
return shop;
}
//已过期，缓存重建
//重建缓存
//获取互斥锁
String lockey = "lock:shop:" + id;
boolean isLock = tryLock(lockey);
//判断是否获取成功
if(isLock){
//TODO 注意：获取锁成功再次判断是否过期做doubleCheck，未过期则无需重建（获取的所可能是上一个线程刚刚释放的锁）
CACHE_REBUID_EXECUTOR.submit(()->{
try {
//成功，开启线程，实现重建缓存
this.saveShop2Redis(id,20L);
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
unlock(lockey);
}
});
}
//返回信息
return shop;
}
//设置逻辑过期时间函数：redisData为Shop+expiredSeconds
private void saveShop2Redis(Long id,Long expiredSeconds) {
Shop shop = getById(id);
RedisData redisData = new RedisData();
redisData.setData(shop);
redisData.setExpireTime(LocalDateTime.now().plusSeconds(expiredSeconds));
stringRedisTemplate.opsForValue().set("lock:shop:" + id,JSONUtil.toJsonStr(redisData));
}
使用数据库自增ID存在的问题：
场景分析：如果我们的id具有太明显的规则，用户或者说商业对手很容易猜测出来我们的一些敏感信息，比如商城在一天时间内，卖出了多少单，这明显不合适。
场景分析二：随着我们商城规模越来越大，mysql的单表的容量不宜超过500W，数据量过大之后，我们要进行拆库拆表，但拆分表了之后，他们从逻辑上讲他们是同一张表，所以他们的id是不能一样的， 于是乎我们需要保证id的唯一性。
全局ID生成器，是一种在分布式系统下用来生成全局唯一ID的工具，一般要满足特性：唯一性、高性能、高可用、递增行、安全性
private static final long BEGIN_TIMESTAMP = 10995200L; //自定义开始值单位：s
private static final int COUNT_BITS = 32;
private StringRedisTemplate stringRedisTemplate;
public RedisIdWorker(StringRedisTemplate stringRedisTemplate) {
this.stringRedisTemplate = stringRedisTemplate;
}
public long nextId(String keyPrefix){
//生成时间戳
LocalDateTime now = LocalDateTime.now();
long nowSecond = now.toEpochSecond(ZoneOffset.UTC);
long timeStamp = nowSecond - BEGIN_TIMESTAMP;
// 2.生成序列号
// 2.1.获取当前日期，精确到天
String date = now.format(DateTimeFormatter.ofPattern("yyyy:MM:dd"));
// 2.2.自增长 || 每一天下的单为一个key
long count = stringRedisTemplate.opsForValue().increment("icr:" + keyPrefix + ":" + date);
// 3.拼接并返回
return timeStamp << COUNT_BITS | count;
}
基于List结构
Redis的list数据结构是一个双向链表，很容易模拟出队列效果。
可以利用：LPUSH 结合 RPOP、或者 RPUSH 结合 LPOP来实现。需要注意，当队列中没有消息时RPOP或LPOP操作会返回null，并不像JVM的阻塞队列那样会阻塞并等待消息。因此这里应该使用BRPOP或者BLPOP来实现阻塞效果。
优缺点：
基于PubSub
PubSub（发布订阅）是Redis2.0版本引入的消息传递模型。消费者可以订阅一个或多个channel，生产者向对应channel发送消息后，所有订阅者都能收到相关消息。
SUBSCRIBE channel [channel] ：订阅一个或多个频道
PUBLISH channel msg ：向一个频道发送消息
PSUBSCRIBE pattern[pattern] ：订阅与pattern格式匹配的所有频道
优缺点：
优点：采用发布订阅模型，支持多生产、多消费
缺点：不支持数据持久化、无法避免消息丢失、消息堆积有上限，超出时数据丢失
基于Stream
Stream 是 Redis 5.0 引入的一种新数据类型，可以实现一个功能非常完善的消息队列。
发送消息xadd:
读取消息：
在业务代码可以使用循环使用XREAD阻塞方式来查询最新消息，从而实现持续监听队列的效果，伪代码如下
while(true){
//尝试读取队列里面的消息，最多阻塞2秒
Object msg = redis.execute("XREAD COUNT 1 BLOCK 2000 STREAMS users $")
if(msg == null){
continue;
}
//处理消息
handleMessage(msg)
}
注意：当我们指定起始ID为$时，代表读取最新的消息，如果我们处理一条消息的过程中，又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现漏读消息的问题
STREAM类型消息队列的XREAD命令特点：
消费者组
消费者组（Consumer Group）：将多个消费者划分到一个组中，监听同一个队列。具备下列特点：
创建消费者组：
XGROUP CREATE key groupName ID [MKSTREAM]
key：队列名称
groupName：消费者组名称
ID：起始ID标示，$代表队列中最后一个消息，0则代表队列中第一个消息
MKSTREAM：队列不存在时自动创建队列
其它常见命令：
删除指定的消费者组
XGROUP DESTORY key groupName
给指定的消费者组添加消费者
XGROUP CREATECONSUMER key groupname consumername
删除消费者组中的指定消费者
XGROUP DELCONSUMER key groupname consumername
从消费者组读取消息：
XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
“>”：从下一个未消费的消息开始
其它：根据指定id从pending-list中获取已消费但未确认的消息，例如0，是从pending-list中的第一个消息开始
消费者监听消息的基本思路：
STREAM类型消息队列的XREADGROUP命令特点：
最终版本：Lua脚本+Redission分布式锁+Stream消息队列
-- 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)) <= 0) then
-- 3.2.库存不足，返回1
return 1
end
-- 3.2.判断用户是否下单 SISMEMBER orderKey userId
if(redis.call('sismember', orderKey, userId) == 1) then
-- 3.3.存在，说明是重复下单，返回2
return 2
end
-- 3.4.扣库存 incrby stockKey -1
redis.call('incrby', stockKey, -1)
-- 3.5.下单（保存用户）sadd orderKey userId
redis.call('sadd', orderKey, userId)
-- 3.6.发送消息到队列中， XADD stream.orders * k1 v1 k2 v2 ...
redis.call('xadd', 'stream.orders', '*', 'userId', userId, 'voucherId', voucherId, 'id', orderId)
return 0
核心业务代码：
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
@Resource
private ISeckillVoucherService seckillVoucherService;
@Resource
private RedisIdWorker redisIdWorker;
@Resource
private StringRedisTemplate stringRedisTemplate;
@Resource
private RedissonClient redissonClient;
private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
//加载lua脚本
static {
SECKILL_SCRIPT = new DefaultRedisScript<>();
SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
SECKILL_SCRIPT.setResultType(Long.class);
}
private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor(); //线程池
private IVoucherOrderService proxy;
@PostConstruct //再类初始化后先开启异步添加订单服务，准备
private void init(){
SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
}
/**
* 订单秒杀业务最终方案
* @param voucherId
* @return
*/
@Override
public Result seckillVoucher(Long voucherId) {
Long userId = UserHolder.getUser().getId();
long orderId = redisIdWorker.nextId("order");
// 1.执行lua脚本
Long result = stringRedisTemplate.execute(
SECKILL_SCRIPT,
Collections.emptyList(),
voucherId.toString(), userId.toString(), String.valueOf(orderId)
);
int r = result.intValue();
// 2.判断结果是否为0
if (r != 0) {
// 2.1.不为0 ，代表没有购买资格
return Result.fail(r == 1 ? "库存不足" : "不能重复下单");
}
// 3.返回订单id
return Result.ok(orderId);
}
private class VoucherOrderHandler implements Runnable {
@Override
public void run() {
while (true) {
try {
// 1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS s1 >
List<MapRecord<String, Object, Object>> 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<String, Object, Object> record = list.get(0);
Map<Object, Object> 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<MapRecord<String, Object, Object>> 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<String, Object, Object> record = list.get(0);
Map<Object, Object> 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("处理pendding订单异常", e);
try{
Thread.sleep(20);
}catch(Exception e){
e.printStackTrace();
}
}
}
}
}
}
创建订单：
/将订单信息写入数据库
@Transactional
public void createVoucherOrder(VoucherOrder voucherOrder) {
Long userId = voucherOrder.getUserId();
int count = query().eq("user_id", userId).eq("voucher_id", voucherOrder.getVoucherId()).count();
if (count > 0) {
return;
}
// 6.扣减库存
boolean success = seckillVoucherService.update()
.setSql("stock = stock - 1") // set stock = stock - 1
.eq("voucher_id", voucherOrder.getVoucherId()).gt("stock", 0) // where id = ? and stock > 0
.update();
if (!success) {
// 扣减失败
return;
}
save(voucherOrder);
}
}
GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。常见的命令有：
@Override
public Result queryShopByType(Integer typeId, Integer current, Double x, Double y) {
// 1.判断是否需要根据坐标查询
if (x == null || y == null) {
// 不需要坐标查询，按数据库查询
Page<Shop> 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<RedisGeoCommands.GeoLocation<String>> 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<GeoResult<RedisGeoCommands.GeoLocation<String>>> list = results.getContent();
if (list.size() <= from) {
// 没有下一页了，结束
return Result.ok(Collections.emptyList());
}
// 4.1.截取 from ~ end的部分
List<Long> ids = new ArrayList<>(list.size());
Map<String, Distance> distanceMap = new HashMap<>(list.size());
list.stream().skip(from).forEach(result -> {
// 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<Shop> 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);
}
用户签到完全可以单独创建一个mysql表存储用户每天签到信息，不过我们可以使用BitMap简化签到记录方式。
我们按月来统计用户签到信息，签到记录为1，未签到则记录为0。把每一个bit位对应当月的每一天，形成了映射关系。用0和1标示业务状态，这种思路就称为位图（BitMap）。这样我们就用极小的空间，来实现了大量数据的表示
Redis中是利用string类型数据结构实现BitMap，因此最大上限是512M，转换为bit则是 2^32个bit位。
BitMap的操作命令有：
签到：
@Override
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();
}
签到统计：
@Override
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<Long> 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);
}
@Override
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<Follow>()
.eq("user_id", userId).eq("follow_user_id", followUserId));
if (isSuccess) {
// 把关注用户的id从Redis集合中移除
stringRedisTemplate.opsForSet().remove(key, followUserId.toString());
}
}
return Result.ok();
}
好友关注-共同关注
@Override
public Result followCommons(Long id) {
// 1.获取当前用户
Long userId = UserHolder.getUser().getId();
String key = "follows:" + userId;
// 2.求交集
String key2 = "follows:" + id;
Set<String> intersect = stringRedisTemplate.opsForSet().intersect(key, key2);
if (intersect == null || intersect.isEmpty()) {
// 无交集
return Result.ok(Collections.emptyList());
}
// 3.解析id集合
List<Long> ids = intersect.stream().map(Long::valueOf).collect(Collectors.toList());
// 4.查询用户
List<UserDTO> users = userService.listByIds(ids)
.stream()
.map(user -> BeanUtil.copyProperties(user, UserDTO.class))
.collect(Collectors.toList());
return Result.ok(users);
}
Fead流：当我们关注了用户后，这个用户发了动态，把这些数据推送给用户，这种消息推送的方式叫做Feed流，关注推送也叫做Feed流，直译为投喂。
Feed流产品有两种常见模式：
我们本次针对好友的操作，采用的就是Timeline的方式，只需要拿到我们关注用户的信息，然后按照时间排序即可
，因此采用Timeline的模式。该模式的实现方案有三种：
@Override
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<Follow> 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());
}
我们需要记录每次操作的最后一条，然后从这个位置开始去读取数据
举个例子：我们从t1时刻开始，拿第一页数据，拿到了10~6，然后记录下当前最后一次拿取的记录，就是6，t2时刻发布了新的记录，此时这个11放到最顶上，但是不会影响我们之前记录的6，此时t3时刻来拿第二页，第二页这个时候拿数据，还是从6后一点的5去拿，就拿到了5-1的记录。我们这个地方可以采用sortedSet来做，可以进行范围查询，并且还可以记录当前获取数据时间戳最小值，就可以实现滚动分页了
一、定义出来具体的返回值实体类
@Data
public class ScrollResult {
private List<?> list;
private Long minTime;
private Integer offset;
}
BlogController
@GetMapping("/of/follow")
public Result queryBlogOfFollow(
@RequestParam("lastId") Long max, @RequestParam(value = "offset", defaultValue = "0") Integer offset){
return blogService.queryBlogOfFollow(max, offset);
}
BlogServiceImpl
@Override
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<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet()
.reverseRangeByScoreWithScores(key, 0, max, offset, 2);
// 3.非空判断
if (typedTuples == null || typedTuples.isEmpty()) {
return Result.ok();
}
// 4.解析数据：blogId、minTime（时间戳）、offset
List<Long> ids = new ArrayList<>(typedTuples.size());
long minTime = 0; // 2
int os = 1; // 2
for (ZSetOperations.TypedTuple<String> 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;
}
}
os = minTime == max ? os : os + offset;
// 5.根据id查询blog
String idStr = StrUtil.join(",", ids);
List<Blog> 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，已点赞过则点赞数-1，且需要排序。综合考虑使用SortedSet实现
ZSCORE方法判断用户是否存在ZRANGE方法实现范围查询//点赞
@Override
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的sortedSet集合 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的sortedSet集合移除
if (isSuccess) {
stringRedisTemplate.opsForZSet().remove(key, userId.toString());
}
}
return Result.ok();
}
private void isBlogLiked(Blog blog) {
// 1.获取登录用户
UserDTO user = UserHolder.getUser();
if (user == null) {
// 用户未登录，无需查询是否点赞
return;
}
Long userId = user.getId();
// 2.判断当前登录用户是否已经点赞
String key = "blog:liked:" + blog.getId();
Double score = stringRedisTemplate.opsForZSet().score(key, userId.toString());
blog.setIsLike(score != null);
}
//点赞排行查询
@Override
public Result queryBlogLikes(Long id) {
String key = BLOG_LIKED_KEY + id;
// 1.查询top5的点赞用户 zrange key 0 4
Set<String> top5 = stringRedisTemplate.opsForZSet().range(key, 0, 4);
if (top5 == null || top5.isEmpty()) {
return Result.ok(Collections.emptyList());
}
// 2.解析出其中的用户id
List<Long> 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<UserDTO> 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);
}
首先我们搞懂两个概念：
通常来说UV会比PV大很多，所以衡量同一个网站的访问量，我们需要综合考虑很多因素，所以我们只是单纯的把这两个值作为一个参考值
UV统计在服务端做会比较麻烦，因为要判断该用户是否已经统计过了，需要将统计过的用户信息保存。但是如果每个访问的用户都保存到Redis中，数据量会非常恐怖，那怎么处理呢？
Hyperloglog(HLL)是从Loglog算法派生的概率算法，用于确定非常大的集合的基数，而不需要存储其所有值。相关算法原理大家可以参考：https://juejin.cn/post/6844903785744056333#heading-0
Redis中的HLL是基于string结构实现的，单个HLL的内存永远小于16kb，内存占用低的令人发指！作为代价，其测量结果是概率性的，有小于0.81％的误差。不过对于UV统计来说，这完全可以忽略。
/**
* 测试 HyperLogLog 实现 UV 统计的误差
*/
@Test
public void testHyperLogLog() {
String[] values = new String[1000];
// 批量保存100w条用户记录，每一批1个记录
int j = 0;
for (int i = 0; i < 1000000; i++) {
j = i % 1000;
values[j] = "user_" + i;
if (j == 999) {
// 发送到Redis
stringRedisTemplate.opsForHyperLogLog().add("hl2", values);
}
}
// 统计数量
Long count = stringRedisTemplate.opsForHyperLogLog().size("hl2");
System.out.println("count = " + count);
}
因篇幅问题不能全部显示，请点此查看更多更全内容
Copyright © 2019- huatuo9.cn 版权所有 赣ICP备2023008801号-1
违法及侵权请联系：TEL:199 18 7713 E-MAIL:2724546146@qq.com
本站由北京市万商天勤律师事务所王兴未律师提供法律服务