From 1daf485e92f70bb094fa034a56ec2db5fe668a01 Mon Sep 17 00:00:00 2001
From: liujun <liujun@zcloudchina.com>
Date: Tue, 12 Dec 2023 18:02:22 +0800
Subject: [PATCH] =?UTF-8?q?=E5=85=AB=E9=A1=B9=E4=BD=9C=E4=B8=9Amq=E5=8A=9F?=
 =?UTF-8?q?=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pom.xml                                       |  15 ++
 src/main/java/com/zcloud/dto/mq/Response.java |  68 +++++++++
 .../java/com/zcloud/dto/mq/TenCorpDto.java    |  46 +++++++
 .../mq/controller/EightWorkListener.java      |  87 ++++++++++++
 .../com/zcloud/mq/controller/MqListener.java  |  32 +++++
 .../zcloud/mq/controller/SendController.java  |  54 ++++++++
 .../mq/mapper/MqErrorMessageLogMapper.java    |  59 ++++++++
 .../zcloud/mq/mapper/MqMessageLogMapper.java  |  59 ++++++++
 .../com/zcloud/mq/service/LogService.java     |  11 ++
 .../zcloud/mq/service/SendMessageService.java |  11 ++
 .../mq/service/impl/LogServiceImpl.java       |  43 ++++++
 .../service/impl/SendMessageServiceImpl.java  | 129 ++++++++++++++++++
 src/main/java/com/zcloud/mq/util/Const.java   |   6 +
 src/main/java/com/zcloud/mq/util/MqUtil.java  |  84 ++++++++++++
 14 files changed, 704 insertions(+)
 create mode 100644 src/main/java/com/zcloud/dto/mq/Response.java
 create mode 100644 src/main/java/com/zcloud/dto/mq/TenCorpDto.java
 create mode 100644 src/main/java/com/zcloud/mq/controller/EightWorkListener.java
 create mode 100644 src/main/java/com/zcloud/mq/controller/MqListener.java
 create mode 100644 src/main/java/com/zcloud/mq/controller/SendController.java
 create mode 100644 src/main/java/com/zcloud/mq/mapper/MqErrorMessageLogMapper.java
 create mode 100644 src/main/java/com/zcloud/mq/mapper/MqMessageLogMapper.java
 create mode 100644 src/main/java/com/zcloud/mq/service/LogService.java
 create mode 100644 src/main/java/com/zcloud/mq/service/SendMessageService.java
 create mode 100644 src/main/java/com/zcloud/mq/service/impl/LogServiceImpl.java
 create mode 100644 src/main/java/com/zcloud/mq/service/impl/SendMessageServiceImpl.java
 create mode 100644 src/main/java/com/zcloud/mq/util/Const.java
 create mode 100644 src/main/java/com/zcloud/mq/util/MqUtil.java

diff --git a/pom.xml b/pom.xml
index a046f419..6576b988 100644
--- a/pom.xml
+++ b/pom.xml
@@ -461,6 +461,21 @@
 			<artifactId>hutool-all</artifactId>
 			<version>5.0.5</version>
 		</dependency>
+
+		<dependency>
+			<groupId>org.apache.rocketmq</groupId>
+			<artifactId>rocketmq-client-java</artifactId>
+			<version>5.0.5</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.rocketmq</groupId>
+			<artifactId>rocketmq-spring-boot-starter</artifactId>
+			<version>2.2.2</version>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+		</dependency>
 	</dependencies>
 	<build>
 		<!-- <finalName>qa-prevention-gwj</finalName>
