Improve MonitoringTimerService logic
This commit is contained in:
Mikhail Trofimov
parent
8071209098
commit
e65f002672
5 changed files with 26 additions and 60 deletions
|
|
@ -11,7 +11,7 @@ import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
|
|||
import org.springframework.kafka.core.*;
|
||||
import org.springframework.kafka.listener.ContainerProperties;
|
||||
import org.springframework.kafka.support.serializer.JsonDeserializer;
|
||||
import ru.nbch.credit_tracker.service.mode.monitoring.TestData;
|
||||
import ru.nbch.credit_tracker.service.mode.monitoring.PackageMessage;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
|
@ -21,8 +21,8 @@ import java.util.Map;
|
|||
public class KafkaConfig {
|
||||
|
||||
@Bean
|
||||
public ConcurrentKafkaListenerContainerFactory<String, TestData> kafkaListenerContainerFactory(ConsumerFactory<String, TestData> consumerFactory) {
|
||||
ConcurrentKafkaListenerContainerFactory<String, TestData> factory =
|
||||
public ConcurrentKafkaListenerContainerFactory<String, PackageMessage> kafkaListenerContainerFactory(ConsumerFactory<String, PackageMessage> consumerFactory) {
|
||||
ConcurrentKafkaListenerContainerFactory<String, PackageMessage> factory =
|
||||
new ConcurrentKafkaListenerContainerFactory<>();
|
||||
factory.setConsumerFactory(consumerFactory);
|
||||
factory.setBatchListener(true);
|
||||
|
|
@ -32,14 +32,14 @@ public class KafkaConfig {
|
|||
}
|
||||
|
||||
@Bean
|
||||
public ConsumerFactory<String, TestData> consumerFactory() {
|
||||
public ConsumerFactory<String, PackageMessage> consumerFactory() {
|
||||
Map<String, Object> props = new HashMap<>();
|
||||
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092");
|
||||
props.put(ConsumerConfig.GROUP_ID_CONFIG, "credit-tracker");
|
||||
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
|
||||
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 20);
|
||||
|
||||
JsonDeserializer<TestData> jsonDeserializer = new JsonDeserializer<>(TestData.class);
|
||||
JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(PackageMessage.class);
|
||||
jsonDeserializer.addTrustedPackages("*");
|
||||
jsonDeserializer.setUseTypeHeaders(false);
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,6 @@ public class CTMonitoringApplication {
|
|||
// MonitoringService monitoringService = ctx.getBean(MonitoringService.class);
|
||||
// monitoringService.run();
|
||||
|
||||
// ctx.getBean(KafkaMonitoringService.class).generateMessages();
|
||||
ctx.getBean(KafkaMonitoringService.class).generateMessages();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import java.nio.file.Path;
|
|||
import java.nio.file.Paths;
|
||||
import java.nio.file.StandardOpenOption;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
|
|
@ -64,21 +65,26 @@ public class KafkaMonitoringService {
|
|||
monitoringTimerService.registerFile(msg.packageId(), filePath);
|
||||
}
|
||||
|
||||
// TODO send PackageMessage
|
||||
public void generateMessages() throws JsonProcessingException {
|
||||
ObjectMapper mapper = JsonMapper.builder()
|
||||
.build();
|
||||
Random rand = new Random();
|
||||
int max = 10;
|
||||
int min = 1;
|
||||
for (int i = 0; i < 500; i++) {
|
||||
TestData data = new TestData();
|
||||
data.setId(UUID.randomUUID());
|
||||
data.setMessage("Message " + i);
|
||||
send(data.getId().toString(), mapper.writeValueAsString(data));
|
||||
PackageMessage data = new PackageMessage(
|
||||
rand.nextInt((max - min) + 1) + min,
|
||||
"Message " + i,
|
||||
UUID.randomUUID()
|
||||
);
|
||||
|
||||
send(data.uuid().toString(), mapper.writeValueAsString(data));
|
||||
}
|
||||
}
|
||||
|
||||
private void send(String id, String data) {
|
||||
CompletableFuture<SendResult<String, String>> future =
|
||||
kafkaTemplate.send("test-topic", id, data);
|
||||
kafkaTemplate.send("my-topic", id, data);
|
||||
|
||||
future.whenComplete((result, exception) -> {
|
||||
if (exception == null) {
|
||||
|
|
|
|||
|
|
@ -32,7 +32,7 @@ public class MonitoringTimerService {
|
|||
|
||||
public void registerFile(Integer packageId, Path filePath) {
|
||||
bundles.compute(packageId, (id, existing) -> {
|
||||
if (existing == null || existing.isIdle()) {
|
||||
if (existing == null) {
|
||||
PackageBundle bundle = new PackageBundle();
|
||||
bundle.addFile(filePath);
|
||||
scheduleFlush(packageId, bundle);
|
||||
|
|
@ -40,19 +40,21 @@ public class MonitoringTimerService {
|
|||
return bundle;
|
||||
}
|
||||
existing.addFile(filePath);
|
||||
log.debug("Queued file for packageId={} state={}", packageId, existing.getState());
|
||||
return existing;
|
||||
});
|
||||
}
|
||||
|
||||
private void scheduleFlush(Integer packageId, PackageBundle bundle) {
|
||||
bundle.setState(BundleState.SCHEDULED);
|
||||
scheduler.schedule(() -> flush(packageId, bundle), DELAY_MINUTES, TimeUnit.MINUTES);
|
||||
}
|
||||
|
||||
private void flush(Integer packageId, PackageBundle bundle) {
|
||||
bundle.setState(BundleState.MERGING);
|
||||
List<Path> snapshot = bundle.drainFiles();
|
||||
// Атомарно забираем snapshot и удаляем из мапы
|
||||
List<Path> snapshot = new ArrayList<>();
|
||||
bundles.compute(packageId, (id, current) -> {
|
||||
snapshot.addAll(bundle.drainFiles());
|
||||
return null; // удаляем этот пакет
|
||||
});
|
||||
|
||||
try {
|
||||
if (!snapshot.isEmpty()) {
|
||||
|
|
@ -74,19 +76,8 @@ public class MonitoringTimerService {
|
|||
} catch (IOException e) {
|
||||
log.error("Flush failed for packageId={}", packageId, e);
|
||||
} finally {
|
||||
bundle.setState(BundleState.IDLE);
|
||||
bundles.remove(packageId);
|
||||
|
||||
// Файлы, добавленные во время мержа — забираем и стартуем новый цикл
|
||||
List<Path> leftover = bundle.drainFiles();
|
||||
if (!leftover.isEmpty()) {
|
||||
log.info("Files arrived during merge, restarting timer for packageId={}, count={}",
|
||||
packageId, leftover.size());
|
||||
PackageBundle next = new PackageBundle();
|
||||
leftover.forEach(next::addFile);
|
||||
bundles.put(packageId, next);
|
||||
scheduleFlush(packageId, next);
|
||||
}
|
||||
bundle.drainFiles().forEach(leftover -> registerFile(packageId, leftover));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -105,11 +96,9 @@ public class MonitoringTimerService {
|
|||
log.info("Merged {} files → {} for packageId={}", files.size(), merged, packageId);
|
||||
}
|
||||
|
||||
private enum BundleState {SCHEDULED, MERGING, IDLE}
|
||||
|
||||
private static class PackageBundle {
|
||||
private final Queue<Path> files = new ConcurrentLinkedQueue<>();
|
||||
private volatile BundleState state = BundleState.IDLE;
|
||||
|
||||
void addFile(Path path) {
|
||||
files.add(path);
|
||||
|
|
@ -121,17 +110,5 @@ public class MonitoringTimerService {
|
|||
while ((p = files.poll()) != null) result.add(p);
|
||||
return result;
|
||||
}
|
||||
|
||||
BundleState getState() {
|
||||
return state;
|
||||
}
|
||||
|
||||
void setState(BundleState s) {
|
||||
this.state = s;
|
||||
}
|
||||
|
||||
boolean isIdle() {
|
||||
return state == BundleState.IDLE;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,17 +0,0 @@
|
|||
package ru.nbch.credit_tracker.service.mode.monitoring;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
import lombok.Setter;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
@Getter
|
||||
@Setter
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public class TestData {
|
||||
private UUID id;
|
||||
private String message;
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue