wwwwwwwxqxq / huaweicloud-amqp-adapter Goto Github PK

// 属性类
public class HealthProperties extends Properties {
    // properties
    ...
}

// AMQP 消息接收类
public class HealthAmqpMsg extends BaseMessageTemplate<HealthProperties> {}

// 消息转换工具类
public class HealthAmqpMsgConverter {
    public HealthDataDTO convertToDTO(HealthAmqpMsg healthData){
        HealthDataDTO dto = new HealthDataDTO();
        // build header
        Header header = healthData.getNotifyData().getHeader();
        dto.setAppId(header.getAppId());
        dto.setDeviceId(header.getDeviceId());
        dto.setNodeId(header.getNodeId());
        dto.setProductId(header.getProductId());
        dto.setGatewayId(header.getGatewayId());

        // build body
        Body<HealthProperties> body = healthData.getNotifyData().getBody();
        Services<HealthProperties> services =  body.getServices().get(0);
        dto.setServiceId(services.getServiceId());
        HealthProperties healthProperties = services.getProperties();

        // build properties
        
        return dto;
    }
}

// 开发消费者
public class HealthConsumer extends QpidConsumer {

    @Autowired
    private HealthService heartHealthService;

    @Autowired
    private HealthAmqpMsgConverter heartConverter;

    private void save(HealthAmqpMsg healthData){
        HealthDataDTO healthDataDTO = heartConverter.convertToDTO(healthData);
        heartHealthService.save(healthDataDTO);
    }

    @Scheduled(fixedDelay = 300000)
    public void receiveHealthData() {
        log.info("[定时任务开始], {}", LocalDateTime.now());
        List<HealthAmqpMsg> healthData = qpidJmsTemplate.receiveAndConvertToJson(HealthAmqpMsg.class);
        log.info("共接收到 [{}] 条消息", healthData.size());

        if (!healthData.isEmpty()) {
            log.info("[准备入库], {}, 入库信息为: {}",LocalDateTime.now(), healthData);
            healthData.forEach(this::save);
            log.info("[已入库，定时任务结束], {} ", LocalDateTime.now());
        }
    }
}

@Data   //lombok注释，用于自动生成getter和setter
@Entity //javax.persistence注释，用于注释PO表明其为数据库实体类
// 类名需和表名对应，此类对应health_data表
public class HealthData extends BasePO { 
    // header
    private String appId;
    private String deviceId;
    private String nodeId;
    private String productId;
    private String gatewayId;

    // body.services
    private String serviceId;
    private Date eventTime;
    
    // properties
    ...
}

@Data
@AllArgsConstructor
@NoArgsConstructor
// 用于在DAO和AMQP Consumer与Service层间传递消息
public class HealthDataDTO extends BaseDTO { 
    // header
    private String appId;
    private String deviceId;
    private String nodeId;
    private String productId;
    private String gatewayId;

    // body.services
    private String serviceId;
    private Date eventTime;
  
    // properties
    ...
}

@Data
@AllArgsConstructor
@NoArgsConstructor
// 响应HTTP请求时返回的数据
public class HealthDataVO implements Serializable { 
    // 前端需要展示的数据字段
    ...
}

//Vo<->DTO
public class HealthDVConverter 
  implements BaseConverter<HealthDataDTO, HealthDataVO> {}

//DTO<->PO
public class HealthPDConverter 
  implements BaseConverter<HealthData, HealthDataDTO> {}

@CrossOrigin
@RestController
@RequestMapping("/health")
public class HealthController {

    @Autowired
    private HealthService healthService;

    @Autowired
    private HealthDVConverter converter;
		
  	// 监听/health/queryPage的HTTP请求
    // Page为分页信息类，WebResponse为统一返回类
    @RequestMapping(value = "/queryPage", method = RequestMethod.POST)
    public WebResponse<Page<HealthDataVO>> queryPage(@RequestBody(required = false) AmqpQuery amqpQuery){
        PageDTO<HealthDataDTO> pageDTO = healthService.queryPage(amqpQuery);
        return WebResponse.success(Page.from(pageDTO, dto ->converter.toTarget(dto)));
    }
}

// 面向接口开发，定义规范
package edu.hrbust.iot.amqp.web.service
public interface HealthService {

    PageDTO<HealthDataDTO> queryPage(AmqpQuery amqpQuery);

    void save(HealthDataDTO healthDataDTO);

    List<HealthDataDTO> queryAll();
}

// 接口实现类
package edu.hrbust.iot.amqp.web.service.impl
public class DefaultHealthService implements HealthService {
    
    @Autowired
    private HealthDataRepository repository;

    @Autowired
    private HealthPDConverter converter;
  
    //编写具体的实现代码
    @Override
    public PageDTO<HealthDataDTO> queryPage(AmqpQuery amqpQuery){}

  	@Override
    public void save(HealthDataDTO healthDataDTO){}
  
    @Override
    public List<HealthDataDTO> queryAll(){}
  
}

// Dao层采用Spring Data JPA实现，只需继承JpaRepository即可实现简单的数据库查找功能
// JpaSpecificationExecutor接口提供分页功能
public interface HealthDataRepository extends
        JpaRepository<HealthData, Long>, JpaSpecificationExecutor<HealthData> {}