Chronicle Consumer 没有正确读取记录？

Question

我正在使用编年史队列 (5.16.13) 将 json 值写入和读取编年史文件。为了编写对象，我在循环中使用以下内容

try (final DocumentContext dc = appender.writingDocument()) {
        dc.wire().write(() -> "msg").text("Hallo asdf");
        System.out.println("your data was store to index="+ dc.index());
        return true;
    } catch (Exception e) {
        logger.warn("Unable to store value to chronicle", e);
        return false;
    }

为了阅读项目，我循环执行以下调用

DocumentContext documentContext;
    do {
        documentContext = tailer.readingDocument();
        currentOffset = documentContext.index();
        System.out.println("Current offset: " + currentOffset);
    } while (!documentContext.isData());

我观察到变量 currentOffset 没有改变，一段时间后（似乎取决于负载大小）循环变为无限，当前偏移量具有疯狂的值。第一个循环的输出（缩短）是

Writing 0
your data was store to index=76385993359360
Writing 1
your data was store to index=76385993359361
Writing 2
your data was store to index=76385993359362
Writing 3
your data was store to index=76385993359363
Writing 4
your data was store to index=76385993359364
Writing 5
your data was store to index=76385993359365
Writing 6
your data was store to index=76385993359366
Writing 7
your data was store to index=76385993359367
Writing 8
your data was store to index=76385993359368
Writing 9
your data was store to index=76385993359369
Writing 10
your data was store to index=76385993359370
Writing 11
your data was store to index=76385993359371
Writing 12
your data was store to index=76385993359372
Writing 13
your data was store to index=76385993359373
Writing 14
your data was store to index=76385993359374
Writing 15
your data was store to index=76385993359375
Writing 16
your data was store to index=76385993359376
Writing 17
your data was store to index=76385993359377
Writing 18
your data was store to index=76385993359378
Writing 19
your data was store to index=76385993359379
Writing 20
your data was store to index=76385993359380
Writing 21
your data was store to index=76385993359381
Writing 22
your data was store to index=76385993359382
Writing 23
your data was store to index=76385993359383
Writing 24
your data was store to index=76385993359384
Writing 25
your data was store to index=76385993359385
Writing 26
your data was store to index=76385993359386

第二个循环

Reading 0
Current offset: 76385993359360
Reading 1
Current offset: 76385993359360
Reading 2
Current offset: 76385993359360
Reading 3
Current offset: 76385993359360
Reading 4
Current offset: 76385993359360
Reading 5
Current offset: 76385993359360
Reading 6
Current offset: 76385993359360
Reading 7
Current offset: 76385993359360
Reading 8
Current offset: 76385993359360
Reading 9
Current offset: 76385993359360
Reading 10
Current offset: 76385993359360
Reading 11
Current offset: 76385993359360
Reading 12
Current offset: 76385993359360
Reading 13
Current offset: 76385993359360
Reading 14
Current offset: 76385993359360
Reading 15
Current offset: 76385993359360
Reading 16
Current offset: 76385993359360
Reading 17
Current offset: 76385993359360
Reading 18
Current offset: 76385993359360
Reading 19
Current offset: 76385993359360
Reading 20
Current offset: 76385993359360
Reading 21
Current offset: 76385993359360
Reading 22
Current offset: 76385993359360
Reading 23
Current offset: 76385993359360
Reading 24
Current offset: 76385993359360
Reading 25
Current offset: -9223372036854775808

我做错了什么吗？那么有人可以提示我正确的用法吗？

非常感谢！

编辑：添加了最小的工作示例

以下单元测试对我来说失败了。

