treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

QueueConsumer защита от переполнения спамом логов

Browse source
This commit is contained in:
AKurakin 2023-05-03 19:02:00 +03:00
parent 2ad812d603
commit 29b47b8b96
2 changed files with 11 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/domain/Consts.java
View file

@ -99,7 +99,6 @@ public interface Consts {
String SDF11_PROCESS = "sdf11-process"; String SDF11_PROCESS = "sdf11-process";
String EXPORT_PROCESS = "export-process"; String EXPORT_PROCESS = "export-process";
String S_TRADES_IMPORTED = "s_trades-imported"; String S_TRADES_IMPORTED = "s_trades-imported";
String ACCOUNT_NEW = "account-new";
String ACCOUNT_TERMINATION = "account-termination"; String ACCOUNT_TERMINATION = "account-termination";
String BALANCE_ACCOUNT_NEW = "balance-account-new"; String BALANCE_ACCOUNT_NEW = "balance-account-new";
String BALANCE_ACCOUNT_UPDATE = "balance-account-update"; String BALANCE_ACCOUNT_UPDATE = "balance-account-update";

11
platform-parent/platform-messaging/src/main/java/ru/spcex/clearing/platform/messaging/service/QueueConsumer.java
View file

@ -80,6 +80,7 @@ public class QueueConsumer implements AutoCloseable {
consumer.subscribe(callbacks.keySet()); consumer.subscribe(callbacks.keySet());
} }
Object o = null; Object o = null;
int lastErrors = 0;
while (!closed.get()) { while (!closed.get()) {
try { try {
ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS)); ConsumerRecords<String, Object> records = consumer.poll(Duration.of(10, ChronoUnit.SECONDS));
@ -95,11 +96,21 @@ public class QueueConsumer implements AutoCloseable {
} }
} }
} }
lastErrors = 0;
} catch (Throwable e) { } catch (Throwable e) {
log.error(ExceptionUtils.getStackTrace(e)); log.error(ExceptionUtils.getStackTrace(e));
if (producer != null && o != null) { if (producer != null && o != null) {
sendErrorResponse((BaseRequest<?>) o); sendErrorResponse((BaseRequest<?>) o);
} }
if (lastErrors++ > 20) {
log.warn("Too many error at row, {}. Sleep.", lastErrors);
try {
Thread.sleep(1000L);
} catch (InterruptedException ie) {
log.info("Thread interrupted. {}", ExceptionUtils.getStackTrace(ie));
break;
}
}
} }
} }
} catch (WakeupException e) { } catch (WakeupException e) {