MallRetryPersonLabelConsumer.java
4.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package com.viontech.keliu.consumer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.viontech.keliu.constants.KafkaConstants;
import com.viontech.keliu.constants.RedisConstants;
import com.viontech.keliu.dao.DPersonLabelDao;
import com.viontech.keliu.entity.PersonLabelContent;
import com.viontech.keliu.service.SpeedStatService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@Component
@Slf4j
public class MallRetryPersonLabelConsumer {
@Autowired
private ObjectMapper objectMapper;
@Resource
private DPersonLabelDao dPersonLabelDao;
@Resource
private SpeedStatService speedStatService;
@KafkaListener(topics = KafkaConstants.TOPIC_MALL_RETRY_PERSONLABEL
, autoStartup = "${vion.consumer.mallRetryPersonLabel.autoStartup:false}"
, groupId = "MallRetryPersonLabelToDb"
, concurrency = "${vion.consumer.mallRetryPersonLabel.concurrency:1}")
public void consumerMallRetryPersonLabel(List<ConsumerRecord<String, String>> recordList, Consumer<?, ?> consumer) {
if (CollectionUtils.isEmpty(recordList)) {
return;
}
try {
Map<String, List<ConsumerRecord<String, String>>> topicPartitionDataMap = recordList.stream().collect(Collectors.groupingBy(d -> d.topic() + "-" + d.partition()));
for (Map.Entry<String, List<ConsumerRecord<String, String>>> entry : topicPartitionDataMap.entrySet()) {
try {
long startTime = System.currentTimeMillis();
List<ConsumerRecord<String, String>> recordValues = entry.getValue();
if (!CollectionUtils.isEmpty(recordValues)) {
ConsumerRecord<String, String> lastRecord = recordValues.get(recordValues.size() - 1);
List<PersonLabelContent> labelList = new ArrayList<>();
for (ConsumerRecord<String, String> consumerRecord : recordValues) {
try {
PersonLabelContent dataContent = objectMapper.readValue(consumerRecord.value(), PersonLabelContent.class);
if (dataContent != null) {
labelList.add(dataContent);
}
} catch (Exception ee) {
log.error("处理Mall_Retry_PersonLabel[{}], JsonDeserializerThrowable={}", entry.getKey(), ee.getMessage(), ee);
}
}
if (!CollectionUtils.isEmpty(labelList)) {
for (PersonLabelContent content : labelList) {
try {
dPersonLabelDao.insert(content);
speedStatService.stat(RedisConstants.PDS_MALLRETRYPERSONLABEL_WRITE, 1);
} catch (Exception ex) {
log.error("处理Mall_Retry_PersonLabel[{}], mallid={}, unid={}, insert.Exception={}", entry.getKey(), content.getMallId(), content.getRecognitionUnid(), ex.getMessage(), ex);
}
}
}
// 提交Offset
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(
new TopicPartition(lastRecord.topic(), lastRecord.partition()),
new OffsetAndMetadata(lastRecord.offset() + 1) // 提交下一条偏移量
);
consumer.commitSync(offsets);
}
log.info("处理Mall_Retry_PersonLabel[{}], batchHandle耗时：{} ms", entry.getKey(), System.currentTimeMillis() - startTime);
} catch (Throwable e) {
log.error("处理Mall_Retry_PersonLabel[{}], Throwable={}", entry.getKey(), e.getMessage(), e);
}
}
} catch (Throwable exx) {
log.error("处理Mall_Retry_PersonLabel.Throwable={}", exx.getMessage(), exx);
}
}
}