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

Drain ivents from queue

Browse source
This commit is contained in:
Mikhail Trofimov 2026-06-29 15:45:42 +03:00
parent c5ace5e151
commit a3a1debe45
1 changed files with 24 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/SignalKafkaEventWriterService.java
View file

@ -30,6 +30,8 @@ import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption; import java.nio.file.StandardOpenOption;
import java.time.LocalDateTime; import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter; import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
@ -40,7 +42,7 @@ import java.util.concurrent.TimeUnit;
@Service @Service
public class SignalKafkaEventWriterService { public class SignalKafkaEventWriterService {
private final static Charset encoding = Charset.forName("windows-1251"); private final static Charset encoding = Charset.forName("windows-1251");
private final static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy"); private final static DateTimeFormatter formatter = DateTimeFormatter.ofPattern("dd.MM.yyyy HH:mm:ss"); // TODO убрать время
private final ExecutorService executorService = Executors.newSingleThreadExecutor( private final ExecutorService executorService = Executors.newSingleThreadExecutor(
new CustomizableThreadFactory("signal-kafka-event-writer-") new CustomizableThreadFactory("signal-kafka-event-writer-")
@ -65,20 +67,22 @@ public class SignalKafkaEventWriterService {
public void startWriting() { public void startWriting() {
while (true) { while (true) {
SignalKafkaEvent event = take(); List<SignalKafkaEvent> events = drain();
if (event == null) { for (SignalKafkaEvent event : events) {
continue; if (event == null) {
} continue;
if (event.getEventType() == EventType.STOP) { }
break; if (event.getEventType() == EventType.STOP) {
} break;
}
if (event.getEventType() == EventType.WRITE) { if (event.getEventType() == EventType.WRITE) {
writeSignal(event); writeSignal(event);
} }
if (event.getEventType() == EventType.ROTATE) { if (event.getEventType() == EventType.ROTATE) {
rotate(); rotate();
}
} }
} }
} }
@ -295,13 +299,14 @@ public class SignalKafkaEventWriterService {
.replace("'", "&apos;"); .replace("'", "&apos;");
} }
private SignalKafkaEvent take() { private List<SignalKafkaEvent> drain() {
try { try {
return signalEventsQueue.take(); List<SignalKafkaEvent> batch = new ArrayList<>(1000);
} catch (InterruptedException e) { signalEventsQueue.drainTo(batch, 1000);
log.error("Error while taking signal event", e); return batch;
Thread.currentThread().interrupt(); } catch (Exception e) {
return null; log.error("Error while drain signal events", e);
return Collections.emptyList();
} }
} }