QuestDB远程写入最佳实践：避免API误用，使用客户端协议

嵌入式API的局限性：为何不适用于远程连接

CairoEngine是QuestDB核心的嵌入式引擎接口，它被设计用于应用程序直接管理数据库的生命周期和数据文件。这意味着当您通过new CairoEngine(this.configuration)实例化CairoEngine时，它会尝试独占地打开和管理configuration中指定的数据目录下的所有文件，包括_tab_index.d等关键索引文件。

当QuestDB服务器已经作为独立进程运行，并且正在管理着同一数据目录时，应用程序再通过CairoEngine尝试访问该目录，就会产生文件锁冲突。操作系统会阻止第二个进程（即您的应用程序）以读写模式打开已被服务器进程独占的文件，从而导致CairoException。即使您将文件权限设置为777，也无法解决这种进程间的文件锁冲突问题。

以下是尝试使用CairoEngine连接到已运行服务器时可能遇到的典型代码和错误：

// 错误的使用方式：尝试通过CairoEngine连接到已运行的QuestDB服务器
public class MyQuestDbComponent {
    private CairoConfiguration configuration;

    public MyQuestDbComponent(String dataDirPath) {
        // 配置指向已运行QuestDB服务器的数据目录
        this.configuration = new DefaultCairoConfiguration(dataDirPath);
    }

    public void connectAndWrite() {
        try (CairoEngine engine = new CairoEngine(this.configuration)) {
            // ... 尝试通过engine进行数据操作 ...
            System.out.println("CairoEngine initialized successfully (will likely fail if server is running)");
        } catch (Exception e) {
            // 将抛出 io.questdb.cairo.CairoException: [2] could not open read-write [file=/_tab_index.d]
            e.printStackTrace();
            System.err.println("错误：试图通过嵌入式API访问已由QuestDB服务器独占的数据目录。");
        }
    }
}

正确的解决方案：利用QuestDB客户端协议

对于需要连接到独立运行的QuestDB服务器并进行数据交互的场景，正确的做法是使用QuestDB提供的各种客户端协议。QuestDB支持多种客户端协议，以适应不同的应用需求：

1. InfluxDB行协议 (InfluxDB Line Protocol)：这是QuestDB推荐的高吞吐量数据写入协议，尤其适用于时间序列数据的批量插入。它基于TCP/IP，性能卓越。
2. PostgreSQL线协议 (PostgreSQL Wire Protocol)：QuestDB兼容PostgreSQL协议，这意味着您可以使用任何标准的PostgreSQL客户端库（如JDBC）来连接QuestDB，执行SQL查询和DML操作。
3. REST API：QuestDB也提供RESTful接口，用于执行SQL查询和管理任务。

在大多数需要从应用程序向QuestDB服务器写入数据的场景中，InfluxDB行协议因其简洁高效而成为首选。

使用InfluxDB行协议进行数据写入

要通过InfluxDB行协议向QuestDB服务器写入数据，您需要使用QuestDB官方提供的Java客户端库——questdb-client。

1. 添加Maven依赖

首先，在您的pom.xml文件中添加questdb-client的依赖：

    io.questdb
    questdb-client
    7.3.10 

2. 编写数据写入代码

使用Sender类可以方便地通过InfluxDB行协议发送数据。以下是一个示例，展示了如何在Apache Camel的Producer或Spring Boot的服务中实现数据写入逻辑：

import io.questdb.client.Sender;
import io.questdb.griffin.engine.functions.constants.StrConstant; // 注意：这个导入可能不需要，取决于具体版本和用法

import java.io.IOException;
import java.time.Instant;

public class QuestDbLineProtocolWriter {

    private final String questDbHost;
    private final int questDbPort; // InfluxDB Line Protocol 默认端口是9009

    public QuestDbLineProtocolWriter(String questDbHost, int questDbPort) {
        this.questDbHost = questDbHost;
        this.questDbPort = questDbPort;
    }

    public void insertData(String tableName, String bornSymbol, long id, String name) {
        try (Sender sender = Sender.builder()
                .address(questDbHost + ":" + questDbPort)
                .build()) {

            // 插入第一条数据
            sender.table(tableName)
                .symbol("born", bornSymbol)
                .longColumn("id", id)
                .stringColumn("name", name)
                .atNow(); // 使用当前时间戳

            // 示例：插入另一条数据
            sender.table("inventors")
                .symbol("born", "USA")
                .longColumn("id", 1)
                .stringColumn("name", "Thomas Alva Edison")
                .at(Instant.parse("1847-02-11T00:00:00Z")); // 指定时间戳

            sender.flush(); // 确保所有缓冲数据被发送
            System.out.println("数据成功通过InfluxDB行协议写入。");

        } catch (IOException e) {
            System.err.println("通过InfluxDB行协议写入数据失败: " + e.getMessage());
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        // 假设QuestDB服务器运行在本地的9009端口
        QuestDbLineProtocolWriter writer = new QuestDbLineProtocolWriter("lxyrpc01.gsi.de", 9009);
        writer.insertData("inventors", "Austrian Empire", 0, "Nicola Tesla");
    }
}

在上述代码中：

• Sender.builder().address("host:port").build()：用于建立与QuestDB服务器的连接。请确保host和port指向您的QuestDB服务器的InfluxDB行协议监听地址和端口（默认是9009）。
• sender.table("tableName")：指定要写入的表名。如果表不存在，QuestDB会自动创建它。
• symbol("columnName", "value")：用于写入SYMBOL类型的列。
• longColumn("columnName", value)、stringColumn("columnName", "value")等：用于写入其他类型（如LONG, STRING）的列。
• atNow() 或 at(Instant)：指定行的时间戳。atNow()使用当前系统时间，at(Instant)允许您指定一个特定的时间。
• sender.flush(): 在某些情况下，为了确保数据立即发送，可以调用此方法。try-with-resources块结束时，Sender会自动关闭并刷新。

集成考量与注意事项

1. 连接管理：在生产环境中，频繁地创建和关闭Sender实例可能会带来性能开销。考虑在应用程序启动时初始化Sender实例并复用，或者实现一个简单的连接池来管理Sender对象。
2. 错误处理：实际应用中，需要对IOException进行更健壮的处理，包括日志记录、重试机制或故障转移。
3. QuestDB服务器配置：确保您的QuestDB服务器已启用InfluxDB行协议监听，并且防火墙允许来自应用程序的连接访问QuestDB的InfluxDB行协议端口（默认9009）。您可以在QuestDB的server.conf文件中检查相关配置。
4. Camel集成：在Apache Camel中，您可以将上述insertData方法封装在一个自定义的Processor或Producer中，并通过Camel路由调用它。例如，在自定义的QuestDbProducer中，doInsert方法可以调用QuestDbLineProtocolWriter来发送数据。

总结

正确理解QuestDB的连接机制对于高效和稳定地集成至关重要。当您的应用程序需要与一个独立运行的QuestDB服务器交互时，请务必使用其提供的客户端协议，尤其是针对高吞吐量数据写入的InfluxDB行协议。避免将嵌入式API（如CairoEngine）误用于远程连接场景，这将有效避免文件访问冲突和相关异常，确保您的数据能够顺畅、可靠地写入QuestDB。通过采用questdb-client库，您可以轻松地实现这一目标，并充分利用QuestDB的性能优势。