mirror of
https://gitee.com/hhyykk/ipms-sjy.git
synced 2025-07-24 07:55:06 +08:00
Merge branch 'master' of https://gitee.com/zhijiantianya/ruoyi-vue-pro into feature/mall_product
This commit is contained in:
@ -36,7 +36,7 @@ import java.util.Set;
|
||||
* 注意,使用 DeptDataPermissionRule 时,需要保证表中有 dept_id 部门编号的字段,可自定义。
|
||||
*
|
||||
* 实际业务场景下,会存在一个经典的问题?当用户修改部门时,冗余的 dept_id 是否需要修改?
|
||||
* 1. 一般情况下,dept_id 不进行修改,则会导致用户看到之前的数据。【yudao-server 采用该方案】
|
||||
* 1. 一般情况下,dept_id 不进行修改,则会导致用户看不到之前的数据。【yudao-server 采用该方案】
|
||||
* 2. 部分情况下,希望该用户还是能看到之前的数据,则有两种方式解决:【需要你改造该 DeptDataPermissionRule 的实现代码】
|
||||
* 1)编写洗数据的脚本,将 dept_id 修改成新部门的编号;【建议】
|
||||
* 最终过滤条件是 WHERE dept_id = ?
|
||||
|
@ -3,11 +3,15 @@ package cn.iocoder.yudao.framework.captcha.config;
|
||||
import cn.hutool.core.util.ClassUtil;
|
||||
import cn.iocoder.yudao.framework.captcha.core.enums.CaptchaRedisKeyConstants;
|
||||
import cn.iocoder.yudao.framework.captcha.core.service.RedisCaptchaServiceImpl;
|
||||
import com.xingyuv.captcha.properties.AjCaptchaProperties;
|
||||
import com.xingyuv.captcha.service.CaptchaCacheService;
|
||||
import com.xingyuv.captcha.service.impl.CaptchaServiceFactory;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
@AutoConfiguration
|
||||
public class YudaoCaptchaConfiguration {
|
||||
|
||||
@ -17,9 +21,17 @@ public class YudaoCaptchaConfiguration {
|
||||
ClassUtil.loadClass(CaptchaRedisKeyConstants.class.getName());
|
||||
}
|
||||
|
||||
@Resource
|
||||
private StringRedisTemplate stringRedisTemplate;
|
||||
|
||||
@Bean
|
||||
public CaptchaCacheService captchaCacheService(StringRedisTemplate stringRedisTemplate) {
|
||||
return new RedisCaptchaServiceImpl(stringRedisTemplate);
|
||||
public CaptchaCacheService captchaCacheService(AjCaptchaProperties config) {
|
||||
// 缓存类型 redis/local/....
|
||||
CaptchaCacheService ret = CaptchaServiceFactory.getCache(config.getCacheType().name());
|
||||
if (ret instanceof RedisCaptchaServiceImpl) {
|
||||
((RedisCaptchaServiceImpl) ret).setStringRedisTemplate(stringRedisTemplate);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
}
|
||||
|
@ -25,6 +25,10 @@ public class RedisCaptchaServiceImpl implements CaptchaCacheService {
|
||||
return "redis";
|
||||
}
|
||||
|
||||
public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) {
|
||||
this.stringRedisTemplate = stringRedisTemplate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void set(String key, String value, long expiresInSeconds) {
|
||||
stringRedisTemplate.opsForValue().set(key, value, expiresInSeconds, TimeUnit.SECONDS);
|
||||
|
@ -7,17 +7,14 @@ import lombok.AllArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.redisson.api.RLock;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.springframework.data.redis.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
import org.springframework.data.domain.Range;
|
||||
import org.springframework.data.redis.connection.stream.*;
|
||||
import org.springframework.data.redis.core.StreamOperations;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* 这个任务用于处理,crash 之后的消费者未消费完的消息
|
||||
@ -28,6 +25,14 @@ public class RedisPendingMessageResendJob {
|
||||
|
||||
private static final String LOCK_KEY = "redis:pending:msg:lock";
|
||||
|
||||
/**
|
||||
* 消息超时时间,默认 5 分钟
|
||||
*
|
||||
* 1. 超时的消息才会被重新投递
|
||||
* 2. 由于定时任务 1 分钟一次,消息超时后不会被立即重投,极端情况下消息5分钟过期后,再等 1 分钟才会被扫瞄到
|
||||
*/
|
||||
private static final int EXPIRE_TIME = 5 * 60;
|
||||
|
||||
private final List<AbstractStreamMessageListener<?>> listeners;
|
||||
private final RedisMQTemplate redisTemplate;
|
||||
private final String groupName;
|
||||
@ -51,29 +56,44 @@ public class RedisPendingMessageResendJob {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 执行清理逻辑
|
||||
*
|
||||
* @see <a href="https://gitee.com/zhijiantianya/ruoyi-vue-pro/pulls/480/files">讨论</a>
|
||||
*/
|
||||
private void execute() {
|
||||
StreamOperations<String, Object, Object> ops = redisTemplate.getRedisTemplate().opsForStream();
|
||||
listeners.forEach(listener -> {
|
||||
PendingMessagesSummary pendingMessagesSummary = ops.pending(listener.getStreamKey(), groupName);
|
||||
PendingMessagesSummary pendingMessagesSummary = Objects.requireNonNull(ops.pending(listener.getStreamKey(), groupName));
|
||||
// 每个消费者的 pending 队列消息数量
|
||||
Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
|
||||
pendingMessagesPerConsumer.forEach((consumerName, pendingMessageCount) -> {
|
||||
log.info("[processPendingMessage][消费者({}) 消息数量({})]", consumerName, pendingMessageCount);
|
||||
|
||||
// 从消费者的 pending 队列中读取消息
|
||||
List<MapRecord<String, Object, Object>> records = ops.read(Consumer.from(groupName, consumerName), StreamOffset.create(listener.getStreamKey(), ReadOffset.from("0")));
|
||||
if (CollUtil.isEmpty(records)) {
|
||||
// 每个消费者的 pending消息的详情信息
|
||||
PendingMessages pendingMessages = ops.pending(listener.getStreamKey(), Consumer.from(groupName, consumerName), Range.unbounded(), pendingMessageCount);
|
||||
if (pendingMessages.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
for (MapRecord<String, Object, Object> record : records) {
|
||||
pendingMessages.forEach(pendingMessage -> {
|
||||
// 获取消息上一次传递到 consumer 的时间,
|
||||
long lastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery().getSeconds();
|
||||
if (lastDelivery < EXPIRE_TIME){
|
||||
return;
|
||||
}
|
||||
// 获取指定 id 的消息体
|
||||
List<MapRecord<String, Object, Object>> records = ops.range(listener.getStreamKey(),
|
||||
Range.of(Range.Bound.inclusive(pendingMessage.getIdAsString()), Range.Bound.inclusive(pendingMessage.getIdAsString())));
|
||||
if (CollUtil.isEmpty(records)) {
|
||||
return;
|
||||
}
|
||||
// 重新投递消息
|
||||
redisTemplate.getRedisTemplate().opsForStream().add(StreamRecords.newRecord()
|
||||
.ofObject(record.getValue()) // 设置内容
|
||||
.ofObject(records.get(0).getValue()) // 设置内容
|
||||
.withStreamKey(listener.getStreamKey()));
|
||||
|
||||
// ack 消息消费完成
|
||||
redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, record);
|
||||
}
|
||||
redisTemplate.getRedisTemplate().opsForStream().acknowledge(groupName, records.get(0));
|
||||
log.info("[processPendingMessage][消息({})重新投递成功]", records.get(0).getId());
|
||||
});
|
||||
});
|
||||
});
|
||||
}
|
||||
|
Reference in New Issue
Block a user