集成mq发送ai

ruoyi-admin/src/main/resources/application.yml

@@ -33,6 +33,15 @@ server:
       # Tomcat启动初始化的线程数，默认值10
       min-spare: 100
 
+rocketmq:
+  name-server: 49.232.250.68:9876
+  producer:
+    group: abc
+    send-message-timeout: 60000
+    retry-times-when-send-failed: 3
+    retry-times-when-send-async-failed: 3
+    customized-trace-topic: ai
+
 # 日志配置
 logging:
   level:
@@ -70,14 +79,14 @@ spring:
   # redis 配置
   redis:
     # 地址
-    host: 43.140.224.18
+    host: 49.232.250.68
 #    host: localhost
     # 端口，默认为6379
-    port: 6379
+    port: 16379
     # 数据库索引
     database: 0
     # 密码
-    password: xxhy@2024
+    password: MjJUyaRZ_HDgZ2w_6XxK
     # 连接超时时间
     timeout: 10s
     lettuce:

ruoyi-system/pom.xml

@@ -22,7 +22,12 @@
             <groupId>com.ruoyi</groupId>
             <artifactId>ruoyi-common</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+            <version>2.0.4</version>
+        </dependency>
 
     </dependencies>
 
-</project>
+</project>

ruoyi-system/src/main/java/com/ruoyi/business/domain/MqDataInfo.java

@@ -0,0 +1,42 @@
+package com.ruoyi.business.domain;
+
+public class MqDataInfo{
+
+    private String content;
+    private String channel;
+
+    private BusOrderInfo busOrderInfo;
+    private BusAgentInfo agentInfo;
+
+    public BusAgentInfo getAgentInfo() {
+        return agentInfo;
+    }
+
+    public void setAgentInfo(BusAgentInfo agentInfo) {
+        this.agentInfo = agentInfo;
+    }
+
+    public BusOrderInfo getBusOrderInfo() {
+        return busOrderInfo;
+    }
+
+    public void setBusOrderInfo(BusOrderInfo busOrderInfo) {
+        this.busOrderInfo = busOrderInfo;
+    }
+
+    public String getContent() {
+        return content;
+    }
+
+    public void setContent(String content) {
+        this.content = content;
+    }
+
+    public String getChannel() {
+        return channel;
+    }
+
+    public void setChannel(String channel) {
+        this.channel = channel;
+    }
+}

ruoyi-system/src/main/java/com/ruoyi/business/service/IAiService.java

@@ -16,4 +16,6 @@ public interface IAiService {
     void sendToAi();
 
     void sms();
+
+    void sendSmsTask(String content, BusOrderInfo busOrderInfo);
 }

ruoyi-system/src/main/java/com/ruoyi/business/service/impl/AiConsumer.java

