TDengine 与 taosAdapter 的结合（二）

 五、开发实战步骤

（一）环境搭建

在开始 TDengine 与 taosAdapter 结合的 RESTful 接口开发之前，需要先完成相关环境的搭建，包括 TDengine 和 taosAdapter 的安装与配置，以及相关依赖的安装。

• TDengine 安装：

• • Linux 系统：可以从 TDengine 官方网站（https://www.taosdata.com/cn/all-downloads/ ）下载适合系统的安装包，目前支持 x86_64、ARM64 等多种架构 。下载完成后，解压安装包，进入解压目录，执行安装脚本./install.sh。在安装过程中，可能需要设置一些参数，如主机名、集群节点信息等，按照提示进行设置即可。安装完成后，可以通过systemctl start taosd命令启动 TDengine 服务，使用systemctl status taosd命令查看服务状态。

• • Windows 系统：同样从官方网站下载 Windows 版本的安装包，双击安装程序，按照安装向导的提示完成安装。安装完成后，可以在开始菜单中找到 TDengine 的相关程序，或者在安装目录下找到taosd.exe文件，通过命令行启动服务，如taosd -c C:\TDengine\cfg（假设安装目录为 C:\TDengine）。

• taosAdapter 安装与配置：

• • 安装：taosAdapter 通常随 TDengine 安装包一起提供，在安装 TDengine 时会自动安装。如果需要单独安装，可以从 TDengine 官方网站下载对应的 taosAdapter 安装包。在 Linux 系统上，安装完成后，taosAdapter 服务默认由 systemd 管理 。

• • 配置：taosAdapter 的配置文件默认位于/etc/taos/taosadapter.toml（Linux）或C:\TDengine\cfg\taosadapter.toml（Windows）。打开配置文件，可以进行以下常见配置：

• • • 端口配置：默认情况下，taosAdapter 监听 6041 端口提供 RESTful 服务 。如果该端口已被占用，可以修改[http]部分的port参数，指定其他可用端口。

• • • 认证配置：可以配置用户名和密码进行身份认证。在[http]部分，设置user和password参数，确保安全性。

• • • 其他配置：根据实际需求，还可以配置日志级别、连接池大小等参数。例如，修改[log]部分的level参数来调整日志记录的详细程度，将level = "info"改为level = "debug"可以获取更详细的调试信息。

配置完成后，通过systemctl start taosadapter（Linux）或在 Windows 服务中启动 taosAdapter 服务，使用systemctl status taosadapter（Linux）或检查 Windows 服务状态来确认服务是否正常运行。

• 相关依赖安装：根据开发语言和框架的不同，可能需要安装一些相关依赖。如果使用 Python 进行开发，并且使用requests库来发送 HTTP 请求与 taosAdapter 进行交互，可以使用pip install requests命令安装requests库；如果使用 Java 开发，并且使用HttpClient来处理 HTTP 请求，需要在项目的pom.xml文件中添加HttpClient的依赖：

<dependency>

<groupId>org.apache.httpcomponents</groupId>

<artifactId>httpclient</artifactId>

<version>4.5.13</version>

</dependency>

以上步骤完成后，就搭建好了 TDengine 与 taosAdapter 结合开发 RESTful 接口的基础环境。

（二）接口开发流程

完成环境搭建后，就可以开始进行 RESTful 接口的开发了。以下是从需求分析、设计接口到编码实现的具体步骤：

• 需求分析：首先需要明确业务需求，确定需要通过 RESTful 接口对 TDengine 进行哪些数据操作。在一个物联网设备监控项目中，可能需要实现以下功能：

• • 设备数据的实时写入，包括设备 ID、时间戳、各种传感器数据（如温度、湿度、压力等）。

• • 根据设备 ID 和时间范围查询设备的历史数据。

• • 获取所有设备的列表信息。

• 设计接口：根据需求分析的结果，按照 RESTful 架构风格设计接口。设计原则是将 TDengine 中的数据和操作抽象为资源，并使用 HTTP 方法进行访问。

• • 数据写入接口：使用 POST 方法，将设备数据以 JSON 格式发送到/api/v1/databases/{database_name}/tables/{table_name}/data 。假设数据库名为iot_data，表名为device_sensor_data，则接口地址为/api/v1/databases/iot_data/tables/device_sensor_data/data。请求体示例如下：

