Merge branch 'master' into normalize

austin-common/src/main/java/com/java3y/austin/common/constant/CommonConstant.java

@@ -35,7 +35,7 @@ public class CommonConstant {
     /**
      * 编码格式
      */
-    public static final String CHARSET_NAME = "UTF-8";
+    public static final String CHARSET_UTF_8 = "UTF-8";
     /**
      * HTTP请求内容格式
      */
@@ -44,6 +44,12 @@ public class CommonConstant {
     public static final String CONTENT_TYPE_XML = "application/xml; charset=UTF-8";
     public static final String CONTENT_TYPE_FORM_URL_ENCODE = "application/x-www-form-urlencoded;charset=utf-8;";
     public static final String CONTENT_TYPE_MULTIPART_FORM_DATA = "multipart/form-data";
+    /**
+     * 协议
+     */
+    public static final String HTTP = "http";
+    public static final String HTTPS = "https";
+    public static final String OSS = "oss";
     /**
      * HTTP 请求方法
      */

austin-common/src/main/java/com/java3y/austin/common/dto/account/sms/LinTongSmsAccount.java

@@ -2,6 +2,8 @@ package com.java3y.austin.common.dto.account.sms;
 
 import lombok.*;
 
+import java.util.Objects;
+
 /**
  * <span>Form File</span>
  * <p>Description</p>
@@ -13,7 +15,6 @@ import lombok.*;
  * @Description
  * @see com.java3y.austin.common.dto.account austin
  */
-@EqualsAndHashCode(callSuper = true)
 @Data
 @Builder
 @AllArgsConstructor
@@ -30,4 +31,36 @@ public class LinTongSmsAccount extends SmsAccount {
     private String userName;
     private String password;
 
+    /**
+     * 重写equals方法
+     *
+     * @param o
+     * @return
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        LinTongSmsAccount that = (LinTongSmsAccount) o;
+        return url.equals(that.url) &&
+                userName.equals(that.userName) &&
+                password.equals(that.password);
+    }
+
+    /**
+     * 重写hashCode方法
+     *
+     * @return
+     */
+    @Override
+    public int hashCode() {
+        return Objects.hash(url, userName, password);
+    }
 }

austin-common/src/main/java/com/java3y/austin/common/dto/account/sms/SmsAccount.java

@@ -4,6 +4,8 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.Objects;
+
 /**
  * @author 3y
  */
@@ -27,5 +29,33 @@ public class SmsAccount {
      */
     protected String scriptName;
 
+    /**
+     * 重写equals方法
+     *
+     * @param o
+     * @return
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        SmsAccount that = (SmsAccount) o;
+        return supplierId.equals(that.supplierId) &&
+                supplierName.equals(that.supplierName) &&
+                scriptName.equals(that.scriptName);
+    }
 
+    /**
+     * 重写hashCode方法
+     *
+     * @return
+     */
+    @Override
+    public int hashCode() {
+        return Objects.hash(supplierId, supplierName, scriptName);
+    }
 }

austin-common/src/main/java/com/java3y/austin/common/dto/account/sms/TencentSmsAccount.java

@@ -4,6 +4,8 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.Objects;
+
 /**
  * 腾讯短信参数
  * <p>
@@ -42,4 +44,41 @@ public class TencentSmsAccount extends SmsAccount {
     private String smsSdkAppId;
     private String templateId;
     private String signName;
+
+    /**
+     * 重写equals方法
+     *
+     * @param o
+     * @return
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        TencentSmsAccount that = (TencentSmsAccount) o;
+        return url.equals(that.url) &&
+                region.equals(that.region) &&
+                secretId.equals(that.secretId) &&
+                secretKey.equals(that.secretKey) &&
+                smsSdkAppId.equals(that.smsSdkAppId) &&
+                templateId.equals(that.templateId) &&
+                signName.equals(that.signName);
+    }
+
+    /**
+     * 重写hashCode方法
+     *
+     * @return
+     */
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), url, region, secretId, secretKey, smsSdkAppId, templateId, signName);
+    }
 }

austin-common/src/main/java/com/java3y/austin/common/dto/account/sms/YunPianSmsAccount.java

@@ -4,6 +4,8 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.util.Objects;
+
 /**
  * 云片账号信息
  * <p>
@@ -31,4 +33,36 @@ public class YunPianSmsAccount extends SmsAccount {
      */
     private String url;
 
+    /**
+     * 重写equals方法
+     *
+     * @param o
+     * @return
+     */
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        if (!super.equals(o)) {
+            return false;
+        }
+        YunPianSmsAccount that = (YunPianSmsAccount) o;
+        return apikey.equals(that.apikey) &&
+                tplId.equals(that.tplId) &&
+                url.equals(that.url);
+    }
+
+    /**
+     * 重写hashCode方法
+     *
+     * @return
+     */
+    @Override
+    public int hashCode() {
+        return Objects.hash(super.hashCode(), apikey, tplId, url);
+    }
 }