@@ -0,0 +1,72 @@
+package com.ruoyi.business.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.ruoyi.business.domain.*;
+import com.ruoyi.business.mapper.BusReturnVisitInfoMapper;
+import com.ruoyi.business.service.IAiService;
+import com.ruoyi.business.service.IBusAgentInfoService;
+import com.ruoyi.business.service.IBusIntegralRecordService;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+import java.util.Date;
+import java.util.List;
+
+@Component
+public class AiConsumer {
+
+    @Autowired
+    private IAiService aiService;
+    @Autowired
+    private IBusAgentInfoService agentInfoService;
+    @Autowired
+    private IBusIntegralRecordService integralRecordService;
+    @Autowired
+    private BusReturnVisitInfoMapper busReturnVisitInfoMapper;
+
+    @Service
+    @RocketMQMessageListener(topic="ai",selectorExpression = "tag1",consumerGroup = "abc")
+    public class consumer implements RocketMQListener<MqDataInfo> {
+
+        @Override
+        public void onMessage(MqDataInfo mqDataInfo) {
+            BusReturnVisitInfo returnVisitInfo = new BusReturnVisitInfo();
+            System.out.println("==============================="+returnVisitInfo);
+            returnVisitInfo.setPhoneNumber(mqDataInfo.getBusOrderInfo().getPrivcyPhone());
+            if("AI".equals(mqDataInfo.getChannel())){
+//                aiService.sendAiTask(mqDataInfo.getContent(),mqDataInfo.getBusOrderInfo());
+                returnVisitInfo.setIsCharging("AI");
+            }else if("sms".equals(mqDataInfo.getChannel())){
+//                aiService.sendSmsTask(mqDataInfo.getContent(),mqDataInfo.getBusOrderInfo());
+                BusAgentInfo busAgentInfo = mqDataInfo.getAgentInfo();
+                int a = 1;
+                int reduce = busAgentInfo.getAiDurationBalance() - a;
+                busAgentInfo.setAiDurationBalance(reduce);
+                agentInfoService.updateBusAgentInfo(busAgentInfo);
+
+                BusIntegralRecord record = new BusIntegralRecord();
+                record.setAccount(busAgentInfo.getAgentAccount());
+                record.setIntegralUpdateTime(new Date());
+                record.setUpdateType("消耗账分");
+                record.setIntegralUpdateNum(a);
+                record.setIntegralBalance(reduce);
+                record.setRemark(mqDataInfo.getBusOrderInfo().getPrivcyPhone());
+                record.setAgentRemark("sms");
+                integralRecordService.insertBusIntegralRecord(record);
+                returnVisitInfo.setIsCharging("sms");
+            }else{
+                System.out.println("未配置营销渠道=====================");
+            }
+            List<BusReturnVisitInfo> list2 = busReturnVisitInfoMapper.selectBusReturnVisitInfoList(returnVisitInfo);
+            if (list2.size() > 0) {
+                System.out.println("修改营销事件=====================");
+                returnVisitInfo.setId(list2.get(0).getId());
+                returnVisitInfo.setReturnVisitTime(new Date());
+                busReturnVisitInfoMapper.updateBusReturnVisitInfo(returnVisitInfo);
+            }
+        }
+    }
+}

ruoyi-system/src/main/java/com/ruoyi/business/service/impl/AiProducter.java

@@ -0,0 +1,18 @@
+package com.ruoyi.business.service.impl;
+
+import com.ruoyi.business.domain.MqDataInfo;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class AiProducter {
+
+
+    @Autowired
+    private RocketMQTemplate rocketMQTemplate;
+
+    public void sendInfo (MqDataInfo mqDataInfo){
+        rocketMQTemplate.convertAndSend("ai"+":tag1",mqDataInfo);
+    }
+}

ruoyi-system/src/main/java/com/ruoyi/business/service/impl/AiServiceImpl.java

@@ -8,7 +8,13 @@ import com.ruoyi.business.mapper.BusReturnVisitInfoMapper;
 import com.ruoyi.business.service.*;
 import com.ruoyi.common.config.BusTencentProperties;
 import com.ruoyi.common.utils.HttpClientUtilT;
+import com.ruoyi.common.utils.StringUtils;
 import com.ruoyi.system.mapper.SysDictDataMapper;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.MQProducer;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
@@ -21,6 +27,7 @@ import java.time.LocalTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * 代理管理Service业务层处理
@@ -55,6 +62,8 @@ public class AiServiceImpl implements IAiService {
 
     @Autowired
     private SysDictDataMapper sysDictDataMapper;
+    @Autowired
+    private AiProducter aiProducer;
 
     private String userSn = "SYSUSER|8398f13f3feccef770ee4d465fe22fbf";
     private String aiUserSn = "SYSUSER|2665bcca63a2b5c724095fa01dce0470";
@@ -266,7 +275,8 @@ public class AiServiceImpl implements IAiService {
                 }
                 agentInfoService.updateBusAgentInfo(agentInfoL.get(0));
             } else continue;
