加上调用业务
This commit is contained in:
@@ -53,17 +53,23 @@ public class NameplateKafkaConsumer {
|
|||||||
log.info("nameplateKafkaConsumerMessage,消息:{}", recordMessages);
|
log.info("nameplateKafkaConsumerMessage,消息:{}", recordMessages);
|
||||||
if (StringUtils.isEmpty(recordMessages)) {
|
if (StringUtils.isEmpty(recordMessages)) {
|
||||||
ack.acknowledge();
|
ack.acknowledge();
|
||||||
log.info("nameplateKafkaConsumerMessage,消息为空");
|
log.info("nameplateKafkaConsumer为空");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
|
NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
|
||||||
if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
|
if (!"INSERT".equals(tmNameplateCorpus.getType()) ) {
|
||||||
log.info("nameplateKafkaConsumerMessage,不满足条件");
|
log.info("nameplateKafka不是INSERT");
|
||||||
|
}
|
||||||
|
if (CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
|
||||||
|
log.info("nameplateKafkagetData为空");
|
||||||
|
ack.acknowledge();
|
||||||
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
tmNameplateCorpus.getData().forEach(nameplate ->
|
||||||
|
tmNameplateCorpusService.processItem(nameplate));
|
||||||
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
log.error("处理消息失败", e);
|
log.error("处理消息失败", e);
|
||||||
|
|||||||
Reference in New Issue
Block a user