[

{

"device_id": "device_001",

"ts": "2023-10-01T12:00:00Z",

"temperature": 25.5,

"humidity": 60.0,

"pressure": 1013.2

},

{

"device_id": "device_002",

"ts": "2023-10-01T12:00:00Z",

"temperature": 26.0,

"humidity": 58.0,

"pressure": 1012.8

}

]

• 数据查询接口：使用 GET 方法，通过在 URL 中添加参数来指定查询条件。查询设备device_001在 2023 年 10 月 1 日的所有数据，接口地址可以设计为/api/v1/databases/iot_data/tables/device_sensor_data/data?device_id=device_001&start_time=2023-10-01T00:00:00Z&end_time=2023-10-01T23:59:59Z 。

• 获取设备列表接口：使用 GET 方法，访问/api/v1/databases/{database_name}/devices 。对于上述物联网项目，接口地址为/api/v1/databases/iot_data/devices，返回所有设备的基本信息，如设备 ID、设备名称、设备类型等。

• 编码实现：根据设计好的接口，使用选定的开发语言和框架进行编码实现。如果使用 Python 和 Flask 框架来实现上述接口，可以参考以下代码示例：

from flask import Flask, request, jsonify

import requests

app = Flask(__name__)

# TDengine RESTful 接口地址前缀

TDENGINE_REST_URL = "http://localhost:6041/rest/sql"

TDENGINE_USER = "root"

TDENGINE_PASSWORD = "taosdata"

# 数据写入接口

@app.route('/api/v1/databases/<database_name>/tables/<table_name>/data', methods=['POST'])

def write_data(database_name, table_name):

data = request.json

sql_values = []

for item in data:

device_id = item['device_id']

ts = item['ts']

temperature = item['temperature']

humidity = item['humidity']

pressure = item['pressure']

sql_values.append(f"('{device_id}', '{ts}', {temperature}, {humidity}, {pressure})")

sql = f"INSERT INTO {database_name}.{table_name} (device_id, ts, temperature, humidity, pressure) VALUES {','.join(sql_values)}"

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

return jsonify({"status": "success", "message": "Data inserted successfully"}), 200

else:

return jsonify({"status": "error", "message": "Failed to insert data"}), 500

# 数据查询接口

@app.route('/api/v1/databases/<database_name>/tables/<table_name>/data', methods=['GET'])

def query_data(database_name, table_name):

device_id = request.args.get('device_id')

start_time = request.args.get('start_time')

end_time = request.args.get('end_time')

sql_conditions = []

if device_id:

sql_conditions.append(f"device_id = '{device_id}'")

if start_time:

sql_conditions.append(f"ts >= '{start_time}'")

if end_time:

sql_conditions.append(f"ts <= '{end_time}'")

sql_condition_str = " AND ".join(sql_conditions)

sql = f"SELECT * FROM {database_name}.{table_name}"

if sql_condition_str:

sql += f" WHERE {sql_condition_str}"

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

result = response.json()

return jsonify({"status": "success", "data": result['data']}), 200

else:

return jsonify({"status": "error", "message": "Failed to query data"}), 500

# 获取设备列表接口

@app.route('/api/v1/databases/<database_name>/devices', methods=['GET'])

def get_devices(database_name):

sql = f"SELECT DISTINCT device_id FROM {database_name}.device_sensor_data"

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

result = response.json()

device_list = [item[0] for item in result['data']]

return jsonify({"status": "success", "devices": device_list}), 200

else:

return jsonify({"status": "error", "message": "Failed to get device list"}), 500

if __name__ == '__main__':

app.run(debug=True, host='0.0.0.0', port=5000)

以上代码通过 Flask 框架创建了一个简单的 Web 服务，实现了数据写入、查询和获取设备列表的 RESTful 接口。每个接口通过requests库向 taosAdapter 提供的 RESTful 接口发送 SQL 语句，实现对 TDengine 的操作，并返回相应的结果。

（三）代码示例与解释

下面给出使用 Python 和 Java 调用 RESTful 接口进行数据操作的实际代码示例，并对关键代码进行解释。

