功能开发

对象扩展

@Data
public class TaskMessageEntityCommand {
  
    /**
     * 任务ID
     */
    private String taskId;

    /**
     * 任务名称
     */
    private String taskName;

    /**
     * 任务类型
     */
    private String notifyType;

    /**
     * 通知配置
     */
    private NotifyConfig notifyConfig;

    /**
     * 任务状态
     */
    private Integer status;

    /**
     * 参数配置
     */
    private String parameterJson;

    public TaskMessageEntityCommand() {
    }

    public TaskMessageEntityCommand(String taskId, String taskName, TaskNotifyEnum taskNotifyEnum, NotifyConfig notifyConfig, String parameterJson) {
        this.taskId = taskId;
        this.taskName = taskName;
        this.notifyType = taskNotifyEnum.getType();
        this.notifyConfig = notifyConfig;
        this.status = 0;
        this.parameterJson = parameterJson;
    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class NotifyConfig {
        // mq 配置
        private MQ mq;
        // http 配置
        private HTTP http;

        @Data
        @Builder
        @AllArgsConstructor
        @NoArgsConstructor
        public static class MQ {
            private String topic;
            private String exchange;
        }

        @Data
        @Builder
        @AllArgsConstructor
        @NoArgsConstructor
        public static class HTTP {
            private String url;
            private String method;
            private String contentType;
            private String authorization;
        }
    }

}

• 为了满足新的数据库表的操作，这里要对 TaskMessageEntityCommand 对象扩展字段。基本是与数据库所需的配置是一致的。只不过多增加了 NotifyConfig 下的对象 MQ、HTTP，这样设计的目的是为了让使用方清楚的在一个类下知道有哪些对象可用。这部分也可以单独拆分为值对象，MQ、HTTP 配置就是值对象，但就是用户使用的时候，会感觉从一个包下引入下的，体验不如从一个类下获取更清晰。

dao 操作

@Slf4j
@Component
public class TaskMessageDaoImpl implements ITaskMessageDao {

    private final DataSource dataSource;

    public TaskMessageDaoImpl(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public int insert(TaskMessagePO taskMessagePO) throws SQLException {
        String sql = "INSERT INTO local_task_message (task_id, task_name, notify_type, notify_config, status, parameter_json, house_number ,create_time, update_time) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";

        try (Connection connection = dataSource.getConnection();
             PreparedStatement ps = connection.prepareStatement(sql)) {

            ps.setString(1, taskMessagePO.getTaskId());
            ps.setString(2, taskMessagePO.getTaskName());
            ps.setString(3, taskMessagePO.getNotifyType());
            ps.setString(4, taskMessagePO.getNotifyConfig());
            ps.setInt(5, taskMessagePO.getStatus());
            ps.setString(6, taskMessagePO.getParameterJson());
            ps.setInt(7, taskMessagePO.getHouseNumber());
            ps.setObject(8, taskMessagePO.getCreateTime());
            ps.setObject(9, taskMessagePO.getUpdateTime());

            return ps.executeUpdate();

        } catch (SQLException e) {
            log.error("插入任务消息失败，taskId: {}", taskMessagePO.getTaskId(), e);
            throw e;
        }
    }

}

• 注入 DataSource 数据源，通过 jdbc 方式，操作数据库插入的处理。兼容性更强

适配接口

public interface ILocalTaskMessageRepository {

    /**
     * 保存任务消息
     * @param command 任务消息实体命令
     */
    void saveTaskMessage(TaskMessageEntityCommand command) throws Exception;

}

@Slf4j
@Repository
public class LocalTaskMessageRepository implements ILocalTaskMessageRepository {

    private final ITaskMessageDao taskMessageDao;

    public LocalTaskMessageRepository(ITaskMessageDao taskMessageDao) {
        this.taskMessageDao = taskMessageDao;
    }

    @Override
    public void saveTaskMessage(TaskMessageEntityCommand command) throws Exception {

        TaskMessagePO po = new TaskMessagePO();
        po.setTaskId(command.getTaskId());
        po.setTaskName(command.getTaskName());
        po.setNotifyType(command.getNotifyType());
        po.setStatus(command.getStatus());
        po.setParameterJson(command.getParameterJson());

        // 将NotifyConfig对象转换为JSON字符串
        if (command.getNotifyConfig() != null) {
            po.setNotifyConfig(JSON.toJSONString(command.getNotifyConfig()));
        }

        // 根据任务ID计算哈希值，取正数，获取最后一位数字作为门牌号
        int hashCode = Math.abs(command.getTaskId().hashCode());
        int houseNumber = hashCode % 10;
        po.setHouseNumber(houseNumber);

        po.setCreateTime(LocalDateTime.now());
        po.setUpdateTime(LocalDateTime.now());

        try {
            int result = taskMessageDao.insert(po);
            if (1 != result) {
                throw new RuntimeException("result is not 1 taskId:{}" + command.getTaskId());
            }
        } catch (Exception e) {
            log.error("保存任务消息失败，taskId: {} {}", command.getTaskId(), JSON.toJSONString(command), e);
            throw e;
        }

    }

}

• 它的实现就是把 TaskMessageEntityCommand 领域对象，转换为数据库的 PO 对象，调用 dao 完成数据插入操作。

服务调用

@Slf4j
@Service
public class LocalTaskMessageHandleService implements ILocalTaskMessageHandleService {

    @Resource
    private ILocalTaskMessageEvent event;

    @Resource
    private ILocalTaskMessageRepository repository;

    @Override
    public void acceptTaskMessage(TaskMessageEntityCommand command) {
        try {
            log.info("受理任务消息: {}", command);

            // 1. 保存任务消息
            repository.saveTaskMessage(command);

            // 2. 发布事件消息
            event.publishEvent(command);

        } catch (Exception e) {
            log.error("受理任务消息执行失败 {}", JSON.toJSONString(command), e);
        }
    }

}

• 之后就要扩展服务的调用了，在受理任务消息的时候，调用数据库操作，保存消息，之后才是发送事件消息。