rocketmq+oth

pom.xml

@@ -79,7 +79,17 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-client</artifactId>
-            <version>5.2.0</version>
+            <version>5.3.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-common</artifactId>
+            <version>5.3.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-acl</artifactId>
+            <version>5.3.0</version>
         </dependency>
         <dependency>
             <groupId>org.elasticsearch.client</groupId>
@@ -319,6 +329,7 @@
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-maven-plugin</artifactId>
                 <configuration>
+                    <includeSystemScope>true</includeSystemScope>
                     <excludes>
                         <exclude>
                             <groupId>org.projectlombok</groupId>

src/main/java/com/scbfkj/uni/aop/UniLogAop.java

@@ -51,7 +51,23 @@ public class UniLogAop {
             // 执行被拦截的方法，并返回结果
             Object proceed = joinPoint.proceed(args);
             ObjectNode objectNode = SystemUtil.OBJECT_MAPPER.createObjectNode();
-            objectNode.set("input", dataFormatUtil.toJsonNode(args));
+
+            Object[] newArgs = new Object[args.length];
+//            if(args.length > 1){
+//                newArgs[0] = args[0];
+//            }
+            Integer num = 0;
+            for(Object obj:args){
+                String name = obj.getClass().getSimpleName();
+                if(name.contains("HashMap")){
+                    newArgs[num] = obj;
+                }
+                else if(name.contains("MultipartFile")){
+//                    newArgs[num] = obj;
+                }
+                num++;
+            }
+            objectNode.set("input", dataFormatUtil.toJsonNode(newArgs));
             objectNode.set("output", dataFormatUtil.toJsonNode(proceed));
             LOG.info(objectNode.toString(), LogTarget.INTERFACE);
             return proceed;

src/main/java/com/scbfkj/uni/apis/GenericApi.java

@@ -1,5 +1,9 @@
 package com.scbfkj.uni.apis;
 
+import cn.hutool.json.JSONObject;
+import com.alibaba.fastjson.JSON;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import org.springframework.http.HttpStatusCode;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.PostMapping;
@@ -25,7 +29,10 @@ import com.scbfkj.uni.library.RequestUtil;
 import com.scbfkj.uni.repositries.ServiceInfoRepository;
 import com.scbfkj.uni.services.GenericService;
 import com.scbfkj.uni.utils.SystemUtil;
+import org.springframework.web.multipart.MultipartFile;
 
+import java.io.File;
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
@@ -69,7 +76,55 @@ public class GenericApi {
         Object result = genericService.execute(body);
         return ResponseEntity.ok(escapeXss(result));
     }
-    
+
+    /**
+     * 根据请求URI匹配对应的服务，并执行服务处理。
+     *
+     * @param body 请求体，包含需要处理的数据
+     *
+     * @return 返回处理结果，包含处理后的数据和状态信息
+     */
+    @PostMapping("/openApi/fileupload")
+    public ResponseEntity<JsonNode> fileUploadService(@RequestParam(required = false) Map<String, Object> params,
+                                                      @RequestParam("file") MultipartFile file
+    ) throws IOException {
+        String osName = System.getProperty("os.name").toLowerCase();
+        String uploadDir;
+        if (osName.contains("windows")){
+            uploadDir = "D:\\test\\";
+        }
+        else{
+            uploadDir = "./uploads/";
+        }
+        File uploadDirFile = new File(uploadDir);
+        if (!uploadDirFile.exists()) {
+            uploadDirFile.mkdir();
+        }
+        String fileName = file.getOriginalFilename();
+        int lastIndex = fileName.lastIndexOf(".");
+        String extension = fileName.substring(lastIndex);
+        String extFileName = fileName.substring(0,lastIndex);
+        long timestamp1 = System.currentTimeMillis();
+
+        if(!extension.equals(".xls") && !extension.equals(".xlsx")){
+            // 如果没有找到匹配的服务，返回404状态码
+            return ResponseEntity.status(HttpStatusCode.valueOf(404)).build();
+        }
+
+        String filePath = uploadDir + extFileName + "_" + timestamp1 + extension;
+
+        file.transferTo(new File(filePath));
+
+        ObjectMapper mapper = new ObjectMapper();
+        params.put("dataContent", filePath);
+        ObjectNode objectNode = mapper.valueToTree(params);
+
+        Object result = genericService.execute(objectNode);
+        return ResponseEntity.ok(escapeXss(result));
+
+//            return ResponseEntity.ok(escapeXss(UniResult.success(uploadDir)));
+    }
+
     /**
      * 根据请求URI匹配对应的服务，并执行服务处理。
      *

src/main/java/com/scbfkj/uni/dtos/requests/UniRequest.java

@@ -16,6 +16,7 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import java.io.IOException;
 import java.util.*;
 
 /**
@@ -160,13 +161,13 @@ public class UniRequest<T extends IRequestContent> {
             uniRequest.setUri(urlNode.asText());
         }
         // 如果uniRequest中的URI为空，尝试从RequestUtil Bean中获取URI
-        if (uniRequest.getUri() == null) {
-            RequestUtil bean = SpringUtil.getBean(RequestUtil.class);
-            // 如果RequestUtil Bean存在，将其URI设置到uniRequest中
-            if (bean != null) {
-                uniRequest.setUri(bean.getUri());
-            }
-        }
+//        if (uniRequest.getUri() == null) {
+//            RequestUtil bean = SpringUtil.getBean(RequestUtil.class);
+//            // 如果RequestUtil Bean存在，将其URI设置到uniRequest中
+//            if (bean != null) {
+//                    uniRequest.setUri(bean.getUri());
+//            }
+//        }
     }
     
     /**
@@ -404,8 +405,15 @@ public class UniRequest<T extends IRequestContent> {
             // 如果不满足特殊格式，遍历数组并将每个元素转换为SimpleRequestContent。
             for (JsonNode it : dataContentNode) {
                 // 将JsonNode转换为SimpleRequestContent对象。
-                SimpleRequestContent simpleRequestContent =
-                        SystemUtil.OBJECT_MAPPER.readValue(it.toString(), SimpleRequestContent.class);
+                SimpleRequestContent simpleRequestContent = new SimpleRequestContent();
+                try {
+                    simpleRequestContent = SystemUtil.OBJECT_MAPPER.readValue(it.toString(), SimpleRequestContent.class);
+                }
+                catch (IOException e) {
+                    ObjectNode objectNode = SystemUtil.OBJECT_MAPPER.createObjectNode();
+                    objectNode.put("bfRequestData", it);
+                    simpleRequestContent = SystemUtil.OBJECT_MAPPER.readValue(objectNode.toString(), SimpleRequestContent.class);
+                }
                 // 将转换后的请求内容添加到列表中。
                 contents.add(simpleRequestContent);
             }

src/main/java/com/scbfkj/uni/entities/DataSourceInfo.java

@@ -53,12 +53,18 @@ public class DataSourceInfo {
      */
     @Column(name = "http_headers", columnDefinition = "LONGTEXT")
     private String httpHeaders;
-    
+
     /**
      * HTTP请求体信息，用于特殊的数据源交互需求。
      */
     @Column(name = "http_body", columnDefinition = "LONGTEXT")
     private String httpBody;
+
+    /**
+     * HTTP请求类型，针对HTTP类型数据源。
+     */
+    @Column(name = "content_type", columnDefinition = "LONGTEXT")
+    private String contentType;
     
     /**
      * 数据源驱动类名称，用于加载相应的数据库驱动。

src/main/java/com/scbfkj/uni/library/DataEncryptionUtil.java

@@ -1,5 +1,8 @@
 package com.scbfkj.uni.library;
 
+import cn.hutool.core.util.HexUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.asymmetric.SM2;
 import org.springframework.stereotype.Component;
 
 import cn.hutool.crypto.SecureUtil;
@@ -355,4 +358,54 @@ public class DataEncryptionUtil {
         // 解密时同时需要提供公钥信息，这是为了验证解密过程的正确性或用于特定的解密算法需求
         return rsaUtil.decrypt(data, systemConfig.getPrivateKey(), systemConfig.getPublicAppKey(), KeyType.PrivateKey);
     }
+
+
+    /**
+     * SM2私钥签名
+     *
+     * @param privateKey 私钥
+     * @param content    待签名内容
+     * @return 签名值
+     */
+    public static String signSM2(String privateKey, String content) {
+        SM2 sm2 = new SM2(privateKey, null);
+        return sm2.signHex(HexUtil.encodeHexStr(content));
+    }
+
+    /**
+     * SM2公钥验签
+     *
+     * @param publicKey 公钥
+     * @param content   原始内容
+     * @param sign      签名
+     * @return 验签结果
+     */
+    public static boolean checkSignSM2(String publicKey, String content, String sign) {
+        SM2 sm2 = new SM2(null, publicKey);
+        return sm2.verifyHex(HexUtil.encodeHexStr(content), sign);
+    }
+
+    /**
+     * SM2公钥加密
+     *
+     * @param publicKey SM2公钥
+     * @param content   原文
+     * @return
+     */
+    public static String encryptBase64SM2(String publicKey,String content) {
+        SM2 sm2 = new SM2(null, publicKey);
+        return sm2.encryptBase64(content, KeyType.PublicKey);
+    }
+
+    /**
+     * SM2私钥解密
+     *
+     * @param privateKey SM2私钥
+     * @param content SM2加密字符串
+     * @return
+     */
+    public static String decryptBase64SM2(String privateKey,String content) {
+        SM2 sm2 = new SM2(privateKey, null);
+        return StrUtil.utf8Str(sm2.decrypt(content, KeyType.PrivateKey));
+    }
 }