austin-common/src/main/java/com/java3y/austin/common/vo/BasicResultVO.java

@@ -6,6 +6,8 @@ import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.ToString;
 
+import java.io.Serializable;
+
 /**
  * @author zzb
  * @since 2021.11.17
@@ -15,7 +17,7 @@ import lombok.ToString;
 @ToString(callSuper = true)
 @AllArgsConstructor
 @NoArgsConstructor
-public final class BasicResultVO<T> {
+public final class BasicResultVO<T> implements Serializable {
 
     /**
      * 响应状态

austin-cron/src/main/java/com/java3y/austin/cron/utils/ReadFileUtils.java

@@ -4,12 +4,16 @@ import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.text.csv.*;
+import cn.hutool.core.util.CharsetUtil;
 import com.google.common.base.Throwables;
 import com.java3y.austin.cron.csv.CountFileRowHandler;
 import com.java3y.austin.cron.vo.CrowdInfoVo;
 import lombok.extern.slf4j.Slf4j;
 
-import java.io.FileReader;
+import java.io.FileInputStream;
+import java.io.InputStreamReader;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.*;
 
 /**
@@ -37,7 +41,8 @@ public class ReadFileUtils {
     public static void getCsvRow(String path, CsvRowHandler csvRowHandler) {
 
         // 把首行当做是标题，获取reader
-        try (CsvReader reader = CsvUtil.getReader(new FileReader(path),
+        try (CsvReader reader = CsvUtil.getReader(
+                new InputStreamReader(Files.newInputStream(Paths.get(path)), CharsetUtil.CHARSET_UTF_8),
                 new CsvReadConfig().setContainsHeader(true))) {
             reader.read(csvRowHandler);
         } catch (Exception e) {
@@ -54,7 +59,8 @@ public class ReadFileUtils {
     public static long countCsvRow(String path, CountFileRowHandler countFileRowHandler) {
 
         // 把首行当做是标题，获取reader
-        try (CsvReader reader = CsvUtil.getReader(new FileReader(path),
+        try (CsvReader reader = CsvUtil.getReader(
+                new InputStreamReader(new FileInputStream(path), CharsetUtil.CHARSET_UTF_8),
                 new CsvReadConfig().setContainsHeader(true))) {
 
             reader.read(countFileRowHandler);

austin-cron/src/main/java/com/java3y/austin/cron/xxl/service/impl/CronTaskServiceImpl.java

@@ -164,10 +164,12 @@ public class CronTaskServiceImpl implements CronTaskService {
             }
         } catch (Exception e) {
             log.error("CronTaskService#getGroupId fail,e:{},param:{},response:{}", Throwables.getStackTraceAsString(e)
-                    , JSON.toJSONString(params), JSON.toJSONString(response.body()));
+                    , JSON.toJSONString(params),
+                    response != null ? JSON.toJSONString(response.body()) : "");
         }
         invalidateCookie();
-        return BasicResultVO.fail(RespStatusEnum.SERVICE_ERROR, JSON.toJSONString(response.body()));
+        return BasicResultVO.fail(RespStatusEnum.SERVICE_ERROR,
+                response != null ? JSON.toJSONString(response.body()) : "");
     }
 
     @Override

austin-cron/src/main/java/com/java3y/austin/cron/xxl/utils/XxlJobUtils.java

@@ -84,7 +84,7 @@ public class XxlJobUtils {
         if (Objects.isNull(basicResultVO.getData())) {
             XxlJobGroup xxlJobGroup = XxlJobGroup.builder().appname(appName).title(jobHandlerName).addressType(CommonConstant.FALSE).build();
             if (RespStatusEnum.SUCCESS.getCode().equals(cronTaskService.createGroup(xxlJobGroup).getStatus())) {
-                return (int) cronTaskService.getGroupId(appName, jobHandlerName).getData();
+                return (Integer) cronTaskService.getGroupId(appName, jobHandlerName).getData();
             }
         }
         return basicResultVO.getData();

austin-handler/src/main/java/com/java3y/austin/handler/domain/sms/LinTongSendResult.java

@@ -32,7 +32,7 @@ public class LinTongSendResult {
     @Data
     @AllArgsConstructor
     @NoArgsConstructor
-    public class DataDTO {
+    public static class DataDTO {
         Integer code;
         String message;
         Long msgId;

austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/DingDingRobotHandler.java

@@ -128,7 +128,7 @@ public class DingDingRobotHandler extends BaseHandler{
             Mac mac = Mac.getInstance(CommonConstant.HMAC_SHA256_ENCRYPTION_ALGO);
             mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), CommonConstant.HMAC_SHA256_ENCRYPTION_ALGO));
             byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
-            sign = URLEncoder.encode(new String(Base64.encodeBase64(signData)), CommonConstant.CHARSET_NAME);
+            sign = URLEncoder.encode(new String(Base64.encodeBase64(signData), CommonConstant.CHARSET_UTF_8));
         } catch (Exception e) {
             log.error("DingDingHandler#assembleSign fail!:{}", Throwables.getStackTraceAsString(e));
         }

austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/DingDingWorkNoticeHandler.java

@@ -9,7 +9,6 @@ import com.dingtalk.api.request.OapiMessageCorpconversationAsyncsendV2Request;
 import com.dingtalk.api.request.OapiMessageCorpconversationGetsendresultRequest;
 import com.dingtalk.api.request.OapiMessageCorpconversationRecallRequest;
 import com.dingtalk.api.response.OapiMessageCorpconversationAsyncsendV2Response;
-import com.dingtalk.api.response.OapiMessageCorpconversationGetsendresultResponse;
 import com.dingtalk.api.response.OapiMessageCorpconversationRecallResponse;
 import com.google.common.base.Throwables;
 import com.java3y.austin.common.constant.AustinConstant;
@@ -179,7 +178,7 @@ public class DingDingWorkNoticeHandler extends BaseHandler{
             OapiMessageCorpconversationGetsendresultRequest req = new OapiMessageCorpconversationGetsendresultRequest();
             req.setAgentId(Long.valueOf(account.getAgentId()));
             req.setTaskId(456L);
-            OapiMessageCorpconversationGetsendresultResponse rsp = client.execute(req, accessToken);
+            client.execute(req, accessToken);
         } catch (Exception e) {
             log.error("DingDingWorkNoticeHandler#pull fail:{}", Throwables.getStackTraceAsString(e));
         }

austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/SmsHandler.java

@@ -24,10 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.stereotype.Component;
 
+import java.security.SecureRandom;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.Random;
 
 /**
  * 短信发送处理
@@ -44,6 +44,11 @@ public class SmsHandler extends BaseHandler{
     private static final Integer AUTO_FLOW_RULE = 0;
     private static final String FLOW_KEY = "msgTypeSmsConfig";
     private static final String FLOW_KEY_PREFIX = "message_type_";
+    /**
+     * 安全随机数，重用性能与随机数质量更高
+     */
+    private static final SecureRandom secureRandom = new SecureRandom();
+
     @Autowired
     private SmsRecordDao smsRecordDao;
     @Autowired
