feat:肇庆提交，任务延迟生成
wyb
3 weeks ago
parent
4bc453269f
commit
ffb7399202
@ -0,0 +1,9 @@
|
||||
package com.docus.server.message.enums;
|
||||
|
||||
/**
|
||||
* @author YongBin Wen
|
||||
* @date 2025/6/3 0003 15:44
|
||||
*/
|
||||
public interface TaskWait {
|
||||
String SUBMIT_TASK_WAIT_DELAY_MSG_TYPE = "SUBMIT_TASK_WAIT_DELAY_MSG_TYPE";
|
||||
}
|
||||
@ -0,0 +1,25 @@
|
||||
package com.docus.server.message.mapper;
|
||||
|
||||
import com.docus.server.message.entity.DelayedMessages;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 延迟消息 mapper
|
||||
* @author wyb
|
||||
*/
|
||||
@Mapper
|
||||
public interface DelayedMessagesMapper {
|
||||
|
||||
int delById(@Param("id") Long id);
|
||||
|
||||
int updateById(@Param("message") DelayedMessages message);
|
||||
|
||||
int insert(@Param("message") DelayedMessages message);
|
||||
|
||||
List<DelayedMessages> findExecutableMessages(@Param("condition") DelayedMessages condition,@Param("currentDateTime") String formatCurrentDateTime);
|
||||
|
||||
DelayedMessages findMessage(@Param("condition") DelayedMessages condition);
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.docus.server.message.scheduler;
|
||||
|
||||
import com.docus.server.message.thread.DelayTaskIssuanceHelper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* @author YongBin Wen
|
||||
* @date 2024/1/23 14:04
|
||||
*/
|
||||
|
||||
public class JobScheduler {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(JobScheduler.class);
|
||||
|
||||
public void init() throws Exception {
|
||||
DelayTaskIssuanceHelper.getInstance().start();
|
||||
|
||||
logger.info(">>>>>>>>> init job admin success.");
|
||||
}
|
||||
|
||||
|
||||
public void destroy() throws Exception {
|
||||
DelayTaskIssuanceHelper.getInstance().toStop();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,110 @@
|
||||
package com.docus.server.message.thread;
|
||||
|
||||
import com.docus.core.util.DateUtil;
|
||||
import com.docus.core.util.Func;
|
||||
import com.docus.infrastructure.web.api.CommonResult;
|
||||
import com.docus.server.message.config.JobAdminConfig;
|
||||
import com.docus.server.message.entity.DelayedMessages;
|
||||
import com.docus.server.message.enums.TaskWait;
|
||||
import com.docus.server.message.feign.dto.CompensateTasRequest;
|
||||
import com.docus.server.message.feign.service.CollectTaskService;
|
||||
import com.docus.server.message.mapper.DelayedMessagesMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Collections;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author YongBin Wen
|
||||
* @date 2024/1/23 10:59
|
||||
*/
|
||||
|
||||
public class DelayTaskIssuanceHelper {
|
||||
private static Logger logger = LoggerFactory.getLogger(DelayTaskIssuanceHelper.class);
|
||||
private static DelayTaskIssuanceHelper instance = new DelayTaskIssuanceHelper();
|
||||
|
||||
public static DelayTaskIssuanceHelper getInstance() {
|
||||
return instance;
|
||||
}
|
||||
|
||||
private Thread delayTaskIssuanceThread;
|
||||
private volatile boolean toStop = false;
|
||||
|
||||
public void start() {
|
||||
delayTaskIssuanceThread = new Thread(() -> {
|
||||
final String messageType = TaskWait.SUBMIT_TASK_WAIT_DELAY_MSG_TYPE;
|
||||
DelayedMessages condition = new DelayedMessages();
|
||||
condition.setMessageType(messageType);
|
||||
DelayedMessagesMapper delayedMessagesMapper = JobAdminConfig.getJobAdminConfig().getDelayedMessagesMapper();
|
||||
CollectTaskService collectTaskService = JobAdminConfig.getJobAdminConfig().getCollectTaskService();
|
||||
while (!toStop) {
|
||||
try {
|
||||
LocalDateTime currentDateTime = LocalDateTime.now();
|
||||
String formatCurrentDateTime = Func.formatDateTime(currentDateTime);
|
||||
// 数据量暂时不大，使用
|
||||
List<DelayedMessages> delayedMessages = delayedMessagesMapper.findExecutableMessages(condition, formatCurrentDateTime);
|
||||
if (Func.isNotEmpty(delayedMessages)) {
|
||||
for (DelayedMessages message : delayedMessages) {
|
||||
try {
|
||||
String messageContent = message.getMessage();
|
||||
String[] split = messageContent.split("-");
|
||||
String patientId = split[0];
|
||||
String collectorId = split[1];
|
||||
List<String> patientIds = Collections.singletonList(patientId);
|
||||
List<String> collectorIds = Collections.singletonList(collectorId);
|
||||
|
||||
logger.info("提交延迟,补偿任务 patientIds:{} collectId:{}", patientIds, collectorIds);
|
||||
CompensateTasRequest dto = new CompensateTasRequest();
|
||||
dto.setPatientIds(patientIds);
|
||||
dto.setCollectorIds(collectorIds);
|
||||
dto.setPriority(3);
|
||||
CommonResult<String> result = collectTaskService.compensateTask(dto);
|
||||
logger.info("提交延迟，补偿任务结果：{}",Func.toJson(result));
|
||||
delayedMessagesMapper.delById(message.getId());
|
||||
} catch (Exception ex) {
|
||||
if (!toStop) {
|
||||
logger.error(">>>>>>>>>>> Delay Task Issuance thread error:" + ex.getMessage() + " ,delayMessage:" + Func.toJson(message), ex);
|
||||
int retries = message.getRetries() + 1;
|
||||
Integer delaySeconds = message.getDelaySeconds();
|
||||
LocalDateTime nexExecuteLocalDateTime = LocalDateTime.now().plusSeconds(delaySeconds);
|
||||
String dateTime = Func.formatDateTime(nexExecuteLocalDateTime);
|
||||
Date nexExecuteTime = Func.parseDate(dateTime, DateUtil.PATTERN_DATETIME);
|
||||
|
||||
message.setRetries(retries);
|
||||
message.setErrorMessage(ex.getMessage());
|
||||
message.setNextExecutionTime(nexExecuteTime);
|
||||
delayedMessagesMapper.updateById(message);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
TimeUnit.SECONDS.sleep(10);
|
||||
} catch (Exception ex) {
|
||||
if (!toStop) {
|
||||
logger.error(">>>>>>>>>>> Delay Task Issuance thread error:" + ex.getMessage(), ex);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
logger.info(">>>>>>>>>>> Delay Task Issuance thread stop");
|
||||
});
|
||||
delayTaskIssuanceThread.setDaemon(true);
|
||||
delayTaskIssuanceThread.setName("delayTaskIssuanceHelper");
|
||||
delayTaskIssuanceThread.start();
|
||||
}
|
||||
|
||||
public void toStop() {
|
||||
toStop = false;
|
||||
delayTaskIssuanceThread.interrupt();
|
||||
try {
|
||||
delayTaskIssuanceThread.join();
|
||||
} catch (Exception e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,56 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE mapper
|
||||
PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
|
||||
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
|
||||
<mapper namespace="com.docus.server.message.mapper.DelayedMessagesMapper">
|
||||
<insert id="insert">
|
||||
INSERT INTO
|
||||
`docus_medicalrecord`.`delayed_messages` (`id`, `message_type`, `keyword`, `message`, `next_execution_time`, `delay_seconds`, `create_time`, `retries`, `error_message`)
|
||||
VALUES (#{message.id}, #{message.messageType}, #{message.keyword}, #{message.message}, #{message.nextExecutionTime}, #{message.delaySeconds}, #{message.createTime}, 0, #{message.errorMessage})
|
||||
</insert>
|
||||
<update id="updateById">
|
||||
UPDATE `docus_medicalrecord`.`delayed_messages`
|
||||
SET
|
||||
`message_type` = #{message.messageType},
|
||||
`keyword` = #{message.keyword},
|
||||
`message` = #{message.message},
|
||||
`next_execution_time` = #{message.nextExecutionTime},
|
||||
`delay_seconds` = #{message.delaySeconds},
|
||||
`create_time` = #{message.createTime},
|
||||
`retries` = #{message.retries},
|
||||
`error_message` = #{message.errorMessage}
|
||||
WHERE `id` = #{message.id}
|
||||
</update>
|
||||
|
||||
<delete id="delById">
|
||||
DELETE FROM `docus_medicalrecord`.`delayed_messages` WHERE `id`=#{id}
|
||||
</delete>
|
||||
<select id="findExecutableMessages" resultType="com.docus.server.message.entity.DelayedMessages">
|
||||
select
|
||||
*
|
||||
from
|
||||
`docus_medicalrecord`.`delayed_messages`
|
||||
where
|
||||
`next_execution_time` <![CDATA[<=]]> #{currentDateTime}
|
||||
<if test="condition.keyword != null and condition.keyword != '' ">
|
||||
and `keyword` = #{condition.keyword}
|
||||
</if>
|
||||
<if test="condition.messageType != null and condition.messageType != '' ">
|
||||
and `message_type` = #{condition.messageType}
|
||||
</if>
|
||||
</select>
|
||||
<select id="findMessage" resultType="com.docus.server.message.entity.DelayedMessages">
|
||||
select
|
||||
*
|
||||
from
|
||||
`docus_medicalrecord`.`delayed_messages`
|
||||
where
|
||||
1=1
|
||||
<if test="condition.keyword != null and condition.keyword != '' ">
|
||||
and `keyword` = #{condition.keyword}
|
||||
</if>
|
||||
<if test="condition.messageType != null and condition.messageType != '' ">
|
||||
and `message_type` = #{condition.messageType}
|
||||
</if>
|
||||
</select>
|
||||
</mapper>
|
||||
Loading…
Reference in New Issue