|
|
@ -112,7 +112,7 @@ public class RedisStreamConsumer { |
|
|
// 定时统计接收次数
|
|
|
// 定时统计接收次数
|
|
|
statsScheduler.scheduleAtFixedRate(() -> { |
|
|
statsScheduler.scheduleAtFixedRate(() -> { |
|
|
long count = receiveCounter.sum(); |
|
|
long count = receiveCounter.sum(); |
|
|
log.info("RedisStreamConsumer receive count={}", count); |
|
|
log.info("[F10] RedisStreamConsumer receive count={}", count); |
|
|
}, 10, 10, TimeUnit.SECONDS); |
|
|
}, 10, 10, TimeUnit.SECONDS); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -234,7 +234,7 @@ public class RedisStreamConsumer { |
|
|
boolean success = false; |
|
|
boolean success = false; |
|
|
try { |
|
|
try { |
|
|
Map<Object, Object> body = message.getValue(); |
|
|
Map<Object, Object> body = message.getValue(); |
|
|
log.info("stream={}, id={}, body={}", |
|
|
log.info("[F10] stream={}, id={}, body={}", |
|
|
message.getStream(), |
|
|
message.getStream(), |
|
|
message.getId(), |
|
|
message.getId(), |
|
|
body); |
|
|
body); |
|
|
@ -246,14 +246,14 @@ public class RedisStreamConsumer { |
|
|
String jsonParams = gson.toJson(body); |
|
|
String jsonParams = gson.toJson(body); |
|
|
MyHTTPResponse response = this.defaultHttpRequestUtil.postJson(dataCenterEnvConfig.getF10ApiUrl(), jsonParams); |
|
|
MyHTTPResponse response = this.defaultHttpRequestUtil.postJson(dataCenterEnvConfig.getF10ApiUrl(), jsonParams); |
|
|
if (response.getCode() == 200) { |
|
|
if (response.getCode() == 200) { |
|
|
log.info("F10 data sent successfully...."); |
|
|
log.info("[F10] F10 data sent successfully: {}", jsonParams); |
|
|
} else { |
|
|
} else { |
|
|
log.error("F10 data sent failed...."); |
|
|
log.error("[F10] F10 data sent failed...."); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
success = true; |
|
|
success = true; |
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
log.error("process error id={}", message.getId(), e); |
|
|
log.error("[F10] process error id={}", message.getId(), e); |
|
|
} finally { |
|
|
} finally { |
|
|
if (success) { |
|
|
if (success) { |
|
|
ack(stream, message); |
|
|
ack(stream, message); |
|
|
@ -281,9 +281,9 @@ public class RedisStreamConsumer { |
|
|
.withStreamKey(DLQ_STREAM) |
|
|
.withStreamKey(DLQ_STREAM) |
|
|
); |
|
|
); |
|
|
ack(stream, message); |
|
|
ack(stream, message); |
|
|
log.error("move to DLQ :{}", message); |
|
|
log.error("[F10] move to DLQ :{}", message); |
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
log.error("dlq error :{}", message, e); |
|
|
log.error("[F10] dlq error :{}", message, e); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@ -322,7 +322,7 @@ public class RedisStreamConsumer { |
|
|
if (pending.size() < BATCH_SIZE) break; |
|
|
if (pending.size() < BATCH_SIZE) break; |
|
|
} |
|
|
} |
|
|
} catch (Exception e) { |
|
|
} catch (Exception e) { |
|
|
log.error("reclaim error stream={}", stream, e); |
|
|
log.error("[F10] reclaim error stream={}", stream, e); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|