@Test
public void fails() throws Exception {
    String basePath = System.getProperty("java.io.tmpdir");
    String path = Files.createTempDirectory(Paths.get(basePath), "chronicle-")
            .toAbsolutePath()
            .toString();
    logger.info("Using temp path '{}'", path);

    SingleChronicleQueue chronicleQueue = SingleChronicleQueueBuilder
            .single()
            .path(path)
            .build();

    // Create Appender
    ExcerptAppender appender = chronicleQueue.acquireAppender();

    // Create Tailer
    ExcerptTailer tailer = chronicleQueue.createTailer();
    tailer.toStart();

    int numberOfRecords = 10;

    // Write
    for (int i = 0; i <= numberOfRecords; i++) {
        System.out.println("Writing " + i);
        try (final DocumentContext dc = appender.writingDocument()) {
            dc.wire().write(() -> "msg").text("Hello World!");
            System.out.println("your data was store to index=" + dc.index());
        } catch (Exception e) {
            logger.warn("Unable to store value to chronicle", e);
        }
    }
    // Read
    for (int i = 0; i <= numberOfRecords; i++) {
        System.out.println("Reading " + i);
        DocumentContext documentContext = tailer.readingDocument();
        long currentOffset = documentContext.index();
        System.out.println("Current offset: " + currentOffset);

        Wire wire = documentContext.wire();

        if (wire != null) {
            String msg = wire
                    .read("msg")
                    .text();
        }
    }

    chronicleQueue.close();
}

输出为

Writing 0
your data was store to index=76385993359360
Writing 1
your data was store to index=76385993359361
Writing 2
your data was store to index=76385993359362
Writing 3
your data was store to index=76385993359363
Writing 4
your data was store to index=76385993359364
Writing 5
your data was store to index=76385993359365
Writing 6
your data was store to index=76385993359366
Writing 7
your data was store to index=76385993359367
Writing 8
your data was store to index=76385993359368
Writing 9
your data was store to index=76385993359369
Writing 10
your data was store to index=76385993359370
Reading 0
Current offset: 76385993359360
Reading 1
Current offset: 76385993359360
Reading 2
Current offset: 76385993359360
Reading 3
Current offset: 76385993359360
Reading 4
Current offset: -9223372036854775808
Reading 5
Current offset: -9223372036854775808
Reading 6
Current offset: -9223372036854775808
Reading 7
Current offset: -9223372036854775808
Reading 8
Current offset: -9223372036854775808
Reading 9
Current offset: -9223372036854775808
Reading 10
Current offset: -9223372036854775808

Answer 1

我自己使用@PeterLawrey 的建议找到了答案，并将文档上下文包装在 try-with-resources 中。这解决了这个问题。请参阅下面更正后的代码段

@Test
public void works() throws Exception {
    String basePath = System.getProperty("java.io.tmpdir");
    String path = Files.createTempDirectory(Paths.get(basePath), "chronicle-")
            .toAbsolutePath()
            .toString();
    logger.info("Using temp path '{}'", path);

    SingleChronicleQueue chronicleQueue = SingleChronicleQueueBuilder
            .single()
            .path(path)
            .build();

    // Create Appender
    ExcerptAppender appender = chronicleQueue.acquireAppender();

    // Create Tailer
    ExcerptTailer tailer = chronicleQueue.createTailer();
    tailer.toStart();

    int numberOfRecords = 10;

    // Write
    for (int i = 0; i <= numberOfRecords; i++) {
        System.out.println("Writing " + i);
        try (final DocumentContext dc = appender.writingDocument()) {
            dc.wire().write(() -> "msg").text("Hello World!");
            System.out.println("your data was store to index=" + dc.index());
        } catch (Exception e) {
            logger.warn("Unable to store value to chronicle", e);
        }
    }
    // Read
    for (int i = 0; i <= numberOfRecords; i++) {
        System.out.println("Reading " + i);
        try (DocumentContext documentContext = tailer.readingDocument()) {
            long currentOffset = documentContext.index();
            System.out.println("Current offset: " + currentOffset);

            Wire wire = documentContext.wire();

            if (wire != null) {
                String msg = wire
                        .read("msg")
                        .text();
            }
        }
    }

    chronicleQueue.close();
}

产生预期的输出

