From ab3b53c7e8f3a12b54e24c0a3b9ad0a74c2a6177 Mon Sep 17 00:00:00 2001 From: hurui <177768073@qq.com> Date: Mon, 12 Aug 2024 14:59:02 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E4=B8=89=E6=9C=A8=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E5=8A=9F=E8=83=BD=E3=80=81=E6=B2=B9=E4=BB=B7=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E4=B8=8D=E5=87=86=E6=97=B6=E4=BC=98=E5=8C=96=EF=BC=8C?= =?UTF-8?q?=E6=9B=B4=E6=96=B0rocketmq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../hfkj/consumer/TaskOilPriceConsumer.java | 35 ++++++++++++++++++ .../src/main/resources/dev/application.yml | 14 ++++++++ .../src/main/resources/prod/application.yml | 14 ++++++++ service/pom.xml | 2 +- .../main/java/com/hfkj/mqtopic/TaskTopic.java | 26 ++++++++++++++ .../impl/BsGasOilPriceTaskServiceImpl.java | 36 +++++++++++++------ 6 files changed, 116 insertions(+), 11 deletions(-) create mode 100644 schedule/src/main/java/com/hfkj/consumer/TaskOilPriceConsumer.java create mode 100644 service/src/main/java/com/hfkj/mqtopic/TaskTopic.java diff --git a/schedule/src/main/java/com/hfkj/consumer/TaskOilPriceConsumer.java b/schedule/src/main/java/com/hfkj/consumer/TaskOilPriceConsumer.java new file mode 100644 index 0000000..42e118b --- /dev/null +++ b/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 { + + @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); + } + } +} diff --git a/schedule/src/main/resources/dev/application.yml b/schedule/src/main/resources/dev/application.yml index 9ade586..911278c 100644 --- a/schedule/src/main/resources/dev/application.yml +++ b/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: diff --git a/schedule/src/main/resources/prod/application.yml b/schedule/src/main/resources/prod/application.yml index 590f6de..382e69e 100644 --- a/schedule/src/main/resources/prod/application.yml +++ b/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: diff --git a/service/pom.xml b/service/pom.xml index a19e7a1..2a46783 100644 --- a/service/pom.xml +++ b/service/pom.xml @@ -262,7 +262,7 @@ org.apache.rocketmq rocketmq-spring-boot-starter - 2.2.2 + 2.2.3 com.alipay.sdk diff --git a/service/src/main/java/com/hfkj/mqtopic/TaskTopic.java b/service/src/main/java/com/hfkj/mqtopic/TaskTopic.java new file mode 100644 index 0000000..2f376c1 --- /dev/null +++ b/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; + } +} diff --git a/service/src/main/java/com/hfkj/service/gas/impl/BsGasOilPriceTaskServiceImpl.java b/service/src/main/java/com/hfkj/service/gas/impl/BsGasOilPriceTaskServiceImpl.java index 7c017ee..4bb68eb 100644 --- a/service/src/main/java/com/hfkj/service/gas/impl/BsGasOilPriceTaskServiceImpl.java +++ b/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()); } }