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行协议因其简洁高效而成为首选。

Meku

AI应用和网页开发工具

下载

使用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");
    }
}

在上述代码中：