-            for (BusOrderInfo busOrderInfo : list) {
+            List<BusOrderInfo> newList = list.stream().distinct().collect(Collectors.toList());
+            for (BusOrderInfo busOrderInfo : newList) {
                 // 拼好饭
                 if ("N".equals(busStoreConfigInfo.getIsSpliceOrder())) { // 否：开启拼好饭订单过滤(不营销拼好饭)
                     if ("2".equals(busOrderInfo.getIsPhf())) {//否：
@@ -274,7 +284,11 @@ public class AiServiceImpl implements IAiService {
                         continue;
                     }
                 }
+
                 // 下单次数
+                if (StringUtils.isEmpty(busOrderInfo.getOrderCount())){
+                    continue;
+                }
                 int count = Integer.valueOf(busOrderInfo.getOrderCount());
                 if (busStoreConfigInfo.getMaxOrderNum() < count
                         || count < busStoreConfigInfo.getMinOrderNum()) {
@@ -355,33 +369,8 @@ public class AiServiceImpl implements IAiService {
                 BusFifteenInfo fifteenInfo = new BusFifteenInfo();
                 fifteenInfo.setCustomId(busOrderInfo.getCustomId());
                 List<BusFifteenInfo> listInfo = fifteenInfoService.selectBusFifteenInfoList(fifteenInfo);
-                if (listInfo.size() == 0) {
-                    System.out.println("话术：" + content + "; order" + busOrderInfo.toString());
-                    if("AI".equals(busStoreInfo.getMarketingChannels())){
-                        sendAiTask(content,busOrderInfo);
-                        returnVisitInfo.setIsCharging("AI");
-                    }else if("sms".equals(busStoreInfo.getMarketingChannels())){
-                        sendSmsTask(content,busOrderInfo);
-                        BusAgentInfo busAgentInfo = agentInfoL.get(0);
-                        int a = 1;
-                        int reduce = busAgentInfo.getAiDurationBalance() - a;
-                        busAgentInfo.setAiDurationBalance(reduce);
-                        agentInfoService.updateBusAgentInfo(busAgentInfo);
 
-                        BusIntegralRecord record = new BusIntegralRecord();
+                if (listInfo.size() != 0) {
-                        record.setAccount(busAgentInfo.getAgentAccount());
-                        record.setIntegralUpdateTime(new Date());
-                        record.setUpdateType("消耗账分");
-                        record.setIntegralUpdateNum(a);
-                        record.setIntegralBalance(reduce);
-                        record.setRemark(busOrderInfo.getPrivcyPhone());
-                        record.setAgentRemark("sms");
-                        integralRecordService.insertBusIntegralRecord(record);
-                        returnVisitInfo.setIsCharging("sms");
-                    }else{
-                        System.out.println("未配置营销渠道=====================");
-                    }
-                }else {
                     System.out.println("15天拦截=====================");
                     BusAgentInfo busAgentInfo = agentInfoL.get(0);
                     int a = 1;
@@ -399,14 +388,80 @@ public class AiServiceImpl implements IAiService {
                     record.setAgentRemark("其他");
                     integralRecordService.insertBusIntegralRecord(record);
                     returnVisitInfo.setIsCharging("其他");
+
+                    List<BusReturnVisitInfo> list2 = busReturnVisitInfoMapper.selectBusReturnVisitInfoList(returnVisitInfo);
+                    if (list2.size() > 0) {
+                        System.out.println("修改营销事件=====================");
+                        returnVisitInfo.setId(list2.get(0).getId());
+                        returnVisitInfo.setReturnVisitTime(new Date());
+                        busReturnVisitInfoMapper.updateBusReturnVisitInfo(returnVisitInfo);
+                   }
+
+                }else {
+                    MqDataInfo mqDataInfo = new MqDataInfo();
+                    mqDataInfo.setChannel(busStoreInfo.getMarketingChannels());
+                    mqDataInfo.setContent(content);
+                    mqDataInfo.setBusOrderInfo(busOrderInfo);
+                    mqDataInfo.setAgentInfo(agentInfoL.get(0));
+                    try {
+                        aiProducer.sendInfo(mqDataInfo);
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
                 }
-                List<BusReturnVisitInfo> list2 = busReturnVisitInfoMapper.selectBusReturnVisitInfoList(returnVisitInfo);
+
-                if (list2.size() > 0) {
+//                if (listInfo.size() == 0) {
-                    System.out.println("修改营销事件=====================");
+//                    System.out.println("话术：" + content + "; order" + busOrderInfo.toString());
-                    returnVisitInfo.setId(list2.get(0).getId());
+//                    if("AI".equals(busStoreInfo.getMarketingChannels())){
-                    returnVisitInfo.setReturnVisitTime(new Date());
+//                        sendAiTask(content,busOrderInfo);
-                    busReturnVisitInfoMapper.updateBusReturnVisitInfo(returnVisitInfo);
+//                        returnVisitInfo.setIsCharging("AI");
-                }
+//                    }else if("sms".equals(busStoreInfo.getMarketingChannels())){
+//                        sendSmsTask(content,busOrderInfo);
+//                        BusAgentInfo busAgentInfo = agentInfoL.get(0);
+//                        int a = 1;
+//                        int reduce = busAgentInfo.getAiDurationBalance() - a;
+//                        busAgentInfo.setAiDurationBalance(reduce);
+//                        agentInfoService.updateBusAgentInfo(busAgentInfo);
+//
+//                        BusIntegralRecord record = new BusIntegralRecord();
+//                        record.setAccount(busAgentInfo.getAgentAccount());
+//                        record.setIntegralUpdateTime(new Date());
+//                        record.setUpdateType("消耗账分");
+//                        record.setIntegralUpdateNum(a);
+//                        record.setIntegralBalance(reduce);
+//                        record.setRemark(busOrderInfo.getPrivcyPhone());
+//                        record.setAgentRemark("sms");
+//                        integralRecordService.insertBusIntegralRecord(record);
+//                        returnVisitInfo.setIsCharging("sms");
+//                    }else{
+//                        System.out.println("未配置营销渠道=====================");
+//                    }
+//                }else {
+//                    System.out.println("15天拦截=====================");
+//                    BusAgentInfo busAgentInfo = agentInfoL.get(0);
+//                    int a = 1;
+//                    int reduce = busAgentInfo.getAiDurationBalance() - a;
+//                    busAgentInfo.setAiDurationBalance(reduce);
+//                    agentInfoService.updateBusAgentInfo(busAgentInfo);
+//
+//                    BusIntegralRecord record = new BusIntegralRecord();
+//                    record.setAccount(busAgentInfo.getAgentAccount());
+//                    record.setIntegralUpdateTime(new Date());
+//                    record.setUpdateType("消耗账分");
+//                    record.setIntegralUpdateNum(a);
+//                    record.setIntegralBalance(reduce);
+//                    record.setRemark(busOrderInfo.getPrivcyPhone());
+//                    record.setAgentRemark("其他");
+//                    integralRecordService.insertBusIntegralRecord(record);
+//                    returnVisitInfo.setIsCharging("其他");
+//                }
+//                List<BusReturnVisitInfo> list2 = busReturnVisitInfoMapper.selectBusReturnVisitInfoList(returnVisitInfo);
+//                if (list2.size() > 0) {
+//                    System.out.println("修改营销事件=====================");
+//                    returnVisitInfo.setId(list2.get(0).getId());
+//                    returnVisitInfo.setReturnVisitTime(new Date());
+//                    busReturnVisitInfoMapper.updateBusReturnVisitInfo(returnVisitInfo);
+//                }
             }
         }
         System.out.println("结束===");
@@ -429,7 +484,8 @@ public class AiServiceImpl implements IAiService {
         }
     }
 
-    private synchronized void sendSmsTask(String content, BusOrderInfo busOrderInfo) {
+    @Override
+    public synchronized void sendSmsTask(String content, BusOrderInfo busOrderInfo) {
         try {
             String phoneNo = busOrderInfo.getPrivcyPhone().split(",")[0];//#分机号#
             content = content.replaceAll("分机号",busOrderInfo.getPrivcyPhone().split(",")[1]);
@@ -488,6 +544,7 @@ public class AiServiceImpl implements IAiService {
 
     }
 
+
     @Transactional
     void reduceIntegral(BusAiInfo aiInfo) throws ParseException {
 //        BusAiInfo aiInfo = aiInfoService.selectBusAiInfoById(aiId);