discuz做网站赚钱经历,wordpress 漏洞复现,wordpress前台显示作者角色,dw个人主页模板1. 使用redisson完成延迟队列的功能
引入依赖
spring-boot-starter-actuator是Spring Boot提供的一个用于监控和管理应用程序的模块 用于查看应用程序的健康状况、审计信息、指标和其他有用的信息。这些端点可以帮助你监控应用程序的运行状态、性能指标和健康状况。 已经有了…1. 使用redisson完成延迟队列的功能
引入依赖
spring-boot-starter-actuator是Spring Boot提供的一个用于监控和管理应用程序的模块 用于查看应用程序的健康状况、审计信息、指标和其他有用的信息。这些端点可以帮助你监控应用程序的运行状态、性能指标和健康状况。 已经有了其他的监控和管理工具不需要使用Spring Boot Actuator提供的功能。
!-- redisson --
dependencygroupIdorg.redisson/groupIdartifactIdredisson-spring-boot-starter/artifactIdexclusionsexclusiongroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-actuator/artifactId/exclusion/exclusions
/dependency1.1 延时队列工具类
添加延迟队列时使用监测扫描时也会用这个工具类进行获取消息
package cn.creatoo.common.redis.queue;import cn.creatoo.common.core.utils.StringUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;/*** 分布式延时队列工具类* author*/
Component
ConditionalOnBean({RedissonClient.class})
public class RedisDelayQueueUtil {private static final Logger log LoggerFactory.getLogger(RedisDelayQueueUtil.class);Resourceprivate RedissonClient redissonClient;/*** 添加延迟队列** param value 队列值* param delay 延迟时间* param timeUnit 时间单位* param queueCode 队列键* param T*/public T boolean addDelayQueue(NonNull T value, NonNull long delay, NonNull TimeUnit timeUnit, NonNull String queueCode) {if (StringUtils.isBlank(queueCode) || Objects.isNull(value)) {return false;}try {RBlockingDequeObject blockingDeque redissonClient.getBlockingDeque(queueCode);RDelayedQueueObject delayedQueue redissonClient.getDelayedQueue(blockingDeque);delayedQueue.offer(value, delay, timeUnit);//delayedQueue.destroy();log.info((添加延时队列成功) 队列键{}队列值{}延迟时间{}, queueCode, value, timeUnit.toSeconds(delay) 秒);} catch (Exception e) {log.error((添加延时队列失败) {}, e.getMessage());throw new RuntimeException((添加延时队列失败));}return true;}/*** 获取延迟队列** param queueCode* param T*/public T T getDelayQueue(NonNull String queueCode) throws InterruptedException {if (StringUtils.isBlank(queueCode)) {return null;}RBlockingDequeMap blockingDeque redissonClient.getBlockingDeque(queueCode);RDelayedQueueMap delayedQueue redissonClient.getDelayedQueue(blockingDeque);T value (T) blockingDeque.poll();return value;}/*** 删除指定队列中的消息** param o 指定删除的消息对象队列值(同队列需保证唯一性)* param queueCode 指定队列键*/public boolean removeDelayedQueue(NonNull Object o, NonNull String queueCode) {if (StringUtils.isBlank(queueCode) || Objects.isNull(o)) {return false;}RBlockingDequeObject blockingDeque redissonClient.getBlockingDeque(queueCode);RDelayedQueueObject delayedQueue redissonClient.getDelayedQueue(blockingDeque);boolean flag delayedQueue.remove(o);//delayedQueue.destroy();return flag;}
}
1.2 延迟队列执行器
package cn.creatoo.system.handler;/*** 延迟队列执行器*/
public interface RedisDelayQueueHandleT {void execute(T t);}1.3 实现队列执行器
实现队列执行器接口在这里写延迟要做的业务逻辑
package cn.creatoo.system.handler.impl;import cn.creatoo.common.core.domain.vo.WaterVo;
import cn.creatoo.system.api.RemoteFileService;
import cn.creatoo.system.handler.RedisDelayQueueHandle;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.Map;Component(exposeLinkCloudDelay)
public class ExposeLinkCloudDelay implements RedisDelayQueueHandleMap {Autowiredprivate RemoteFileService remoteFileService;Overridepublic void execute(Map map) {long dataId Long.parseLong(map.get(dataId).toString());WaterVo waterVo new WaterVo();waterVo.setFileLink(map.get(fileLink).toString());waterVo.setType(Integer.parseInt(map.get(type).toString()));waterVo.setDataId(dataId);remoteFileService.waterLink(waterVo);}
}1.4 延迟队列业务枚举类
package cn.creatoo.common.core.enums;import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;/*** 延迟队列业务枚举类* author shang tf* data 2024/3/21 14:52*/
Getter
NoArgsConstructor
AllArgsConstructor
public enum FileRedisDelayQueueEnum {EXPOSE_LINK_DELAY(EXPOSE_LINK_DELAY,资源链接处理,exposeLinkDelay),EXPOSE_LINK_CLOUD_DELAY(EXPOSE_LINK_CLOUD_DELAY,资源链接处理,exposeLinkCloudDelay),COMPRESSED_LINK_DELAY(COMPRESSED_LINK_DELAY,文件压缩处理,compressedLinkDelay),UPLOAD_TO_CLOUD_DELAY(UPLOAD_TO_CLOUD_DELAY,资源上传消费端,uploadToCloudDelay),GET_HASHCODE_DELAY(GET_HASHCODE_DELAY,资源hash值获取,getHashcodeDelay),UPLOAD_FILE_TO_CABINET(UPLOAD_FILE_CABINET,异步添加文件到数据柜,uploadFileCabinet);/*** 延迟队列 Redis Key*/private String code;/*** 中文描述*/private String name;/*** 延迟队列具体业务实现的 Bean* 可通过 Spring 的上下文获取*/private String beanId;
}
1.5 启动延迟队列监测扫描
package cn.creatoo.system.handler.impl;import cn.creatoo.common.core.enums.FileRedisDelayQueueEnum;
import cn.creatoo.common.redis.queue.RedisDelayQueueUtil;
import cn.creatoo.system.handler.RedisDelayQueueHandle;
import com.alibaba.fastjson2.JSON;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** author shang tf* data 2024/3/14 10:45* 启动延迟队列监测扫描* 文件处理的延迟队列线程池*/
Slf4j
Component
public class FileRedisDelayQueueRunner implements CommandLineRunner {Autowiredprivate RedisDelayQueueUtil redisDelayQueueUtil;Autowiredprivate ApplicationContext context;Autowiredprivate ThreadPoolTaskExecutor ptask;Value(${file-thread-pool.core-pool-size:1})private int corePoolSize;Value(${file-thread-pool.maximum-pool-size:1})private int maximumPoolSize;private ThreadPoolExecutor executorService;/*** 程序加载配置文件后延迟创建线程池*/PostConstructpublic void init() {executorService new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 30, TimeUnit.SECONDS,new LinkedBlockingQueueRunnable(1000), new ThreadFactoryBuilder().setNameFormat(delay-queue-%d).build());}Overridepublic void run(String... args) {ptask.execute(() - {while (true) {try {FileRedisDelayQueueEnum[] queueEnums FileRedisDelayQueueEnum.values();for (FileRedisDelayQueueEnum queueEnum : queueEnums) {Object value redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());if (value ! null) {System.out.println(----------------value: JSON.toJSONString(value));RedisDelayQueueHandleObject redisDelayQueueHandle (RedisDelayQueueHandleObject) context.getBean(queueEnum.getBeanId());executorService.execute(() - {redisDelayQueueHandle.execute(value);});}}TimeUnit.MILLISECONDS.sleep(500);} catch (InterruptedException e) {log.error((FileRedission延迟队列监测异常中断) {}, e.getMessage());}}});log.info((FileRedission延迟队列监测启动成功));}
}1.6 使用延迟队列
使用时在需要延时的地方。 通过注入RedisDelayQueueUtil使用addDelayQueue方法进行添加延迟任务。
MapString, String map new HashMap();
map.put(dataId, examineVo.getId().toString());
map.put(fileLink, resourceLink);
map.put(type, resourceType.toString());
map.put(remark, 资源链接处理);
// 5秒后执行exposeLinkCloudDelay中的方法
redisDelayQueueUtil.addDelayQueue(map, 5, TimeUnit.SECONDS, FileRedisDelayQueueEnum.EXPOSE_LINK_CLOUD_DELAY.getCode());2. 使用redisson完成不延时队列的功能
2.1 分布式队列工具类
package cn.creatoo.common.redis.queue;import cn.creatoo.common.core.utils.StringUtils;
import org.redisson.api.RBoundedBlockingQueue;
import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.Objects;/*** 分布式队列工具类*/
Component
ConditionalOnBean({RedissonClient.class})
public class RedisBlockQueueUtil {private static final Logger log LoggerFactory.getLogger(RedisBlockQueueUtil.class);Resourceprivate RedissonClient redissonClient;//public T boolean addQueue(NonNull T value, NonNull String queueCode) {if (StringUtils.isBlank(queueCode) || Objects.isNull(value)) {return false;}try {RBoundedBlockingQueueT queue redissonClient.getBoundedBlockingQueue(queueCode);queue.trySetCapacity(10000);queue.put(value);} catch (Exception e) {throw new RuntimeException((添加redisson队列失败));}return true;}/*** 获取队列* param queueCode* param T*/public T T getQueuePeek(NonNull String queueCode) throws InterruptedException {if (StringUtils.isBlank(queueCode)) {return null;}RQueueT queue redissonClient.getBoundedBlockingQueue(queueCode);T obj (T) queue.peek();return obj;}public T T getQueueTake(NonNull String queueCode) throws InterruptedException {if (StringUtils.isBlank(queueCode)) {return null;}RBoundedBlockingQueueT queue redissonClient.getBoundedBlockingQueue(queueCode);T obj (T) queue.take();return obj;}}2.2 队列业务枚举
package cn.creatoo.common.core.enums;import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;/*** 队列业务枚举*/
Getter
NoArgsConstructor
AllArgsConstructor
public enum RedisQueueEnum {FLOW_RECORD(redissionQueue:FLOW_RECORD, 流量流水),USER_LOGIN_RECORD(redissionQueue:USER_LOGIN_RECORD, 用户登录流水),USER_REGISTER_RECORD(redissionQueue:USER_REGISTER_RECORD, 用户注册流水),SMS_SEND_RECORD(redissionQueue:SMS_SEND_RECORD, 短信流水);/*** 队列 Redis Key*/private String code;/*** 中文描述*/private String name;}2.3 启动队列监测扫描
package cn.creatoo.system.handler.impl;import cn.creatoo.common.core.enums.RedisQueueEnum;
import cn.creatoo.common.core.utils.StringUtils;
import cn.creatoo.common.mongodb.model.FlowStatistics;
import cn.creatoo.common.mongodb.model.MessageSendRecord;
import cn.creatoo.common.mongodb.model.UserLogin;
import cn.creatoo.common.mongodb.model.UserRegister;
import cn.creatoo.common.redis.queue.RedisBlockQueueUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** description: 启动队列监测扫描*/
Slf4j
Component
public class RedisQueueRunner implements CommandLineRunner {Autowiredprivate RedisBlockQueueUtil redisBlockQueueUtil;//Autowired//private IBdStatcountService bdStatcountService;Autowiredprivate ThreadPoolTaskExecutor ptask;Resourceprivate MongoTemplate mongoTemplate;//Autowired//private BdAdminHomeService bdAdminHomeService;Value(${prodHost.mall})private String mallHost;ThreadPoolExecutor executorService new ThreadPoolExecutor(4, 8, 30, TimeUnit.SECONDS,new LinkedBlockingQueueRunnable(1000),new ThreadFactoryBuilder().setNameFormat(queue-%d).build());Overridepublic void run(String... args) throws Exception {ptask.execute(() - {while (true){try {RedisQueueEnum[] queueEnums RedisQueueEnum.values();for (RedisQueueEnum queueEnum : queueEnums) {Object value redisBlockQueueUtil.getQueuePeek(queueEnum.getCode());if (value ! null) {executorService.execute(() - {try {//System.out.println(value.toString());if(queueEnum.getCode().equals(RedisQueueEnum.FLOW_RECORD.getCode())){FlowStatistics flowStatistics redisBlockQueueUtil.getQueueTake(queueEnum.getCode());/* if(flowStatistics!null StringUtils.isNotBlank(flowStatistics.getUrl())){mongoTemplate.insert(flowStatistics, pv_ new SimpleDateFormat(yyyy).format(new Date()));// 添加首页统计缓存bdAdminHomeService.addDetailCache(flowStatistics);if(StringUtils.isNotBlank(flowStatistics.getUrl())){bdStatcountService.browseByUrl(flowStatistics.getUrl());}}*/} else if (queueEnum.getCode().equals(RedisQueueEnum.USER_LOGIN_RECORD.getCode())) {UserLogin userLogin redisBlockQueueUtil.getQueueTake(queueEnum.getCode());mongoTemplate.insert(userLogin, user_login_ new SimpleDateFormat(yyyy).format(new Date()));} else if (queueEnum.getCode().equals(RedisQueueEnum.USER_REGISTER_RECORD.getCode())) {UserRegister userRegister redisBlockQueueUtil.getQueueTake(queueEnum.getCode());mongoTemplate.insert(userRegister, user_register);} else if (queueEnum.getCode().equals(RedisQueueEnum.SMS_SEND_RECORD.getCode())) {MessageSendRecord sendRecord redisBlockQueueUtil.getQueueTake(queueEnum.getCode());mongoTemplate.insert(sendRecord, sms_send_ new SimpleDateFormat(yyyy).format(new Date()));}} catch (InterruptedException e) {log.error((Redission队列监测异常中断) {}, e.getMessage());}});}}TimeUnit.MILLISECONDS.sleep(500);} catch (InterruptedException e) {log.error((Redission队列监测异常中断) {}, e.getMessage());}}});log.info((Redission队列监测启动成功));}
}2.4 使用
这个是直接执行没有延迟的功能 redisBlockQueueUtil.addQueue(userRegister, RedisQueueEnum.USER_REGISTER_RECORD.getCode());