Python 代码示例：

import requests

# TDengine RESTful 接口地址前缀

TDENGINE_REST_URL = "http://localhost:6041/rest/sql"

TDENGINE_USER = "root"

TDENGINE_PASSWORD = "taosdata"

# 插入数据示例

def insert_data():

database_name = "test_db"

table_name = "test_table"

data = [

{

"col1": "value1",

"col2": 100,

"ts": "2023-10-05T10:00:00Z"

},

{

"col1": "value2",

"col2": 200,

"ts": "2023-10-05T10:01:00Z"

}

]

sql_values = []

for item in data:

col1 = item['col1']

col2 = item['col2']

ts = item['ts']

sql_values.append(f"('{col1}', {col2}, '{ts}')")

sql = f"INSERT INTO {database_name}.{table_name} (col1, col2, ts) VALUES {','.join(sql_values)}"

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

print("Data inserted successfully")

else:

print("Failed to insert data")

# 查询数据示例

def query_data():

database_name = "test_db"

table_name = "test_table"

sql = f"SELECT * FROM {database_name}.{table_name}"

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

result = response.json()

for row in result['data']:

print(row)

else:

print("Failed to query data")

if __name__ == "__main__":

insert_data()

query_data()

代码解释：

• TDENGINE_REST_URL：定义了 taosAdapter 提供的 RESTful 接口地址，这里假设 taosAdapter 运行在本地，端口为 6041。

• TDENGINE_USER和TDENGINE_PASSWORD：分别为 TDengine 的用户名和密码，用于身份认证。

• insert_data 函数：

• • 构建要插入的数据列表，每个数据项包含col1、col2和ts字段。

• • 通过循环构建 SQL 插入语句的 VALUES 部分，将数据转换为 SQL 可识别的格式。

• • 使用requests.post方法向 TDengine 发送 POST 请求，请求的 URL 为TDENGINE_REST_URL，认证信息为(TDENGINE_USER, TDENGINE_PASSWORD)，请求体为构建好的 SQL 语句。

• • 根据响应状态码判断数据插入是否成功。

• query_data 函数：

• • 构建简单的 SQL 查询语句，查询指定数据库和表中的所有数据。

• • 同样使用requests.post方法发送请求，获取查询结果。

• • 如果响应状态码为 200，解析响应的 JSON 数据，遍历并打印查询结果中的每一行数据。

Java 代码示例：

import org.apache.http.HttpEntity;

import org.apache.http.HttpResponse;

import org.apache.http.NameValuePair;

import org.apache.http.client.HttpClient;

import org.apache.http.client.entity.UrlEncodedFormEntity;

import org.apache.http.client.methods.HttpPost;

import org.apache.http.impl.client.CloseableHttpClient;

import org.apache.http.impl.client.HttpClients;

import org.apache.http.message.BasicNameValuePair;

import org.apache.http.util.EntityUtils;

import java.io.IOException;

import java.util.ArrayList;

import java.util.List;

