完成三木设备功能、油价任务不准时优化，更新rocketmq

schedule/src/main/java/com/hfkj/consumer/TaskOilPriceConsumer.java

@@ -0,0 +1,35 @@
+package com.hfkj.consumer;
+
+import com.hfkj.entity.BsGasOilPriceTask;
+import com.hfkj.entity.BsOrder;
+import com.hfkj.model.order.OrderModel;
+import com.hfkj.service.gas.BsGasOilPriceService;
+import com.hfkj.service.gas.BsGasOilPriceTaskService;
+import com.hfkj.service.order.BsOrderService;
+import com.hfkj.sysenum.gas.GasTaskStatusEnum;
+import com.hfkj.sysenum.order.OrderStatusEnum;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Component
+@Slf4j
+@RocketMQMessageListener(consumerGroup = "task-oil-price-group", topic = "task-topic",selectorExpression = "oil-price")
+public class TaskOilPriceConsumer implements RocketMQListener<String> {
+
+    @Resource
+    private BsGasOilPriceTaskService gasOilPriceTaskService;
+
+    @Override
+    public void onMessage(String oilPriceTaskId) {
+        // 油价任务
+        BsGasOilPriceTask task = gasOilPriceTaskService.getDetailById(Long.parseLong(oilPriceTaskId));
+        if (task != null && task.getStatus().equals(GasTaskStatusEnum.status1.getStatus())) {
+            // 任务处理
+            gasOilPriceTaskService.businessHandle(task);
+        }
+    }
+}

schedule/src/main/resources/dev/application.yml

@@ -39,6 +39,20 @@ spring:
         max-wait: -1
         max-idle: 10
         min-idle: 0
+  #MQTT配置信息
+  mqtt:
+    #MQTT服务地址
+    url: ws://139.9.154.68:8083/mqtt
+    #用户名
+    username: printer_provider
+    #密码
+    password: 123654
+    #客户端id(不能重复)
+    # client:
+    # id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    # default:
+    # topic: topic
   #配置日期返回至前台为时间戳
   jackson:
     serialization:

schedule/src/main/resources/prod/application.yml

@@ -39,6 +39,20 @@ spring:
         max-wait: -1
         max-idle: 10
         min-idle: 0
+  #MQTT配置信息
+  mqtt:
+    #MQTT服务地址
+    url: ws://1.95.43.71:8083/mqtt
+    #用户名
+    username: printer_provider
+    #密码
+    password: abcd123!
+    #客户端id(不能重复)
+    # client:
+    # id: provider-id
+    #MQTT默认的消息推送主题，实际可在调用接口是指定
+    # default:
+    # topic: topic
   #配置日期返回至前台为时间戳
   jackson:
     serialization:

service/pom.xml

@@ -262,7 +262,7 @@
         <dependency>
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-spring-boot-starter</artifactId>
-            <version>2.2.2</version>
+            <version>2.2.3</version>
         </dependency>
         <dependency>
             <groupId>com.alipay.sdk</groupId>

service/src/main/java/com/hfkj/mqtopic/TaskTopic.java

@@ -0,0 +1,26 @@
+package com.hfkj.mqtopic;
+
+import lombok.Getter;
+
+/**
+ * 任务主题
+ * @author hurui
+ */
+@Getter
+public enum TaskTopic {
+    // 任务主题
+    TASK_TOPIC("task-topic", "","任务主题"),
+    // 油价任务
+    OIL_PRICE(TASK_TOPIC.getTopic(),"oil-price", "油价任务"),
+    ;
+
+    private String topic;
+    private String tag;
+    private String name;
+
+    TaskTopic(String topic, String tag, String name) {
+        this.topic = topic;
+        this.tag = tag;
+        this.name = name;
+    }
+}

service/src/main/java/com/hfkj/service/gas/impl/BsGasOilPriceTaskServiceImpl.java

@@ -7,9 +7,9 @@ import com.hfkj.common.utils.DateUtil;
 import com.hfkj.common.utils.RedisUtil;
 import com.hfkj.dao.BsGasOilPriceTaskMapper;
 import com.hfkj.entity.BsGasOilPrice;
-import com.hfkj.entity.BsGasOilPriceOfficial;
 import com.hfkj.entity.BsGasOilPriceTask;
 import com.hfkj.entity.BsGasOilPriceTaskExample;
+import com.hfkj.mqtopic.TaskTopic;
 import com.hfkj.msg.MsgTopic;
 import com.hfkj.service.gas.BsGasOilPriceOfficialService;
 import com.hfkj.service.gas.BsGasOilPriceService;
@@ -19,6 +19,11 @@ import com.hfkj.sysenum.gas.GasTaskPriceTypeEnum;
 import com.hfkj.sysenum.gas.GasTaskStatusEnum;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -35,6 +40,8 @@ public class BsGasOilPriceTaskServiceImpl implements BsGasOilPriceTaskService {
 
     @Resource
     private RedisUtil redisUtil;
+    @Autowired
+    private RocketMQTemplate rocketMQTemplate;
     @Resource
     private BsGasOilPriceTaskMapper gasOilPriceTaskMapper;
     @Resource
@@ -66,17 +73,26 @@ public class BsGasOilPriceTaskServiceImpl implements BsGasOilPriceTaskService {
     @Transactional(propagation= Propagation.REQUIRED)
     public void addTask(BsGasOilPriceTask gasOilPriceTask) {
         editData(gasOilPriceTask);
-        // 立刻执行
-        if (gasOilPriceTask.getExecutionType().equals(GasTaskExecutionTypeEnum.type1.getStatus())) {
-            businessHandle(gasOilPriceTask);
-
-        } else if (gasOilPriceTask.getExecutionType().equals(GasTaskExecutionTypeEnum.type2.getStatus())) {
-            long time = DateUtil.getSecondDiff(new Date(), gasOilPriceTask.getStartTime());
-            if (time >= 1) {
-                redisUtil.set(MsgTopic.oilPriceTask.getName() + "-" + gasOilPriceTask.getId(), "", time);
-            } else {
+        try {
+            // 立刻执行
+            if (gasOilPriceTask.getExecutionType().equals(GasTaskExecutionTypeEnum.type1.getStatus())) {
                 businessHandle(gasOilPriceTask);
+
+            } else if (gasOilPriceTask.getExecutionType().equals(GasTaskExecutionTypeEnum.type2.getStatus())) {
+                long time = DateUtil.getSecondDiff(new Date(), gasOilPriceTask.getStartTime());
+                if (time >= 1) {
+                    Message message = new Message(TaskTopic.TASK_TOPIC.getTopic(),TaskTopic.OIL_PRICE.getTag(),gasOilPriceTask.getId().toString().getBytes());
+                    // 定时投递，执行时间
+                    message.setDeliverTimeMs(gasOilPriceTask.getStartTime().getTime());
+                    SendResult sendResult = rocketMQTemplate.getProducer().send(message);
+                    System.out.println(sendResult);
+                    // redisUtil.set(MsgTopic.oilPriceTask.getName() + "-" + gasOilPriceTask.getId(), "", time);
+                } else {
+                    businessHandle(gasOilPriceTask);
+                }
             }
+        } catch (Exception e) {
+            System.out.println(e.getMessage());
         }
     }