attiya/ebike-plus
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

kafka优化

Browse Source
This commit is contained in:
attiya 2025-12-18 11:23:18 +08:00
parent e9198ae852
commit 492422acf1
2 changed files with 3 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ebike-report/src/main/java/com/cdzy/report/component/KafkaConsumer.java
View File

@ -39,6 +39,6 @@ public class KafkaConsumer {
public void onMessage(ConsumerRecord<?, ?> record) throws JsonProcessingException { public void onMessage(ConsumerRecord<?, ?> record) throws JsonProcessingException {
log.info("[KAFKA接收] 主题: {}, 内容: {}", record.topic(), record.value()); log.info("[KAFKA接收] 主题: {}, 内容: {}", record.topic(), record.value());
JsonNode jsonNode = mapper.readValue(record.value().toString(), JsonNode.class); JsonNode jsonNode = mapper.readValue(record.value().toString(), JsonNode.class);
reoprtHandler.reportHandler(jsonNode,record.topic()); reoprtHandler.reportHandler(jsonNode);
} }
} }

5
ebike-report/src/main/java/com/cdzy/report/component/ReoprtHandler.java
View File

@ -36,11 +36,10 @@ public class ReoprtHandler {
/** /**
* 报文处理 * 报文处理
*/ */
public void reportHandler(JsonNode response,String topic) { public void reportHandler(JsonNode response) {
int c = response.get("c").asInt(); int c = response.get("c").asInt();
JsonNode param = response.get("param"); JsonNode param = response.get("param");
String[] parts = topic.split("/"); String deviceId = param.get("SN").asText();
String deviceId = parts[parts.length - 1];
switch (c) { switch (c) {
case 56: case 56:
gpsMsgHandler(param, deviceId); gpsMsgHandler(param, deviceId);