Я использую очередь хроник 5.16.8.
Я получаю это предупреждение
net.openhft.chronicle.threads.Pauser : Using Pauser.sleepy() as not enough processors, have 1, needs 8+
Можно ли увеличить этот процессор?
исходный код
Мой подход здесь заключается в использовании карты хроник для хранения программы чтения индексов. Я думаю, я могу включить такое же поведение в recordHistory ...
Мне пришлось использовать преобразованный Джексон Json ... Я не понял, как использовать метод writeDocument.
Цикл While true, это еще одна неприятная вещь, которая у меня есть здесь... Я не мог понять, как найти последний индекс в очереди.
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.ExcerptTailer;
import net.openhft.chronicle.queue.RollCycles;
import net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.File;
import java.util.LinkedList;
import java.util.List;
@Service
public class QueueService {
public static final String INDEX = "index";
private ObjectMapper mapper = new ObjectMapper(); // json converter
@PostConstruct
public void init() {
mapper.registerModule(new JavaTimeModule());
mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
}
public void write(List dtos, String path) throws Exception {
try (ChronicleQueue queue = SingleChronicleQueueBuilder.binary(path).rollCycle(RollCycles.DAILY).build()) {
final ExcerptAppender appender = queue.acquireAppender();
for (int i=0; i<dtos.size(); i++) {
appender.writeText(mapper.writeValueAsString(dtos.get(i)));
}
}
}
public void write(Object dto, String path) throws Exception {
try (ChronicleQueue queue = SingleChronicleQueueBuilder.binary(path).rollCycle(RollCycles.DAILY).build()) {
final ExcerptAppender appender = queue.acquireAppender();
appender.writeText(mapper.writeValueAsString(dto));
}
}
public List readList(String path, Class aClass) throws Exception {
List dtoList = new LinkedList<>();
try (ChronicleQueue queue = SingleChronicleQueueBuilder.binary(path).build()) {
final ExcerptTailer tailer = queue.createTailer();
ChronicleMap<String, Long> indexMap = getReaderIndexMap(queue.fileAbsolutePath());
if (indexMap.containsKey(INDEX)) {
tailer.moveToIndex(indexMap.get(INDEX));
}
while (true) { // something smart ?
String json = tailer.readText();
if (json == null) {
break;
} else {
dtoList.add(mapper.readValue(json, aClass));
}
}
indexMap.put(INDEX, tailer.index());
indexMap.close();
}
return dtoList;
}
public ChronicleMap<String, Long> getReaderIndexMap(String queueName) throws Exception {
ChronicleMapBuilder<String, Long> indexReaderMap = ChronicleMapBuilder.of(String.class, Long.class)
.name("index-reader-map")
.averageKey(INDEX)
.entries(1);
ChronicleMap<String, Long> map = indexReaderMap.createPersistedTo(new File(queueName+"/reader.idx"));
return map;
}
}