src/main/java/com/scbfkj/uni/library/DataFormatUtil.java

@@ -1,6 +1,8 @@
 package com.scbfkj.uni.library;
 
 import cn.hutool.core.date.DateUtil;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.XML;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -302,7 +304,7 @@ public class DataFormatUtil {
         List<String> result = new ArrayList<>();
         for (String xml : xmls) {
             // 将XML字符串解析为JSON对象
-            Object json = XML_MAPPER.readValue(xml, Object.class);
+            JSONObject json = XML.toJSONObject(xml);
             // 将JSON对象转换为字符串
             String data = OBJECT_MAPPER.writeValueAsString(json);
             result.add(data);
@@ -319,7 +321,7 @@ public class DataFormatUtil {
      */
     public String xml2json(String xml) throws JsonProcessingException {
         // 将XML字符串解析为JSON对象
-        Object json = XML_MAPPER.readValue(xml, Object.class);
+        JSONObject json = XML.toJSONObject(xml);
         // 将JSON对象转换为字符串
         return OBJECT_MAPPER.writeValueAsString(json);
     }

src/main/java/com/scbfkj/uni/library/JsUtil.java

@@ -59,7 +59,10 @@ public class JsUtil {
      */
     public Object exec(String script, Object args) throws JsonProcessingException {
         Source source = Source.create("js", script);
-        try (Context context = Context.create("js")) {
+//        try (Context context = Context.create("js")) {
+        try (Context context = Context.newBuilder("js")
+                .option("engine.WarnInterpreterOnly", "false")
+                .build()) {
             Value bindings = context.getBindings("js");
             bindings.putMember("paramsArgsObject", dataFormatUtil.toString(args));
             context.eval("js", ARGS);
@@ -120,7 +123,9 @@ public class JsUtil {
     private static String jsToString(Value value) {
         // 创建一个包含转换逻辑的JavaScript源码
         Source source = Source.create("js", TO_STRING_SCRIPT);
-        try (Context context = Context.create("js")) {
+        try (Context context = Context.newBuilder("js")
+                .option("engine.WarnInterpreterOnly", "false")
+                .build()) {
             // 在JavaScript环境中评估该源码，得到一个函数
             Value function = context.eval(source);
             // 使用该函数执行给定的JavaScript值，并将结果转换为字符串

src/main/java/com/scbfkj/uni/library/RequestUtil.java

@@ -134,6 +134,7 @@ public class RequestUtil {
         appConnectLogRepository.deleteAllByExpireTimeBefore(LocalDateTime.now());
         String appToken = getAppToken();
         final String ipAddr = getIpAddr();
+
         Optional<AppConnectLog> appConnectLogOptional =
                 appConnectLogRepository.findFirstByAppTokenAndRequestIpOrderByIdDesc(appToken, ipAddr);
         return appConnectLogOptional.map(it -> {

src/main/java/com/scbfkj/uni/process/Http.java

@@ -1,8 +1,13 @@
 package com.scbfkj.uni.process;
 
+import cn.hutool.json.JSONConverter;
+import cn.hutool.json.JSONObject;
+import com.alibaba.fastjson2.JSON;
 import org.apache.http.HttpResponse;
 import org.apache.http.HttpStatus;
+import org.apache.http.NameValuePair;
 import org.apache.http.client.HttpClient;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
@@ -12,6 +17,7 @@ import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.entity.ContentType;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.util.EntityUtils;
 import org.springframework.http.HttpMethod;
 import org.springframework.stereotype.Component;
@@ -25,11 +31,7 @@ import jakarta.annotation.Resource;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
+import java.util.*;
 
 /**
  * @author l7871
@@ -76,9 +78,16 @@ public class Http {
         if (Objects.isNull(defaultBody)) {
             defaultBody = objectMap.get("body");
         }
-        host = objectMap.get("host").toString();
+        String contentType;
+        if(Objects.isNull(objectMap.get("contentType"))){
+            contentType = "application/json";
+        }
+        else{
+            contentType = objectMap.get("contentType").toString();
+        }
+        host = objectMap.get("url").toString();
         // 执行Web API请求
-        return execWebApi(headers, method, defaultBody, host);
+        return execWebApi(headers, method, defaultBody, host,contentType);
     }
     
     /**
@@ -101,10 +110,11 @@ public class Http {
             hashMap.put("url", it.getHost());
             hashMap.put("headers", it.getHttpHeaders());
             hashMap.put("body", it.getHttpBody());
+            hashMap.put("contentType", it.getContentType());
             return hashMap;
         }).orElse(new HashMap<>());
     }
-    
+
     /**
      * 执行Web API请求。
      *
@@ -117,6 +127,56 @@ public class Http {
      */
     public UniResult<Object> execWebApi(Object headers, String method, Object defaultBody, String url)
             throws IOException {
+        return execWebApi(headers, method, defaultBody, url, "application/json");
+//        HttpUriRequest request;
+//        // 根据请求方法创建相应的HTTP请求对象，并设置请求体
+//        String lowerCase = method.toLowerCase(Locale.getDefault());
+//        if (HttpMethod.GET.name().equalsIgnoreCase(lowerCase)) {
+//            request = new HttpGet(url);
+//        } else if (HttpMethod.PUT.name().equalsIgnoreCase(lowerCase)) {
+//            request = new HttpPut(url);
+//            ((HttpPut) request).setEntity(
+//                    new ByteArrayEntity(dataFormatUtil.toString(defaultBody).getBytes(StandardCharsets.UTF_8)));
+//        } else if (HttpMethod.DELETE.name().equalsIgnoreCase(lowerCase)) {
+//            request = new HttpDelete(url);
+//        } else {
+//            request = new HttpPost(url);
+//            ((HttpPost) request).setEntity(
+//                    new ByteArrayEntity(dataFormatUtil.toString(defaultBody).getBytes(StandardCharsets.UTF_8)));
+//        }
+//        // 如果提供了请求头信息，则添加到请求中
+//        if (headers instanceof Map<?, ?> hs) {
+//            hs.forEach((k, v) -> request.addHeader((String) k, String.valueOf(v)));
+//        }
+//        // 执行HTTP请求并处理响应
+//        HttpResponse response = httpClient.execute(request);
+//        String contentType = response.getEntity().getContentType().getValue();
+//        int statusCode = response.getStatusLine().getStatusCode();
+//        // 根据响应类型和状态码返回结果
+//        if (ContentType.APPLICATION_OCTET_STREAM.getMimeType().equalsIgnoreCase(contentType)) {
+//            return UniResult.success(EntityUtils.toByteArray(response.getEntity()));
+//        }
+//        String responseBody = EntityUtils.toString(response.getEntity());
+//        if (statusCode >= HttpStatus.SC_OK && statusCode < HttpStatus.SC_MULTIPLE_CHOICES) {
+//            return UniResult.success(responseBody);
+//        } else {
+//            return UniResult.fail(responseBody, "-3");
+//        }
+    }
+
+    /**
+     * 执行Web API请求。
+     *
+     * @param headers 请求头信息。
+     * @param method HTTP请求方法。
+     * @param defaultBody 请求体内容。
+     * @param url 请求的URL。
+     * @param contentTypeRequest 请求的CONTENT-TYPE。
+     *
+     * @return 执行结果，为一个Map对象。
+     */
+    public UniResult<Object> execWebApi(Object headers, String method, Object defaultBody, String url,String contentTypeRequest)
+            throws IOException {
         HttpUriRequest request;
         // 根据请求方法创建相应的HTTP请求对象，并设置请求体
         String lowerCase = method.toLowerCase(Locale.getDefault());
@@ -130,8 +190,20 @@ public class Http {
             request = new HttpDelete(url);
         } else {
             request = new HttpPost(url);
-            ((HttpPost) request).setEntity(
-                    new ByteArrayEntity(dataFormatUtil.toString(defaultBody).getBytes(StandardCharsets.UTF_8)));
+
+            if(contentTypeRequest.equals("application/x-www-form-urlencoded")){
+                List<NameValuePair> params = new ArrayList<>();
+                JSONObject jsonObject = new JSONObject(dataFormatUtil.toString(defaultBody));
+                for(String key : jsonObject.keySet()){
+                    params.add(new BasicNameValuePair(key, jsonObject.get(key).toString()));
+                }
+                ((HttpPost) request).setEntity(new UrlEncodedFormEntity(params));
+            }
+            else {
+                ByteArrayEntity byteArrayEntity = new ByteArrayEntity(dataFormatUtil.toString(defaultBody).getBytes(StandardCharsets.UTF_8));
+                byteArrayEntity.setContentType(contentTypeRequest);
+                ((HttpPost) request).setEntity(byteArrayEntity);
+            }
         }
         // 如果提供了请求头信息，则添加到请求中
         if (headers instanceof Map<?, ?> hs) {

src/main/java/com/scbfkj/uni/process/RocketMQClient.java

@@ -3,14 +3,20 @@ package com.scbfkj.uni.process;
 import cn.hutool.extra.spring.SpringUtil;
 import com.alibaba.fastjson.JSONObject;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.scbfkj.uni.dtos.responses.UniResult;
 import com.scbfkj.uni.exceptions.UniException;
 import com.scbfkj.uni.library.DataFormatUtil;
 import com.scbfkj.uni.logs.Log;
 import com.scbfkj.uni.utils.SystemUtil;
+import io.micrometer.common.util.StringUtils;
 import jakarta.annotation.Resource;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.rocketmq.acl.common.AclClientRPCHook;
+import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
@@ -18,15 +24,16 @@ import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.remoting.common.RemotingHelper;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.springframework.jms.core.JmsTemplate;
 import org.springframework.stereotype.Component;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
 
 /**
  * @author l7871
@@ -39,20 +46,43 @@ public class RocketMQClient {
     
     @Resource
     private DataFormatUtil dataFormatUtil;
+    /**
+     * 因为生产者是线程安全的所以相同的连接只需要一个生产者实例 不需要对象池
+     */
+    private final Map<String, DefaultMQProducer> producerMap = new HashMap<>();
 
     /**
      * 读取配置文件中设置的rocketmq相关属性，创建消息生产者
      */
     private DefaultMQProducer getRocketMqProducer(RocketMqConfig rocketMqConfig){
-        // 1 创建消息生产者，指定生成组名
-        DefaultMQProducer defaultMQProducer = new DefaultMQProducer(rocketMqConfig.getGroupId());
-        // 2 指定NameServer的地址
-        defaultMQProducer.setNamesrvAddr(rocketMqConfig.getAddr());
-        // 3 设置消息超时时间
-        defaultMQProducer.setSendMsgTimeout(Integer.parseInt("10000"));
-        // 4 同步发送消息，如果SendMsgTimeout时间内没有发送成功，则重试retryWhenSendFailed次
-        defaultMQProducer.setRetryTimesWhenSendFailed(Integer.parseInt("5"));
-        return defaultMQProducer;
+        synchronized (producerMap) {
+            if (producerMap.containsKey(rocketMqConfig.getAddr() + rocketMqConfig.getGroupId())) {
+                return producerMap.get(rocketMqConfig.getAddr() + rocketMqConfig.getGroupId());
+            }
+            AclClientRPCHook hook = new AclClientRPCHook(new SessionCredentials(rocketMqConfig.getAccessKey(), rocketMqConfig.getSecretKey()));
+            // 1 创建消息生产者，指定生成组名
+            DefaultMQProducer defaultMQProducer = new DefaultMQProducer(rocketMqConfig.getGroupId(),hook);
+            // 2 指定NameServer的地址
+            defaultMQProducer.setNamesrvAddr(rocketMqConfig.getAddr());
+            // 设置ACL参数
+            defaultMQProducer.setVipChannelEnabled(false);
+            // 3 设置消息超时时间
+            defaultMQProducer.setSendMsgTimeout(Integer.parseInt("10000"));
+            // 4 同步发送消息，如果SendMsgTimeout时间内没有发送成功，则重试retryWhenSendFailed次
+            defaultMQProducer.setRetryTimesWhenSendFailed(Integer.parseInt("5"));
+
+            // 将新的生产者实例与连接字符串关联并保存到映射中。
+            producerMap.put(rocketMqConfig.getAddr() + rocketMqConfig.getGroupId(), defaultMQProducer);
+
+            try {
+                defaultMQProducer.start();
+            } catch (MQClientException e) {
+                throw new RuntimeException(e);
+            }
+
+            // 返回新的生产者实例。
+            return defaultMQProducer;
+        }
     }
 
     /**
@@ -68,21 +98,40 @@ public class RocketMQClient {
      * @param queueName 需要发送消息的队列名称
      * @param data 要发送的消息内容，支持发送Object数组或实现Iterable接口的对象
      */
-    public void sendMessage(RocketMqConfig rocketMqConfig,String topic,String tag, Object data)
+    public UniResult<String> sendMessage(RocketMqConfig rocketMqConfig,String topic,String tag, Object data,Object sign)
             throws UniException, JsonProcessingException {
         DefaultMQProducer rocketMqProducer = getRocketMqProducer(rocketMqConfig);
-
+        SendResult sendResult = new SendResult();
         try {
-            rocketMqProducer.start();
             Message message = new Message(topic, tag, data.toString().getBytes());
+            if(!StringUtils.isEmpty(rocketMqConfig.getFirstMsgType())){
+                message.putUserProperty("firstMsgType",rocketMqConfig.getFirstMsgType());
+            }
+            if(!StringUtils.isEmpty(rocketMqConfig.getEncryptedFlag())){
+                message.putUserProperty("encryptedFlag",rocketMqConfig.getEncryptedFlag());
+            }
+            if(!StringUtils.isEmpty(rocketMqConfig.getAppId())){
+                message.putUserProperty("appId",rocketMqConfig.getAppId());
+            }
+            if(sign != null){
+                message.putUserProperty("sign",sign.toString());
+            }
+            sendResult = rocketMqProducer.send(message, 10000);
 
-            SendResult result = null;
-            result = rocketMqProducer.send(message);
+            if (sendResult.getSendStatus() == null) {
+                rocketMqProducer.shutdown();
+                producerMap.remove(rocketMqConfig.getAddr() + rocketMqConfig.getGroupId());
+                return UniResult.fail(String.valueOf(sendResult),"-101");
+            }
 
         } catch (Exception e) {
+            rocketMqProducer.shutdown();
+            producerMap.remove(rocketMqConfig.getAddr() + rocketMqConfig.getGroupId());
+
             throw new RuntimeException(e);
         } finally {
-            rocketMqProducer.shutdown();
+            return UniResult.success(String.valueOf(sendResult));
+//            rocketMqProducer.shutdown();
         }
     }
 
@@ -92,6 +141,7 @@ public class RocketMQClient {
         DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(rocketMqConfig.getGroupId() );
         // 2 指定NameServer的地址
         defaultMQPushConsumer.setNamesrvAddr(rocketMqConfig.getAddr());
+        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);//订阅的主题和标签（*代表所有标签）
         // 3 指定消费者订阅的主题和标签
         try {
             defaultMQPushConsumer.subscribe(topic, tag);
@@ -118,23 +168,24 @@ public class RocketMQClient {
             throws UniException, JsonProcessingException {
         DefaultMQPushConsumer rocketMqConsumer = getRocketMqConsumer(rocketMqConfig, topic,tag);
 
-        rocketMqConsumer.registerMessageListener((List<MessageExt> list, ConsumeConcurrentlyContext context) -> {
-
-            try {
-                System.out.println("收到消息--》" + list);
-                for (MessageExt messageExt : list) {
-                    String message=new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
-                    JSONObject object= JSONObject.parseObject(message);
-                    String fileId = (String) object.get("fileId");
-                    String fileCreateDate = (String) object.get("fileCreateDate");
+        rocketMqConsumer.registerMessageListener(
+                (List<MessageExt> list, ConsumeConcurrentlyContext context) -> {
+                    List<String> messageList = new ArrayList<>();
+                    try {
+                        for (MessageExt messageExt : list) {
+                            String message = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
+        //                    JSONObject object= JSONObject.parseObject(message);
+
+                            messageList.add(message);
+                        }
+
+                    } catch (Exception e) {
+                        throw new UniException(e);
+                    }
+//                 return   ListenerMQ(messageList);
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                 }
-
-            } catch (Exception e) {
-                throw new UniException(e);
-            }
-
-            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-        });
+        );
 
         // 5 启动消费者
         try {
@@ -145,7 +196,10 @@ public class RocketMQClient {
 
         return null;
     }
-
+public UniResult ListenerMQ(List<String> messageList) throws UniException {
+    return UniResult.success(messageList);
+//    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+}
 
     @Data
     @AllArgsConstructor
@@ -158,6 +212,12 @@ public class RocketMQClient {
         private String secretKey;
 
         private String groupId;
+
+        private String firstMsgType;
+
+        private String encryptedFlag;
+
+        private String appId;
     }
 }
 

src/main/java/com/scbfkj/uni/repositries/AppConnectLogRepository.java

@@ -1,9 +1,11 @@
 package com.scbfkj.uni.repositries;
 
+import jakarta.transaction.Transactional;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 
 import com.scbfkj.uni.entities.AppConnectLog;
+import org.springframework.data.jpa.repository.Modifying;
 
 import java.time.LocalDateTime;
 import java.util.List;
@@ -25,6 +27,8 @@ public interface AppConnectLogRepository extends JpaRepository<AppConnectLog, Lo
      *
      * @return 删除的日志记录数量
      */
+    @Modifying
+    @Transactional
     int deleteAllByExpireTimeBefore(LocalDateTime dateTime);
     
     /**