You've already forked FrameTour-BE
feat(kafka): 新增ZT-Source Kafka消息处理功能
- 新增ZTSourceMessage实体类用于接收Kafka消息 - 新增ZTSourceConsumerService监听zt-source主题 - 新增ZTSourceDataService处理消息并保存至数据库- 扩展SourceMapper支持从ZT-Source消息新增素材 - 实现照片类型素材的解析、校验与存储逻辑 - 添加Kafka手动ACK确认机制确保消息可靠处理
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
package com.ycwl.basic.service;
|
||||
|
||||
import com.ycwl.basic.dto.ZTSourceMessage;
|
||||
import com.ycwl.basic.utils.JacksonUtil;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
import org.springframework.kafka.support.Acknowledgment;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* ZT-Source Kafka消费者服务
|
||||
* 监听zt-source topic并处理素材消息
|
||||
*
|
||||
* @author system
|
||||
* @date 2024/12/27
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
@ConditionalOnProperty(name = "kafka.enabled", havingValue = "true")
|
||||
public class ZTSourceConsumerService {
|
||||
|
||||
private static final String ZT_SOURCE_TOPIC = "zt-source";
|
||||
|
||||
private final ZTSourceDataService ztSourceDataService;
|
||||
|
||||
/**
|
||||
* 监听zt-source topic消息
|
||||
* 先解析消息并输出业务日志，然后手动确认处理
|
||||
*
|
||||
* @param message 消息JSON字符串
|
||||
* @param ack 手动ACK确认
|
||||
*/
|
||||
@KafkaListener(topics = ZT_SOURCE_TOPIC, containerFactory = "manualCommitKafkaListenerContainerFactory")
|
||||
public void handleZTSourceMessage(String message, Acknowledgment ack) {
|
||||
ZTSourceMessage sourceMessage = null;
|
||||
|
||||
try {
|
||||
// 先解析消息
|
||||
sourceMessage = JacksonUtil.parseObject(message, ZTSourceMessage.class);
|
||||
|
||||
// 输出业务相关的日志信息
|
||||
log.info("接收到ZT-Source消息, sourceId: {}, sourceType: {}, scenicId: {}, deviceId: {}, sourceUrl: {}",
|
||||
sourceMessage.getSourceId(), sourceMessage.getSourceType(),
|
||||
sourceMessage.getScenicId(), sourceMessage.getDeviceId(), sourceMessage.getSourceUrl());
|
||||
|
||||
// 处理消息
|
||||
boolean processed = ztSourceDataService.processZTSourceMessage(sourceMessage);
|
||||
|
||||
if (processed) {
|
||||
// 只有在处理成功后才手动提交
|
||||
ack.acknowledge();
|
||||
log.info("ZT-Source消息处理成功并已提交, sourceId: {}", sourceMessage.getSourceId());
|
||||
} else {
|
||||
log.warn("ZT-Source消息处理被跳过（非照片类型），消息不会被提交, sourceId: {}, sourceType: {}",
|
||||
sourceMessage.getSourceId(), sourceMessage.getSourceType());
|
||||
// 对于非照片类型，也提交消息避免重复消费
|
||||
ack.acknowledge();
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
String sourceId = sourceMessage != null ? sourceMessage.getSourceId().toString() : "unknown";
|
||||
log.error("处理ZT-Source消息失败，消息不会被提交: sourceId={}, error={}", sourceId, e.getMessage(), e);
|
||||
// 不调用ack.acknowledge()，消息保持未提交状态，可以重新消费
|
||||
}
|
||||
}
|
||||
}
|
||||
125
src/main/java/com/ycwl/basic/service/ZTSourceDataService.java
Normal file
125
src/main/java/com/ycwl/basic/service/ZTSourceDataService.java
Normal file
@@ -0,0 +1,125 @@
|
||||
package com.ycwl.basic.service;
|
||||
|
||||
import com.ycwl.basic.dto.ZTSourceMessage;
|
||||
import com.ycwl.basic.mapper.SourceMapper;
|
||||
import com.ycwl.basic.model.pc.source.entity.SourceEntity;
|
||||
import com.ycwl.basic.utils.SnowFlakeUtil;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import java.sql.Timestamp;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* ZT-Source数据处理服务
|
||||
* 负责将ZT-Source消息转换并保存到数据库
|
||||
*
|
||||
* @author system
|
||||
* @date 2024/12/27
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
@RequiredArgsConstructor
|
||||
public class ZTSourceDataService {
|
||||
|
||||
private final SourceMapper sourceMapper;
|
||||
|
||||
/**
|
||||
* 处理ZT-Source消息，仅处理照片类型(sourceType=2)
|
||||
*
|
||||
* @param message ZT-Source消息
|
||||
* @return 是否处理成功
|
||||
*/
|
||||
public boolean processZTSourceMessage(ZTSourceMessage message) {
|
||||
try {
|
||||
// 仅处理照片类型的消息
|
||||
if (!message.isPhoto()) {
|
||||
log.debug("跳过非照片类型消息: sourceId={}, sourceType={}",
|
||||
message.getSourceId(), message.getSourceType());
|
||||
return false;
|
||||
}
|
||||
|
||||
// 检查必要字段
|
||||
if (!validateMessage(message)) {
|
||||
log.warn("消息校验失败: {}", message);
|
||||
return false;
|
||||
}
|
||||
|
||||
// 转换为SourceEntity
|
||||
SourceEntity sourceEntity = convertToSourceEntity(message);
|
||||
|
||||
// 保存到数据库
|
||||
int result = sourceMapper.addFromZTSource(sourceEntity);
|
||||
|
||||
if (result > 0) {
|
||||
log.info("成功保存ZT-Source照片素材: sourceId={}, entityId={}, scenicId={}, deviceId={}",
|
||||
message.getSourceId(), sourceEntity.getId(), message.getScenicId(), message.getDeviceId());
|
||||
return true;
|
||||
} else {
|
||||
log.error("保存ZT-Source照片素材失败: sourceId={}", message.getSourceId());
|
||||
return false;
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
log.error("处理ZT-Source消息异常: sourceId={}, error={}",
|
||||
message.getSourceId(), e.getMessage(), e);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 校验消息必要字段
|
||||
*/
|
||||
private boolean validateMessage(ZTSourceMessage message) {
|
||||
if (message.getScenicId() == null) {
|
||||
log.warn("scenicId不能为空: sourceId={}", message.getSourceId());
|
||||
return false;
|
||||
}
|
||||
|
||||
if (message.getDeviceId() == null) {
|
||||
log.warn("deviceId不能为空: sourceId={}", message.getSourceId());
|
||||
return false;
|
||||
}
|
||||
|
||||
if (message.getSourceUrl() == null || message.getSourceUrl().trim().isEmpty()) {
|
||||
log.warn("sourceUrl不能为空: sourceId={}", message.getSourceId());
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 将ZTSourceMessage转换为SourceEntity
|
||||
*/
|
||||
private SourceEntity convertToSourceEntity(ZTSourceMessage message) {
|
||||
SourceEntity entity = new SourceEntity();
|
||||
|
||||
// 生成ID
|
||||
entity.setId(SnowFlakeUtil.getLongId());
|
||||
|
||||
// 基本字段映射
|
||||
entity.setScenicId(message.getScenicId());
|
||||
entity.setDeviceId(message.getDeviceId());
|
||||
entity.setUrl(message.getSourceUrl()); // 使用sourceUrl，不使用缩略图
|
||||
entity.setType(2); // 照片类型
|
||||
|
||||
// 人脸样本ID处理
|
||||
entity.setFaceSampleId(message.getFaceSampleId());
|
||||
|
||||
// 时间处理
|
||||
Date shootTime = message.getShootTime();
|
||||
if (shootTime != null) {
|
||||
entity.setCreateTime(shootTime);
|
||||
} else {
|
||||
entity.setCreateTime(new Date());
|
||||
}
|
||||
|
||||
log.debug("转换ZTSourceMessage到SourceEntity: sourceId={} -> entityId={}",
|
||||
message.getSourceId(), entity.getId());
|
||||
|
||||
return entity;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user