diff --git a/src/main/java/com/zcloud/dto/mq/Response.java b/src/main/java/com/zcloud/dto/mq/Response.java
new file mode 100644
index 00000000..0e9ba3dc
--- /dev/null
+++ b/src/main/java/com/zcloud/dto/mq/Response.java
@@ -0,0 +1,68 @@
+package com.zcloud.dto.mq;
+
+import com.alibaba.fastjson.JSON;
+import lombok.Data;
+
+@Data
+public class Response {
+    private String result;
+
+    private String exception;
+
+    private String code;
+
+    private String message;
+
+    public Response Ok() {
+        this.result = "succeed";
+        this.code = "0";
+        return this;
+    }
+
+    public static Response OK() {
+        Response response = new Response();
+        response.result = "succeed";
+        response.code = "0";
+        return response;
+    }
+
+    public Response Error() {
+        this.result = "succeed";
+        this.code = "9999";
+        this.exception = "系统异常";
+        return this;
+    }
+
+    public Response Error(String errorMessage) {
+        this.result = "succeed";
+        this.code = "9999";
+        this.exception = errorMessage;
+        return this;
+    }
+
+    public Response Error(String code, String errorMessage) {
+        this.result = "succeed";
+        this.code = code;
+        this.exception = errorMessage;
+        return this;
+    }
+
+    public static Response ERROR() {
+        Response response = new Response();
+        response.result = "succeed";
+        response.code = "0";
+        return response;
+    }
+
+    public static Response ERROR(String errorMessage) {
+        Response response = new Response();
+        response.result = "succeed";
+        response.code = "9999";
+        response.exception = errorMessage;
+        return response;
+    }
+
+    public void setMessage(Object obj) {
+        this.message = JSON.toJSONString(obj);
+    }
+}
diff --git a/src/main/java/com/zcloud/dto/mq/TenCorpDto.java b/src/main/java/com/zcloud/dto/mq/TenCorpDto.java
new file mode 100644
index 00000000..ca168261
--- /dev/null
+++ b/src/main/java/com/zcloud/dto/mq/TenCorpDto.java
@@ -0,0 +1,46 @@
+package com.zcloud.dto.mq;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.zcloud.entity.PageData;
+import com.zcloud.util.Warden;
+import lombok.Data;
+
+@Data
+public class TenCorpDto {
+    // 请秋id
+    private String id;
+    // 消息来源名称
+    private String producer_name;
+    private String topic;
+    //印记(方便寻找该条请求)
+    private String mark;
+    // 标记名称
+    private String mark_name;
+    // 消息发送时间(yyyy-MM-dd HH:mm:ss:SSS)
+    private String time_stamp;
+    // 消息体
+    private String message;
+
+    private PageData data;
+
+    public TenCorpDto() {
+        this.id = Warden.get32UUID();
+    }
+
+    @Override
+    public String toString() {
+        return JSON.toJSONString(this);
+    }
+
+    public PageData getPd() {
+        PageData info = new PageData();
+        info.put("ID", this.id);
+        info.put("PRODUCER_NAME", this.producer_name);
+        info.put("MARK", this.mark);
+        info.put("MARK_NAME", this.mark_name);
+        info.put("MESSAGE", JSONObject.toJSONString(this.data));
+        info.put("TOPIC",this.topic);
+        return info;
+    }
+}
diff --git a/src/main/java/com/zcloud/mq/controller/EightWorkListener.java b/src/main/java/com/zcloud/mq/controller/EightWorkListener.java
new file mode 100644
index 00000000..83e5dcb5
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/controller/EightWorkListener.java
@@ -0,0 +1,87 @@
+package com.zcloud.mq.controller;
+
+import com.alibaba.fastjson.JSON;
+import com.zcloud.controller.gf.GFEightWorkController;
+import com.zcloud.entity.PageData;
+import com.zcloud.mq.mapper.MqMessageLogMapper;
+import com.zcloud.dto.mq.TenCorpDto;
+import com.zcloud.mq.service.LogService;
+import com.zcloud.mq.service.SendMessageService;
+import com.zcloud.mq.util.MqUtil;
+import com.zcloud.util.DateUtil;
+import com.zcloud.util.Warden;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.Date;
+
+/**
+ * 说明：消费某个 topic 下指定 tag 的消息 对接课件DEMO
+ * 作者：wangxuan
+ * 官网：www.zcloudchina.com
+ */
+@Component
+@RocketMQMessageListener(consumerGroup = "${mq.group.eightWork}",
+        topic = "${mq.topic.eightWork}",
+        selectorType = SelectorType.TAG)
+public class EightWorkListener implements RocketMQListener<String> {
+
+    @Resource
+    private GFEightWorkController eightWorkClient;
+
+    @Resource
+    private SendMessageService sendMessageService;
+
+    @Resource
+    private LogService logService;
+
+    @Resource
+    private MqMessageLogMapper mqMessageLogMapper;
+
+    @Override
+    public void onMessage(String message) {
+        TenCorpDto data = JSON.parseObject(message, TenCorpDto.class);
+        PageData log = data.getPd();
+        log.put("MESSAGE_LOG_ID", Warden.get32UUID());
+        log.put("CREATE_TIME", DateUtil.getTime());
+        log.put("PRODUCER_NAME",data.getProducer_name());
+        log.put("TYPE","1");
+        try {
+            System.out.println("消费者:" + data.toString());
+            Date limit_date = MqUtil.dateFormat(data.getTime_stamp());
+            if (limit_date.getTime() - new Date().getTime() > 1000) {
+                sendMessageService.SendDelayQueue(data);
+                log.put("DIGESTION_FLAG","0");
+            } else {
+                switch (data.getMessage()){
+                    case "confined-space":
+                        System.out.println("受限空间模块消费");
+//                        eightWorkClient.confinedSpaceCancel(data.getData());
+                        break;
+                    case "electricity":
+                        System.out.println("临时用电模块消费");
+//                        eightWorkClient.electricityCancel(data.getData());
+                        break;
+                    case "hot-work":
+                        System.out.println("动火模块消费");
+//                        eightWorkClient.hotWorkCancel(data.getData());
+                        break;
+                    default:
+                        System.out.println("异常模块不消费");
+                        throw new RuntimeException("未找到对应的消费者");
+                }
+                log.put("DIGESTION_FLAG","1");
+            }
+            mqMessageLogMapper.save(log);
+        } catch (Exception e) {
+            e.printStackTrace();
+            logService.saveErrorMessage(log,e.getMessage());
+            throw new RuntimeException(e);
+        }
+    }
+
+}
+
diff --git a/src/main/java/com/zcloud/mq/controller/MqListener.java b/src/main/java/com/zcloud/mq/controller/MqListener.java
new file mode 100644
index 00000000..2ca3f4e0
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/controller/MqListener.java
@@ -0,0 +1,32 @@
+package com.zcloud.mq.controller;
+
+import com.alibaba.fastjson.JSON;
+import com.zcloud.dto.mq.TenCorpDto;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * 说明：消费某个 topic 下指定 tag 的消息 对接课件DEMO
+ * 作者：wangxuan
+ * 官网：www.zcloudchina.com
+ */
+@Component
+@RocketMQMessageListener(consumerGroup = "${mq.group.info}",
+        topic = "${mq.topic.info}",
+        selectorType = SelectorType.TAG)
+public class MqListener implements RocketMQListener<String> {
+
+    @Override
+    public void onMessage(String message) {
+        TenCorpDto data = JSON.parseObject(message, TenCorpDto.class);
+        try {
+            System.out.println("2消费者:" + data.toString());
+        } catch (Exception e) {
+            e.printStackTrace();
+            throw new RuntimeException(e);
+        }
+    }
+}
+
diff --git a/src/main/java/com/zcloud/mq/controller/SendController.java b/src/main/java/com/zcloud/mq/controller/SendController.java
new file mode 100644
index 00000000..b6bbcd50
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/controller/SendController.java
@@ -0,0 +1,54 @@
+package com.zcloud.mq.controller;
+
+import com.zcloud.controller.base.BaseController;
+import com.zcloud.dto.mq.Response;
+import com.zcloud.dto.mq.TenCorpDto;
+import com.zcloud.mq.service.SendMessageService;
+import org.springframework.web.bind.annotation.*;
+
+import javax.annotation.Resource;
+
+/**
+ * 说明：TODO
+ * 作者：wangxuan
+ * 官网：www.zcloudchina.com
+ */
+@RestController
+@RequestMapping("/messageQueue")
+public class SendController extends BaseController {
+
+    @Resource
+    private SendMessageService sendMessageService;
+
+    @RequestMapping("/sendMessage")
+    public Response sendMessage(@RequestParam("tenCorpDto") TenCorpDto tenCorpDto) throws Exception {
+        try {
+            sendMessageService.sendMessage(tenCorpDto);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return Response.OK();
+    }
+
+    @PostMapping("SendTimelyQueue")
+    public Response timelyQueue(@RequestBody TenCorpDto tenCorpDto) throws Exception {
+        try {
+            sendMessageService.sendMessage(tenCorpDto);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return Response.OK();
+    }
+
+    @PostMapping("/SendDelayQueue")
+    public Response SendDelayQueue(@RequestBody TenCorpDto tenCorpDto) throws Exception {
+        try {
+            sendMessageService.SendDelayQueue(tenCorpDto);
+            return Response.OK();
+        } catch (Exception e) {
+            e.printStackTrace();
+            return Response.ERROR(e.getMessage());
+        }
+
+    }
+}
diff --git a/src/main/java/com/zcloud/mq/mapper/MqErrorMessageLogMapper.java b/src/main/java/com/zcloud/mq/mapper/MqErrorMessageLogMapper.java
new file mode 100644
index 00000000..68dd153f
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/mapper/MqErrorMessageLogMapper.java
@@ -0,0 +1,59 @@
+package com.zcloud.mq.mapper;
+
+import com.zcloud.entity.Page;
+import com.zcloud.entity.PageData;
+
+import java.util.List;
+
+/**
+ * 说明：mq错误信息日志表
+ * 作者：luoxiaobao
+ * 时间：2023-06-28
+ * 官网：www.zcloudchina.com
+ */
+public interface MqErrorMessageLogMapper{
+
+	/**新增
+	 * @param pd
+	 * @throws Exception
+	 */
+	void save(PageData pd);
+
+	/**删除
+	 * @param pd
+	 * @throws Exception
+	 */
+	void delete(PageData pd);
+
+	/**修改
+	 * @param pd
+	 * @throws Exception
+	 */
+	void edit(PageData pd);
+
+	/**列表
+	 * @param page
+	 * @throws Exception
+	 */
+	List<PageData> datalistPage(Page page);
+
+	/**列表(全部)
+	 * @param pd
+	 * @throws Exception
+	 */
+	List<PageData> listAll(PageData pd);
+
+	/**通过id获取数据
+	 * @param pd
+	 * @throws Exception
+	 */
+	PageData findById(PageData pd);
+
+	/**批量删除
+	 * @param ArrayDATA_IDS
+	 * @throws Exception
+	 */
+	void deleteAll(String[] ArrayDATA_IDS);
+
+}
+
diff --git a/src/main/java/com/zcloud/mq/mapper/MqMessageLogMapper.java b/src/main/java/com/zcloud/mq/mapper/MqMessageLogMapper.java
new file mode 100644
index 00000000..bd586c64
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/mapper/MqMessageLogMapper.java
@@ -0,0 +1,59 @@
+package com.zcloud.mq.mapper;
+
+import com.zcloud.entity.Page;
+import com.zcloud.entity.PageData;
+
+import java.util.List;
+
+/**
+ * 说明：mq消费者\生产者动作记录表
+ * 作者：luoxiaobao
+ * 时间：2023-06-28
+ * 官网：www.zcloudchina.com
+ */
+public interface MqMessageLogMapper{
+
+	/**新增
+	 * @param pd
+	 * @throws Exception
+	 */
+	void save(PageData pd);
+
+	/**删除
+	 * @param pd
+	 * @throws Exception
+	 */
+	void delete(PageData pd);
+
+	/**修改
+	 * @param pd
+	 * @throws Exception
+	 */
+	void edit(PageData pd);
+
+	/**列表
+	 * @param page
+	 * @throws Exception
+	 */
+	List<PageData> datalistPage(Page page);
+
+	/**列表(全部)
+	 * @param pd
+	 * @throws Exception
+	 */
+	List<PageData> listAll(PageData pd);
+
+	/**通过id获取数据
+	 * @param pd
+	 * @throws Exception
+	 */
+	PageData findById(PageData pd);
+
+	/**批量删除
+	 * @param ArrayDATA_IDS
+	 * @throws Exception
+	 */
+	void deleteAll(String[] ArrayDATA_IDS);
+
+}
+
diff --git a/src/main/java/com/zcloud/mq/service/LogService.java b/src/main/java/com/zcloud/mq/service/LogService.java
new file mode 100644
index 00000000..1c963440
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/service/LogService.java
@@ -0,0 +1,11 @@
+package com.zcloud.mq.service;
+
+import com.zcloud.entity.PageData;
+
+public interface LogService {
+
+    void saveLog(PageData pageData);
+    void saveLog(PageData pageData,String type);
+
+    void saveErrorMessage(PageData pageData,String errorMessage);
+}
diff --git a/src/main/java/com/zcloud/mq/service/SendMessageService.java b/src/main/java/com/zcloud/mq/service/SendMessageService.java
new file mode 100644
index 00000000..1a49f784
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/service/SendMessageService.java
@@ -0,0 +1,11 @@
+package com.zcloud.mq.service;
+
+import com.zcloud.dto.mq.TenCorpDto;
+
+public interface SendMessageService {
+
+    void sendMessage(TenCorpDto tenCorpDto) throws Exception;
+
+    void SendDelayQueue(TenCorpDto tenCorpDto);
+    void SendDelayQueue(String message);
+}
diff --git a/src/main/java/com/zcloud/mq/service/impl/LogServiceImpl.java b/src/main/java/com/zcloud/mq/service/impl/LogServiceImpl.java
new file mode 100644
index 00000000..bd1222b1
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/service/impl/LogServiceImpl.java
@@ -0,0 +1,43 @@
+package com.zcloud.mq.service.impl;
+
+import com.zcloud.entity.PageData;
+import com.zcloud.mq.mapper.MqErrorMessageLogMapper;
+import com.zcloud.mq.mapper.MqMessageLogMapper;
+import com.zcloud.mq.service.LogService;
+import com.zcloud.util.DateUtil;
+import com.zcloud.util.Warden;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+@Service
+public class LogServiceImpl implements LogService {
+
+    @Resource
+    private MqErrorMessageLogMapper mqErrorMessageLogMapper;
+
+    @Resource
+    private MqMessageLogMapper mqMessageLogMapper;
+
+    @Override
+    @Async
+    public void saveLog(PageData pageData) {
+        this.saveLog(pageData,"0");
+    }
+
+    @Override
+    public void saveLog(PageData pageData, String type) {
+
+    }
+
+    @Override
+    @Async
+    public void saveErrorMessage(PageData log,String errorMessage) {
+        log.put("ERROR_MESSAGE", errorMessage);
+        log.put("MESSAGE_ERROR_LOG_ID", Warden.get32UUID());
+        log.put("TYPE", "0");
+        log.put("TIME", DateUtil.getTime());
+        mqErrorMessageLogMapper.save(log);
+    }
+}
diff --git a/src/main/java/com/zcloud/mq/service/impl/SendMessageServiceImpl.java b/src/main/java/com/zcloud/mq/service/impl/SendMessageServiceImpl.java
new file mode 100644
index 00000000..a9bccfff
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/service/impl/SendMessageServiceImpl.java
@@ -0,0 +1,129 @@
+package com.zcloud.mq.service.impl;
+
+import com.alibaba.fastjson.JSONObject;
+import com.zcloud.dto.mq.TenCorpDto;
+import com.zcloud.entity.PageData;
+import com.zcloud.mq.mapper.MqErrorMessageLogMapper;
+import com.zcloud.mq.mapper.MqMessageLogMapper;
+import com.zcloud.mq.service.LogService;
+import com.zcloud.mq.service.SendMessageService;
+import com.zcloud.mq.util.MqUtil;
+import com.zcloud.util.DateUtil;
+import com.zcloud.util.Warden;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+
+@Service
+public class SendMessageServiceImpl implements SendMessageService {
+
+    @Value("${mq.topic.info}")
+    private String info;
+
+    @Resource
+    private RocketMQTemplate rocketMQTemplate;
+
+    @Resource
+    private MqErrorMessageLogMapper mqErrorMessageLogMapper;
+
+    @Resource
+    private LogService logService;
+
+    @Resource
+    private MqMessageLogMapper mqMessageLogMapper;
+
+    public void sendMessage(TenCorpDto tenCorpDto) throws Exception {
+        try {
+            PageData log = tenCorpDto.getPd();
+            log.put("MESSAGE_LOG_ID", Warden.get32UUID());
+            log.put("CREATE_TIME", DateUtil.getTime());
+            mqMessageLogMapper.save(log);
+            System.out.println("生产者:" + tenCorpDto.toString());
+            SendResult sendResult = rocketMQTemplate.syncSend(this.info, tenCorpDto.toString());
+            if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
+                throw new RuntimeException("产品入栈失败");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            PageData log = tenCorpDto.getPd();
+            log.put("MESSAGE_ERROR_LOG_ID", Warden.get32UUID());
+            log.put("ERROR_MESSAGE", e.getMessage());
+            log.put("TYPE", "0");
+            log.put("TIME", DateUtil.getTime());
+            mqErrorMessageLogMapper.save(log);
+        }
+    }
+
+    @Override
+    public void SendDelayQueue(TenCorpDto tenCorpDto) {
+        PageData log = tenCorpDto.getPd();
+        log.put("MESSAGE_LOG_ID", Warden.get32UUID());
+        try {
+            log.put("CREATE_TIME", DateUtil.getTime());
+            log.put("PRODUCER_NAME",tenCorpDto.getProducer_name());
+            log.put("TYPE","0");
+            log.put("PLAN_TIME",tenCorpDto.getTime_stamp());
+            mqMessageLogMapper.save(log);
+            System.out.println("生产者:" + tenCorpDto.toString());
+            // 推送消息
+            SendResult sendResult = rocketMQTemplate.syncSend(
+                    MqUtil.analysistopic(tenCorpDto.getTopic()),
+                    new GenericMessage<>(tenCorpDto.toString()),
+                    3000,
+                    MqUtil.analysisTime(tenCorpDto.getTime_stamp()));
+
+            if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
+                throw new RuntimeException("产品入栈失败");
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.put("ERROR_MESSAGE", e.getMessage());
+            log.put("MESSAGE_ERROR_LOG_ID", Warden.get32UUID());
+            log.put("TYPE", "0");
+            log.put("TIME", DateUtil.getTime());
+            mqErrorMessageLogMapper.save(log);
+        }
+    }
+
+    @Override
+    public void SendDelayQueue(String message) {
+        TenCorpDto tenCorpDto = JSONObject.parseObject(message,TenCorpDto.class);
+        try {
+            PageData log = tenCorpDto.getPd();
+            log.put("CREATE_TIME", DateUtil.getTime());
+            log.put("PRODUCER_NAME",tenCorpDto.getProducer_name());
+            log.put("MESSAGE_LOG_ID", Warden.get32UUID());
+            mqMessageLogMapper.save(log);
+            System.out.println("生产者:" + tenCorpDto.toString());
+            // 推送消息
+            SendResult sendResult = rocketMQTemplate.syncSend(
+                    MqUtil.analysistopic(tenCorpDto.getTopic()),
+                    new GenericMessage<>(message),
+                    3000,
+                    MqUtil.analysisTime(tenCorpDto.getTime_stamp()));
+
+            if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
+                throw new RuntimeException("产品入栈失败");
+            }
+        } catch (Exception e) {
+            PageData log = tenCorpDto.getPd();
+            logService.saveErrorMessage(log,e.getMessage());
+            e.printStackTrace();
+        }
+    }
+
+    public static void main(String[] args) {
+        try {
+            System.out.println(MqUtil.analysisTime("2023-06-30 14:30:00:000"));
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+    }
+
+
+}
diff --git a/src/main/java/com/zcloud/mq/util/Const.java b/src/main/java/com/zcloud/mq/util/Const.java
new file mode 100644
index 00000000..2150eb63
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/util/Const.java
@@ -0,0 +1,6 @@
+package com.zcloud.mq.util;
+
+public class Const {
+    public static final String topic_eightWork="eightWork";
+    public static final String topic_info="info";
+}
diff --git a/src/main/java/com/zcloud/mq/util/MqUtil.java b/src/main/java/com/zcloud/mq/util/MqUtil.java
new file mode 100644
index 00000000..7f1ba12e
--- /dev/null
+++ b/src/main/java/com/zcloud/mq/util/MqUtil.java
@@ -0,0 +1,84 @@
+package com.zcloud.mq.util;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+public class MqUtil {
+
+    private final static SimpleDateFormat timeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
+
+    public static Integer analysisTime(String time) throws Exception{
+        Date agreed_date = timeFormat.parse(time);
+        long time_difference = agreed_date.getTime() - new Date().getTime();
+        if (time_difference < 0){
+            return 0;
+        }
+        if (time_difference > 2*60*60*1000){
+            return 18;
+        }
+        if (time_difference > 60 * 60 * 1000){
+            return 17;
+        }
+        if (time_difference > 30 * 60 * 1000){
+            return 16;
+        }
+        if (time_difference > 20 * 60 * 1000){
+            return 15;
+        }
+        if (time_difference > 10 * 60 * 1000){
+            return 14;
+        }
+        if (time_difference > 9 * 60 * 1000){
+            return 13;
+        }
+        if (time_difference > 8 * 60 * 1000){
+            return 12;
+        }
+        if (time_difference > 7 * 60 * 1000){
+            return 11;
+        }
+        if (time_difference > 6 * 60 * 1000){
+            return 10;
+        }
+        if (time_difference > 5 * 60 * 1000){
+            return 9;
+        }
+        if (time_difference > 4 * 60 * 1000){
+            return 8;
+        }
+        if (time_difference > 3 * 60 * 1000){
+            return 7;
+        }
+        if (time_difference > 2 * 60 * 1000){
+            return 6;
+        }
+        if (time_difference > 60 * 1000){
+            return 5;
+        }
+        if (time_difference > 30 * 1000){
+            return 4;
+        }
+        if (time_difference > 10 * 1000){
+            return 3;
+        }
+        if (time_difference > 5 * 1000){
+            return 2;
+        }
+        if (time_difference > 1000){
+            return 1;
+        }
+        return 0;
+    }
+
+    public static String analysistopic(String producerName) {
+        switch (producerName){
+            case "eightWork" : return Const.topic_eightWork;
+            case "info" : return Const.topic_info;
+            default: return "";
+        }
+    }
+
+    public static Date dateFormat(String date) throws Exception{
+        return timeFormat.parse(date);
+    }
+}