@@ -99,7 +104,7 @@ public class SmsHandler extends BaseHandler{
         }
 
         // 生成一个随机数[1,total]，看落到哪个区间
-        int index = new Random().nextInt(total) + 1;
+        int index = secureRandom.nextInt(total) + 1;
 
         MessageTypeSmsConfig supplier = null;
         MessageTypeSmsConfig supplierBack = null;

austin-handler/src/main/java/com/java3y/austin/handler/receiver/kafka/ReceiverStart.java

@@ -84,14 +84,15 @@ public class ReceiverStart {
     @Bean
     public ConcurrentKafkaListenerContainerFactory filterContainerFactory(@Value("${austin.business.tagId.key}") String tagIdKey,
                                                                           @Value("${austin.business.tagId.value}") String tagIdValue) {
-        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
+        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(consumerFactory);
         factory.setAckDiscarded(true);
 
         factory.setRecordFilterStrategy(consumerRecord -> {
             if (Optional.ofNullable(consumerRecord.value()).isPresent()) {
                 for (Header header : consumerRecord.headers()) {
-                    if (header.key().equals(tagIdKey) && new String(header.value()).equals(new String(tagIdValue.getBytes(StandardCharsets.UTF_8)))) {
+                    if (header.key().equals(tagIdKey) &&
+                            new String(header.value(), StandardCharsets.UTF_8).equals(tagIdValue)) {
                         return false;
                     }
                 }

austin-handler/src/main/java/com/java3y/austin/handler/receiver/rabbit/RabbitMqReceiver.java

@@ -17,6 +17,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 
 
@@ -42,7 +43,7 @@ public class RabbitMqReceiver implements MessageReceiver {
     public void onMessage(Message message) {
         String messageType = message.getMessageProperties().getHeader("messageType");
         byte[] body = message.getBody();
-        String messageContent = new String(body);
+        String messageContent = new String(body, StandardCharsets.UTF_8);
         if (StringUtils.isBlank(messageContent)) {
             return;
         }

austin-handler/src/main/java/com/java3y/austin/handler/receiver/redis/RedisReceiver.java

@@ -0,0 +1,102 @@
+package com.java3y.austin.handler.receiver.redis;
+
+import com.alibaba.fastjson.JSON;
+import com.java3y.austin.common.domain.RecallTaskInfo;
+import com.java3y.austin.common.domain.TaskInfo;
+import com.java3y.austin.handler.receiver.MessageReceiver;
+import com.java3y.austin.handler.receiver.service.ConsumeService;
+import com.java3y.austin.support.constans.MessageQueuePipeline;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+
+/**
+ *
+ * Redis 消息队列实现类
+ *
+ * @author xiaoxiamao
+ * @date 2024/7/4
+ */
+@Slf4j
+@Component
+@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.REDIS)
+public class RedisReceiver implements MessageReceiver {
+
+    @Value("${austin.business.topic.name}")
+    private String sendTopic;
+    @Value("${austin.business.recall.topic.name}")
+    private String recallTopic;
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    private ConsumeService consumeService;
+
+    /**
+     * 初始化调度线程池
+     */
+    @PostConstruct
+    public void init() {
+        // 创建调度线程池
+        ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(2,
+                r -> new Thread(r, "RedisReceiverThread"));
+        // 定时调度
+        scheduler.scheduleWithFixedDelay(this::receiveSendMessage, 0, 1, TimeUnit.SECONDS);
+        scheduler.scheduleWithFixedDelay(this::receiveRecallMessage, 0, 1, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 消费发送消息
+     */
+    public void receiveSendMessage() {
+        receiveMessage(sendTopic, message -> {
+            log.debug("RedisReceiver#receiveSendMessage message:{}", message);
+            List<TaskInfo> taskInfoList = JSON.parseArray(message, TaskInfo.class);
+            consumeService.consume2Send(taskInfoList);
+        });
+    }
+
+    /**
+     * 消费撤回消息
+     */
+    public void receiveRecallMessage() {
+        receiveMessage(recallTopic, message -> {
+            log.debug("RedisReceiver#receiveRecallMessage recallTaskInfo:{}", message);
+            RecallTaskInfo recallTaskInfo = JSON.parseObject(message, RecallTaskInfo.class);
+            consumeService.consume2recall(recallTaskInfo);
+        });
+    }
+
+    /**
+     * 消息处理方法
+     *
+     * 处理责任链有去重处理，此处暂不做
+     *
+     * @param topic 消息主题
+     * @param consumer 消费处理逻辑
+     */
+    private void receiveMessage(String topic, Consumer<String> consumer) {
+        try {
+            while (true) {
+                // 阻塞操作，减少CPU，IO消耗
+                Optional<String> message = Optional.ofNullable(
+                        stringRedisTemplate.opsForList().rightPop(topic, 20, TimeUnit.SECONDS));
+                message.ifPresent(consumer);
+            }
+        } catch (Exception e) {
+            log.error("RedisReceiver#receiveMessage Error receiving messages from Redis topic {}: {}",
+                    topic, e.getMessage());
+        }
+    }
+}

austin-service-api-impl/src/main/java/com/java3y/austin/service/api/impl/action/send/SendAfterCheckAction.java

@@ -31,11 +31,14 @@ public class SendAfterCheckAction implements BusinessProcess<SendTaskModel> {
     /**
      * 邮件和手机号正则
      */
-    protected static final Map<Integer, String> CHANNEL_REGEX_EXP = new HashMap<>();
+    protected static final Map<Integer, String> CHANNEL_REGEX_EXP;
 
     static {
-        CHANNEL_REGEX_EXP.put(IdType.PHONE.getCode(), PHONE_REGEX_EXP);
+        Map<Integer, String> tempMap = new HashMap<>();
-        CHANNEL_REGEX_EXP.put(IdType.EMAIL.getCode(), EMAIL_REGEX_EXP);
+        tempMap.put(IdType.PHONE.getCode(), PHONE_REGEX_EXP);
+        tempMap.put(IdType.EMAIL.getCode(), EMAIL_REGEX_EXP);
+        // 初始化为不可变集合，避免被恶意修改
+        CHANNEL_REGEX_EXP = Collections.unmodifiableMap(tempMap);
     }
 
 

austin-stream/src/main/java/com/java3y/austin/stream/sink/AustinSink.java

@@ -13,6 +13,7 @@ import io.lettuce.core.RedisFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.flink.streaming.api.functions.sink.SinkFunction;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Date;
@@ -50,8 +51,8 @@ public class AustinSink implements SinkFunction<AnchorInfo> {
                  */
                 String redisMessageKey = CharSequenceUtil.join(StrPool.COLON, AustinConstant.CACHE_KEY_PREFIX, AustinConstant.MESSAGE_ID, info.getMessageId());
                 SimpleAnchorInfo messageAnchorInfo = SimpleAnchorInfo.builder().businessId(info.getBusinessId()).state(info.getState()).timestamp(info.getLogTimestamp()).build();
-                redisFutures.add(redisAsyncCommands.lpush(redisMessageKey.getBytes(), JSON.toJSONString(messageAnchorInfo).getBytes()));
+                redisFutures.add(redisAsyncCommands.lpush(redisMessageKey.getBytes(StandardCharsets.UTF_8), JSON.toJSONString(messageAnchorInfo).getBytes(StandardCharsets.UTF_8)));
-                redisFutures.add(redisAsyncCommands.expire(redisMessageKey.getBytes(), Duration.ofDays(3).toMillis() / 1000));
+                redisFutures.add(redisAsyncCommands.expire(redisMessageKey.getBytes(StandardCharsets.UTF_8), Duration.ofDays(3).toMillis() / 1000));
 
                 /**
                  * 1.构建userId维度的链路信息 数据结构list:{key,list}
@@ -59,17 +60,17 @@ public class AustinSink implements SinkFunction<AnchorInfo> {
                  */
                 SimpleAnchorInfo userAnchorInfo = SimpleAnchorInfo.builder().businessId(info.getBusinessId()).state(info.getState()).timestamp(info.getLogTimestamp()).build();
                 for (String id : info.getIds()) {
-                    redisFutures.add(redisAsyncCommands.lpush(id.getBytes(), JSON.toJSONString(userAnchorInfo).getBytes()));
+                    redisFutures.add(redisAsyncCommands.lpush(id.getBytes(StandardCharsets.UTF_8), JSON.toJSONString(userAnchorInfo).getBytes(StandardCharsets.UTF_8)));
-                    redisFutures.add(redisAsyncCommands.expire(id.getBytes(), (DateUtil.endOfDay(new Date()).getTime() - DateUtil.current()) / 1000));
+                    redisFutures.add(redisAsyncCommands.expire(id.getBytes(StandardCharsets.UTF_8), (DateUtil.endOfDay(new Date()).getTime() - DateUtil.current()) / 1000));
                 }
 
                 /**
                  * 2.构建消息模板维度的链路信息 数据结构hash:{key,hash}
                  * key:businessId,hashValue:{state,stateCount}
                  */
-                redisFutures.add(redisAsyncCommands.hincrby(String.valueOf(info.getBusinessId()).getBytes(),
+                redisFutures.add(redisAsyncCommands.hincrby(String.valueOf(info.getBusinessId()).getBytes(StandardCharsets.UTF_8),
-                        String.valueOf(info.getState()).getBytes(), info.getIds().size()));
+                        String.valueOf(info.getState()).getBytes(StandardCharsets.UTF_8), info.getIds().size()));
-                redisFutures.add(redisAsyncCommands.expire(String.valueOf(info.getBusinessId()).getBytes(),
+                redisFutures.add(redisAsyncCommands.expire(String.valueOf(info.getBusinessId()).getBytes(StandardCharsets.UTF_8),
                         ((DateUtil.offsetDay(new Date(), 30).getTime()) / 1000) - DateUtil.currentSeconds()));
 
                 return redisFutures;

austin-stream/src/main/java/com/java3y/austin/stream/utils/LettuceRedisUtils.java

@@ -27,7 +27,7 @@ public class LettuceRedisUtils {
 
     static {
         RedisURI redisUri = RedisURI.Builder.redis(AustinFlinkConstant.REDIS_IP)
-                .withPort(Integer.valueOf(AustinFlinkConstant.REDIS_PORT))
+                .withPort(Integer.parseInt(AustinFlinkConstant.REDIS_PORT))
                 .withPassword(AustinFlinkConstant.REDIS_PASSWORD.toCharArray())
                 .build();
         REDIS_CLIENT = RedisClient.create(redisUri);

austin-support/src/main/java/com/java3y/austin/support/constans/MessageQueuePipeline.java

@@ -8,6 +8,7 @@ package com.java3y.austin.support.constans;
  */
 public class MessageQueuePipeline {
     public static final String EVENT_BUS = "eventBus";
+    public static final String REDIS = "redis";
     public static final String KAFKA = "kafka";
     public static final String ROCKET_MQ = "rocketMq";
     public static final String RABBIT_MQ = "rabbitMq";

austin-support/src/main/java/com/java3y/austin/support/mq/redis/RedisSendMqServiceImpl.java

@@ -0,0 +1,63 @@
+package com.java3y.austin.support.mq.redis;
+
+import com.java3y.austin.support.constans.MessageQueuePipeline;
+import com.java3y.austin.support.mq.SendMqService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Service;
+
+/**
+ * Redis 消息队列实现类
+ *
+ * Guava Eventbus 和 Spring EventBus 只适用于单体服务
+ * Redis 适合单体、微服务，且无需单独部署三方消息队列，方便开发与简单应用
+ *
+ * @author xiaoxiamao
+ * @date 2024/7/4
+ */
+@Slf4j
+@Service
+@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.REDIS)
+public class RedisSendMqServiceImpl implements SendMqService {
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+
+    @Value("${austin.business.topic.name}")
+    private String sendTopic;
+    @Value("${austin.business.recall.topic.name}")
+    private String recallTopic;
+
+    /**
+     * Redis 发送消息，左进右出
+     *
+     * @param topic
+     * @param jsonValue
+     * @param tagId
+     */
+    @Override
+    public void send(String topic, String jsonValue, String tagId) {
+        // 非业务topic，抛错不发送
+        if (!sendTopic.equals(topic) && !recallTopic.equals(topic)) {
+            log.error("RedisSendMqServiceImpl#send The topic type is not supported! topic:{}, jsonValue:{}, tagId:{}",
+                    topic, jsonValue, tagId);
+            return;
+        }
+        log.debug("RedisSendMqServiceImpl#send topic:{}, jsonValue:{}, tagId:{}", topic, jsonValue, tagId);
+        stringRedisTemplate.opsForList().leftPush(topic, jsonValue);
+    }
+
+    /**
+     *  Redis 发送消息
+     *
+     * @param topic
+     * @param jsonValue
+     */
+    @Override
+    public void send(String topic, String jsonValue) {
+        send(topic, jsonValue, null);
+    }
+}

austin-support/src/main/java/com/java3y/austin/support/utils/AustinFileUtils.java

@@ -2,6 +2,7 @@ package com.java3y.austin.support.utils;
 
 import cn.hutool.core.io.IoUtil;
 import com.google.common.base.Throwables;
+import com.java3y.austin.common.constant.CommonConstant;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.File;
@@ -33,15 +34,27 @@ public class AustinFileUtils {
      * @return
      */
     public static File getRemoteUrl2File(String path, String remoteUrl) {
-
         InputStream inputStream = null;
         FileOutputStream fileOutputStream = null;
         try {
             URL url = new URL(remoteUrl);
+            String protocol = url.getProtocol();
+            // 防止SSRF攻击
+            if (!CommonConstant.HTTP.equalsIgnoreCase(protocol)
+                    && !CommonConstant.HTTPS.equalsIgnoreCase(protocol)
+                    && !CommonConstant.OSS.equalsIgnoreCase(protocol)) {
+                log.error("AustinFileUtils#getRemoteUrl2File fail:{}, remoteUrl:{}",
+                        "The remoteUrl is invalid, it can only be of the types http, https, and oss.", remoteUrl);
+                return null;
+            }
             File file = new File(path, url.getPath());
             inputStream = url.openStream();
             if (!file.exists()) {
-                file.getParentFile().mkdirs();
+                boolean res = file.getParentFile().mkdirs();
+                if (!res) {
+                    log.error("AustinFileUtils#getRemoteUrl2File Failed to create folder, path:{}, remoteUrl:{}", path, remoteUrl);
+                    return null;
+                }
                 fileOutputStream = new FileOutputStream(file);
                 IoUtil.copy(inputStream, fileOutputStream);
             }
@@ -49,20 +62,8 @@ public class AustinFileUtils {
         } catch (Exception e) {
             log.error("AustinFileUtils#getRemoteUrl2File fail:{},remoteUrl:{}", Throwables.getStackTraceAsString(e), remoteUrl);
         } finally {
-            if (Objects.nonNull(inputStream)) {
+            closeQuietly(inputStream);
-                try {
+            closeQuietly(fileOutputStream);
-                    inputStream.close();
-                } catch (IOException e) {
-                    log.error("close#inputStream fail:{}", Throwables.getStackTraceAsString(e));
-                }
-            }
-            if (Objects.nonNull(fileOutputStream)) {
-                try {
-                    fileOutputStream.close();
-                } catch (IOException e) {
-                    log.error("close#fileOutputStream fail:{}", Throwables.getStackTraceAsString(e));
-                }
-            }
         }
         return null;
     }
@@ -85,4 +86,33 @@ public class AustinFileUtils {
         return files;
     }
 
+    /**
+     * 关闭InputStream流
+     *
+     * @param inputStream
+     */
+    private static void closeQuietly(InputStream inputStream) {
+        if (Objects.nonNull(inputStream)) {
+            try {
+                inputStream.close();
+            } catch (IOException e) {
+                log.error("close#inputStream fail:{}", Throwables.getStackTraceAsString(e));
+            }
+        }
+    }
+
+    /**
+     * 关闭FileOutputStream流
+     *
+     * @param fileOutputStream
+     */
+    private static void closeQuietly(FileOutputStream fileOutputStream) {
+        if (Objects.nonNull(fileOutputStream)) {
+            try {
+                fileOutputStream.close();
+            } catch (IOException e) {
+                log.error("close#fileOutputStream fail:{}", Throwables.getStackTraceAsString(e));
+            }
+        }
+    }
 }

austin-support/src/main/java/com/java3y/austin/support/utils/OkHttpUtils.java

@@ -70,12 +70,12 @@ public class OkHttpUtils {
         StringBuilder sb = new StringBuilder(url);
         if (Objects.nonNull(params) && params.keySet().size() > 0) {
             boolean firstFlag = true;
-            for (String key : params.keySet()) {
+            for (Map.Entry<String, String> entry : params.entrySet()) {
                 if (firstFlag) {
-                    sb.append("?").append(key).append("=").append(params.get(key));
+                    sb.append("?").append(entry.getKey()).append("=").append(entry.getValue());
                     firstFlag = false;
                 } else {
-                    sb.append("&").append(key).append("=").append(params.get(key));
+                    sb.append("&").append(entry.getKey()).append("=").append(entry.getValue());
                 }
             }
         }
@@ -98,8 +98,8 @@ public class OkHttpUtils {
         FormBody.Builder formBuilder = new FormBody.Builder();
 
         if (Objects.nonNull(params) && params.keySet().size() > 0) {
-            for (String key : params.keySet()) {
+            for (Map.Entry<String, String> entry : params.entrySet()) {
-                formBuilder.add(key, params.get(key));
+                formBuilder.add(entry.getKey(), entry.getValue());
             }
         }
         Request.Builder builder = getBuilderWithHeaders(headers);

austin-support/src/main/java/com/java3y/austin/support/utils/RedisUtils.java

@@ -10,6 +10,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.data.redis.core.script.RedisScript;
 import org.springframework.stereotype.Component;
 
+import java.nio.charset.StandardCharsets;
 import java.util.*;
 
 /**
@@ -82,8 +83,8 @@ public class RedisUtils {
         try {
             redisTemplate.executePipelined((RedisCallback<String>) connection -> {
                 for (Map.Entry<String, String> entry : keyValues.entrySet()) {
-                    connection.setEx(entry.getKey().getBytes(), seconds,
+                    connection.setEx(entry.getKey().getBytes(StandardCharsets.UTF_8), seconds,
-                            entry.getValue().getBytes());
+                            entry.getValue().getBytes(StandardCharsets.UTF_8));
                 }
                 return null;
             });
@@ -99,8 +100,8 @@ public class RedisUtils {
     public void lPush(String key, String value, Long seconds) {
         try {
             redisTemplate.executePipelined((RedisCallback<String>) connection -> {
-                connection.lPush(key.getBytes(), value.getBytes());
+                connection.lPush(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
-                connection.expire(key.getBytes(), seconds);
+                connection.expire(key.getBytes(StandardCharsets.UTF_8), seconds);
                 return null;
             });
         } catch (Exception e) {
@@ -142,8 +143,11 @@ public class RedisUtils {
         try {
             redisTemplate.executePipelined((RedisCallback<String>) connection -> {
                 for (Map.Entry<String, String> entry : keyValues.entrySet()) {
-                    connection.hIncrBy(entry.getKey().getBytes(), entry.getValue().getBytes(), delta);
+                    connection.hIncrBy(entry.getKey().getBytes(StandardCharsets.UTF_8),
-                    connection.expire(entry.getKey().getBytes(), seconds);
+                            entry.getValue().getBytes(StandardCharsets.UTF_8),
+                            delta);
+                    connection.expire(entry.getKey().getBytes(StandardCharsets.UTF_8),
+                            seconds);
                 }
                 return null;
             });

austin-support/src/main/java/com/java3y/austin/support/utils/TaskInfoUtils.java

@@ -13,7 +13,7 @@ import java.util.Date;
  * @author 3y
  */
 public class TaskInfoUtils {
-    private static final int TYPE_FLAG = 1000000;
+    private static final long TYPE_FLAG = 1000000L;
     private static final String CODE = "track_code_bid";
     private TaskInfoUtils() {
     }
@@ -34,7 +34,7 @@ public class TaskInfoUtils {
      */
     public static Long generateBusinessId(Long templateId, Integer templateType) {
         Integer today = Integer.valueOf(DateUtil.format(new Date(), DatePattern.PURE_DATE_PATTERN));
-        return Long.valueOf(String.format("%d%s", templateType * TYPE_FLAG + templateId, today));
+        return Long.valueOf(String.format("%d%s", templateType.longValue() * TYPE_FLAG + templateId, today));
     }
 
     /**

austin-web/src/main/java/com/java3y/austin/web/controller/MessageTemplateController.java

@@ -201,7 +201,11 @@ public class MessageTemplateController {
         try {
             File localFile = new File(filePath);
             if (!localFile.exists()) {
-                localFile.mkdirs();
+                boolean res = localFile.mkdirs();
+                if (!res) {
+                    log.error("MessageTemplateController#upload fail! Failed to create folder.");
+                    throw new CommonException(RespStatusEnum.SERVICE_ERROR);
+                }
             }
             file.transferTo(localFile);
         } catch (Exception e) {

austin-web/src/main/java/com/java3y/austin/web/utils/Convert4Amis.java

@@ -26,7 +26,6 @@ import me.chanjar.weixin.mp.bean.template.WxMpTemplate;
 
 import java.lang.reflect.Field;
 import java.util.*;
-import java.util.stream.Collectors;
 
 /**
  * 【该类的逻辑不用看，没有什么意义】
@@ -231,62 +230,57 @@ public class Convert4Amis {
      * @return
      */
     public static Set<String> getPlaceholderList(String content) {
-        char[] textChars = content.toCharArray();
+
-        StringBuilder textSofar = new StringBuilder();
+        // 内容为空，直接返回
+        if (content == null || content.isEmpty()) {
+            return Collections.emptySet();
+        }
+
         StringBuilder sb = new StringBuilder();
-        // 存储占位符 位置信息集合
+        Set<String> placeholderSet = new HashSet<>();
-        List<String> placeholderList = new ArrayList<>();
-        // 当前标识
         int modeTg = IGNORE_TG;
-        for (int m = 0; m < textChars.length; m++) {
+
-            char c = textChars[m];
+        for (char c : content.toCharArray()) {
-            textSofar.append(c);
             switch (c) {
-                case '{': {
+                case '{':
-                    modeTg = START_TG;
+                    if (modeTg == IGNORE_TG) {
-                    sb.append(c);
+                        sb.append(c);
-                }
+                        modeTg = START_TG;
-                break;
+                    }
-                case '$': {
+                    break;
+                case '$':
                     if (modeTg == START_TG) {
                         sb.append(c);
                         modeTg = READ_TG;
                     } else {
-                        if (modeTg == READ_TG) {
+                        sb.setLength(0);
-                            sb = new StringBuilder();
+                        modeTg = IGNORE_TG;
-                            modeTg = IGNORE_TG;
-                        }
                     }
-                }
+                    break;
-                break;
+                case '}':
-                case '}': {
                     if (modeTg == READ_TG) {
-                        modeTg = IGNORE_TG;
                         sb.append(c);
-                        String str = sb.toString();
+                        String placeholder = sb.toString();
-                        if (StrUtil.isNotEmpty(str)) {
+                        placeholderSet.add(placeholder.replaceAll("[\\{\\$\\}]", ""));
-                            placeholderList.add(str);
+                        sb.setLength(0);
-                            textSofar = new StringBuilder();
+                        modeTg = IGNORE_TG;
-                        }
-                        sb = new StringBuilder();
                     } else if (modeTg == START_TG) {
+                        sb.setLength(0);
                         modeTg = IGNORE_TG;
-                        sb = new StringBuilder();
                     }
                     break;
-                }
+                default:
-                default: {
                     if (modeTg == READ_TG) {
                         sb.append(c);
                     } else if (modeTg == START_TG) {
+                        sb.setLength(0);
                         modeTg = IGNORE_TG;
-                        sb = new StringBuilder();
                     }
-                }
+                    break;
             }
         }
-        Set<String> result = placeholderList.stream().map(s -> s.replaceAll("\\{", "").replaceAll("\\$", "").replaceAll("\\}", "")).collect(Collectors.toSet());
+
-        return result;
+        return placeholderSet;
     }
 
     /**

austin-web/src/main/java/com/java3y/austin/web/utils/SpringFileUtils.java

@@ -1,18 +1,19 @@
 package com.java3y.austin.web.utils;
 
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.web.multipart.MultipartFile;
 
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Objects;
 
 
 /**
  * @author 3y
  * multipartFile 转成 File 对象
  */
+@Slf4j
 public class SpringFileUtils {
     private SpringFileUtils() {
     }
@@ -26,23 +27,14 @@ public class SpringFileUtils {
     public static File getFile(MultipartFile multipartFile) {
         String fileName = multipartFile.getOriginalFilename();
         File file = new File(fileName);
-        OutputStream out = null;
+        try (OutputStream out = new FileOutputStream(file)){
-        try {
-            out = new FileOutputStream(file);
             byte[] ss = multipartFile.getBytes();
             for (int i = 0; i < ss.length; i++) {
                 out.write(ss[i]);
             }
         } catch (IOException e) {
-            e.printStackTrace();
+            log.error("SpringFileUtils#getFile multipartFile is converted to File error:{}", e);
-        } finally {
+            return null;
-            if (Objects.nonNull(out)) {
-                try {
-                    out.close();
-                } catch (IOException e) {
-                    e.printStackTrace();
-                }
-            }
         }
         return file;
     }