public class TDengineRestExample {

private static final String TDENGINE_REST_URL = "http://localhost:6041/rest/sql";

private static final String TDENGINE_USER = "root";

private static final String TDENGINE_PASSWORD = "taosdata";

// 插入数据示例

public static void insertData() {

String databaseName = "test_db";

String tableName = "test_table";

List<NameValuePair> data = new ArrayList<>();

data.add(new BasicNameValuePair("col1", "value1"));

data.add(new BasicNameValuePair("col2", "100"));

data.add(new BasicNameValuePair("ts", "2023-10-05T10:00:00Z"));

data.add(new BasicNameValuePair("col1", "value2"));

data.add(new BasicNameValuePair("col2", "200"));

data.add(new BasicNameValuePair("ts", "2023-10-05T10:01:00Z"));

StringBuilder sqlValues = new StringBuilder();

for (int i = 0; i < data.size(); i += 3) {

String col1 = data.get(i).getValue();

String col2 = data.get(i + 1).getValue();

String ts = data.get(i + 2).getValue();

if (sqlValues.length() > 0) {

sqlValues.append(",");

}

sqlValues.append("('").append(col1).append("',").append(col2).append(",'").append(ts).append("')");

}

String sql = "INSERT INTO " + databaseName + "." + tableName + " (col1, col2, ts) VALUES " + sqlValues;

try (CloseableHttpClient httpClient = HttpClients.createDefault()) {

HttpPost httpPost = new HttpPost(TDENGINE_REST_URL);

httpPost.setHeader("Authorization", "Basic " + org.apache.commons.codec.binary.Base64.encodeBase64String((TDENGINE_USER + ":" + TDENGINE_PASSWORD).getBytes()));

httpPost.setEntity(new UrlEncodedFormEntity(List.of(new BasicNameValuePair("sql", sql))));

HttpResponse response = httpClient.execute(httpPost);

int statusCode = response.getStatusLine().getStatusCode();

if (statusCode == 200) {

System.out.println("Data inserted successfully");

} else {

System.out.println("Failed to insert data");

}

} catch (IOException e) {

e.printStackTrace();

}

}

// 查询数据示例

public static void queryData() {

String databaseName = "test_db";

String tableName = "test_table";

String sql = "SELECT * FROM " + databaseName + "." + tableName;

try (CloseableHttpClient httpClient = HttpClients.createDefault()) {

HttpPost httpPost = new HttpPost(TDENGINE_REST_URL);

httpPost.setHeader("Authorization", "Basic " + org.apache.commons.codec.binary.Base64.encodeBase64String((TDENGINE_USER + ":" + TDENGINE_PASSWORD).getBytes()));

httpPost.setEntity(new UrlEncodedFormEntity(List

## 六、常见问题与解决方案

在TDengine与taosAdapter结合进行RESTful接口开发过程中，可能会遇到一些常见问题，以下是对这些问题的分析及相应解决方案：

- **连接失败**：

- **问题描述**：客户端无法连接到taosAdapter提供的RESTful接口，返回连接超时或拒绝连接的错误信息。

- **原因分析**：可能是网络问题，如防火墙阻止了客户端与服务器之间的通信；也可能是taosAdapter服务未正常启动，或者配置的端口被其他程序占用；还可能是主机地址、端口号、用户名、密码等连接参数配置错误。

- **解决方案**：首先检查网络连接，确保客户端和服务器之间的网络畅通，可以通过ping命令测试网络连通性；检查防火墙设置，确保允许客户端访问taosAdapter服务所在的端口，在Linux系统上，可以使用`iptables -I INPUT -p tcp --dport 6041 -j ACCEPT`命令允许访问6041端口（假设taosAdapter使用该端口） ；确认taosAdapter服务已正常启动，使用`systemctl status taosadapter`命令查看服务状态，如果服务未启动，使用`systemctl start taosadapter`命令启动服务；仔细核对连接参数，确保主机地址、端口号、用户名、密码等配置正确无误。

- **数据传输错误**：

- **问题描述**：在进行数据写入或查询时，出现数据传输错误，如数据丢失、数据格式错误等。

- **原因分析**：数据丢失可能是由于网络不稳定，在数据传输过程中出现丢包现象；数据格式错误可能是因为客户端发送的数据格式不符合TDengine的要求，或者在数据解析过程中出现问题。

- **解决方案**：对于网络不稳定导致的数据丢失问题，可以增加重试机制。在代码中，使用循环和异常处理来实现数据传输失败后的重试操作。在Python中使用`requests`库进行数据写入时：

```python

import requests

import time

TDENGINE_REST_URL = "http://localhost:6041/rest/sql"

TDENGINE_USER = "root"

TDENGINE_PASSWORD = "taosdata"

data = [

{

"col1": "value1",

"col2": 100,

"ts": "2023-10-05T10:00:00Z"

}

]

sql_values = []

for item in data:

col1 = item['col1']

col2 = item['col2']

ts = item['ts']

sql_values.append(f"('{col1}', {col2}, '{ts}')")

sql = f"INSERT INTO test_db.test_table (col1, col2, ts) VALUES {','.join(sql_values)}"

max_retries = 3

retry_delay = 2

for retry in range(max_retries):

try:

response = requests.post(TDENGINE_REST_URL, auth=(TDENGINE_USER, TDENGINE_PASSWORD), data=sql)

if response.status_code == 200:

print("Data inserted successfully")

break

except Exception as e:

print(f"Failed to insert data, retry {retry + 1}: {e}")

time.sleep(retry_delay)

这段代码中，设置了最大重试次数为 3 次，每次重试间隔 2 秒。如果数据传输失败，会捕获异常并进行重试，直到成功插入数据或达到最大重试次数。对于数据格式错误问题，要仔细检查数据格式，确保符合 TDengine 的要求。在进行数据写入时，严格按照 TDengine 支持的数据类型和格式组织数据，在使用 JSON 格式传输数据时，确保 JSON 结构正确，字段名称和数据类型与 TDengine 表结构一致。在解析查询结果时，也要根据 TDengine 返回的数据格式进行正确的解析。

• taosAdapter 无响应：

• • 问题描述：客户端发送请求后，taosAdapter 长时间无响应，服务端也没有返回任何错误信息。

• • 原因分析：可能是 taosAdapter 负载过高，无法及时处理请求；也可能是请求处理过程中出现死锁或其他异常情况，导致 taosAdapter 服务挂起；还可能是某些频繁请求的操作（如健康检查语句）导致服务资源耗尽。

• • 解决方案：监控 taosAdapter 的性能指标，如 CPU 使用率、内存占用、请求队列长度等，使用top、htop等工具查看系统资源使用情况 。如果发现 taosAdapter 负载过高，可以考虑优化系统配置，增加服务器资源，如内存、CPU 等；对 taosAdapter 进行性能调优，如调整连接池大小、优化 SQL 语句等。在 taosadapter.toml 配置文件中，可以适当增加[http]部分的max_connections参数值，以提高并发处理能力 。如果是请求处理过程中出现死锁或异常情况，可以通过查看 taosAdapter 的日志文件（默认位于/var/log/taos/taosadapter.log ）来排查问题。根据日志信息定位异常原因，如 SQL 语句错误、数据库连接问题等，进行相应的修复。对于由于频繁请求某些操作导致的问题，要优化请求策略。在黑格智能 3D 打印业务从 2.x 升级到 3.x 过程中，微服务通过 restful 方式连接 TDengine 时，taosAdapter 出现无响应但 taosd 服务正常的现象，经排查是大量使用 “show cluster alive” 作为微服务监听语句的频繁请求导致。后续将健康检查语句更换为 “select 1”，顺利解决了这个问题 。

七、总结与展望

通过本文的介绍和实践，我们深入了解了 TDengine 与 taosAdapter 结合在 RESTful 接口开发中的应用。TDengine 作为一款高性能的时序数据库，具备强大的时序数据处理能力，而 taosAdapter 则为 TDengine 提供了便捷的 RESTful 接口支持，使得 TDengine 能够更轻松地与各种应用程序集成。

在实际应用中，TDengine 与 taosAdapter 的结合展现出了诸多优势，便捷的跨平台访问能力，让不同操作系统和编程语言的应用都能与 TDengine 进行交互；简化了系统集成过程，降低了与现有系统集成的难度和成本；灵活的数据交互方式，满足了多样化的应用场景需求。通过开发实战步骤，我们学习了如何搭建环境、设计接口并进行编码实现，掌握了使用 Python 和 Java 调用 RESTful 接口进行数据操作的方法。同时，我们也了解了在开发过程中可能遇到的常见问题及解决方案，为实际项目的开发提供了保障。

展望未来，随着物联网、工业互联网等领域的不断发展，时序数据的处理需求将持续增长。TDengine 和 taosAdapter 也将不断演进和完善。在功能方面，可能会进一步优化 RESTful 接口的性能和稳定性，增加更多对复杂查询和数据分析的支持，以满足企业日益增长的业务需求；在兼容性方面，有望支持更多的协议和标准，与更多的第三方工具和系统实现无缝对接，进一步拓展其应用场景；在生态建设方面，社区可能会更加活跃，吸引更多的开发者参与，共同推动 TDengine 和 taosAdapter 的发展，为时序数据处理领域提供更强大、更完善的解决方案。作为开发者，我们应持续关注 TDengine 和 taosAdapter 的发展动态，不断探索其在不同场景下的应用，充分发挥其优势，为项目的成功实施提供有力支持 。