Writing 0
your data was store to index=76385993359360
Writing 1
your data was store to index=76385993359361
Writing 2
your data was store to index=76385993359362
Writing 3
your data was store to index=76385993359363
Writing 4
your data was store to index=76385993359364
Writing 5
your data was store to index=76385993359365
Writing 6
your data was store to index=76385993359366
Writing 7
your data was store to index=76385993359367
Writing 8
your data was store to index=76385993359368
Writing 9
your data was store to index=76385993359369
Writing 10
your data was store to index=76385993359370
Reading 0
Current offset: 76385993359360
Reading 1
Current offset: 76385993359361
Reading 2
Current offset: 76385993359362
Reading 3
Current offset: 76385993359363
Reading 4
Current offset: 76385993359364
Reading 5
Current offset: 76385993359365
Reading 6
Current offset: 76385993359366
Reading 7
Current offset: 76385993359367
Reading 8
Current offset: 76385993359368
Reading 9
Current offset: 76385993359369
Reading 10
Current offset: 76385993359370

希望这对其他人有帮助。

Answer 2

使用 DocumentContext 旨在成为较低级别的接口之一，并不符合每个人的口味。我赞成使用 MethodReader/MethodWriter 方法，除非你有理由在较低级别工作。

@Test
public void works() {
    String path = OS.TMP + "/chronicle-" + System.nanoTime();
    System.out.println("Using temp path " + path);

    try (SingleChronicleQueue queue = SingleChronicleQueueBuilder
            .single()
            .path(path)
            .build()) {

        ExcerptAppender appender = queue.acquireAppender();
        Messager messager = appender.methodWriter(Messager.class);

        int numberOfRecords = 10;

        // Write
        for (int i = 0; i <= numberOfRecords; i++) {
            System.out.print("Writing " + i);
            messager.msg("Hello World!");
            System.out.println(", your data was stored at index=" + appender.lastIndexAppended());
        }

        ExcerptTailer tailer = queue.createTailer();
        MethodReader reader = tailer.methodReader((Messager) msg -> {
            System.out.println("Current offset: " + tailer.index()
                    + " msg: " + msg);
        });

        // Read
        while (reader.readOne()) {
            // busy wait.
        }
    }
}

这会打印

Using temp path C:\Users\peter\AppData\Local\Temp\/chronicle-412979753710181
[main] DEBUG net.openhft.chronicle.bytes.MappedFile - Allocation of 0 chunk in C:\Users\peter\AppData\Local\Temp\chronicle-412979753710181\metadata.cq4t took 15.418 ms.
[main] DEBUG net.openhft.chronicle.bytes.MappedFile - Allocation of 0 chunk in C:\Users\peter\AppData\Local\Temp\chronicle-412979753710181181226.cq4 took 25.061 ms.
Writing 0, your data was stored at index=76841259892736
Writing 1, your data was stored at index=76841259892737
Writing 2, your data was stored at index=76841259892738
Writing 3, your data was stored at index=76841259892739
Writing 4, your data was stored at index=76841259892740
Writing 5, your data was stored at index=76841259892741
Writing 6, your data was stored at index=76841259892742
Writing 7, your data was stored at index=76841259892743
Writing 8, your data was stored at index=76841259892744
Writing 9, your data was stored at index=76841259892745
Writing 10, your data was stored at index=76841259892746

Current offset: 76841259892736 msg: Hello World!
Current offset: 76841259892737 msg: Hello World!
Current offset: 76841259892738 msg: Hello World!
Current offset: 76841259892739 msg: Hello World!
Current offset: 76841259892740 msg: Hello World!
Current offset: 76841259892741 msg: Hello World!
Current offset: 76841259892742 msg: Hello World!
Current offset: 76841259892743 msg: Hello World!
Current offset: 76841259892744 msg: Hello World!
Current offset: 76841259892745 msg: Hello World!
Current offset: 76841259892746 msg: Hello World!
[main] DEBUG net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder - File released C:\Users\peter\AppData\Local\Temp\chronicle-412979753710181181226.cq4
[main] DEBUG net.openhft.chronicle.queue.impl.single.SingleChronicleQueueBuilder - File released C:\Users\peter\AppData\Local\Temp\chronicle-412979753710181181226.cq4

注意：这将写入与原始数据相同的数据 post。

使用这种接口方法的一个优点是您可以完全使用方法接口和 DTO 来实现您的业务组件，而根本不使用 Chronicle（或在传输中）。当您从测试中删除传输时，这会简化测试业务逻辑。

Chronicle Consumer 没有正确读取记录？

Chronicle Consumer not reading records correctly?

java

chronicle

chronicle-queue

编辑：添加了最小的工作示例