数据订阅

TDengine 提供了类似于消息队列产品的数据订阅和消费接口。在许多场景中，采用 TDengine 的时序大数据平台，无须再集成消息队列产品，从而简化应用程序设计并降低运维成本。本章介绍各语言连接器数据订阅的相关 API 以及使用方法。 数据订阅的基础知识请参考 数据订阅

创建主题

请用 taos shell 或者 参考 执行 SQL 章节用程序执行创建主题的 SQL：CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters

上述 SQL 将创建一个名为 topic_meters 的订阅。使用该订阅所获取的消息中的每条记录都由此查询语句 SELECT ts, current, voltage, phase, groupid, location FROM meters 所选择的列组成。

注意 在 TDengine 连接器实现中，对于订阅查询，有以下限制。

• 查询语句限制：订阅查询只能使用 select 语句，并不支持其他类型的SQL，如订阅库，订阅超级表（非 select 方式），insert、update 或 delete 等。
• 原始始数据查询：订阅查询只能查询原始数据，而不能查询聚合或计算结果。
• 时间顺序限制：订阅查询只能按照时间正序查询数据。

创建消费者

TDengine 消费者的概念跟 Kafka 类似，消费者通过订阅主题来接收数据流。消费者可以配置多种参数，如连接方式、服务器地址、自动提交 Offset 等以适应不同的数据处理需求。有的语言连接器的消费者还支持自动重连和数据传输压缩等高级功能，以确保数据的高效和稳定接收。

创建参数

创建消费者的参数较多，非常灵活的支持了各种连接类型、 Offset 提交方式、压缩、重连、反序列化等特性。各语言连接器都适用的通用基础配置项如下表所示：

下面是各语言连接器创建参数：

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C
• REST API

Java 连接器创建消费者的参数为 Properties， 可以设置的参数列表请参考 消费者参数 其他参数请参考上文通用基础配置项。

提供了 td.connect.websocket.scheme 参数来表示协议类型，其他参数同通用基础配置项。

创建消费者支持属性列表：

• ws.url：WebSocket 连接地址。
• ws.message.channelLen：WebSocket 消息通道缓存长度，默认 0。
• ws.message.timeout：WebSocket 消息超时时间，默认 5m。
• ws.message.writeWait：WebSocket 写入消息超时时间，默认 10s。
• ws.message.enableCompression：WebSocket 是否启用压缩，默认 false。
• ws.autoReconnect：WebSocket 是否自动重连，默认 false。
• ws.reconnectIntervalMs：WebSocket 重连间隔时间毫秒，默认 2000。
• ws.reconnectRetryCount：WebSocket 重连重试次数，默认 3。

其他参数见上表。

Rust 连接器创建消费者的参数为 DSN， 可以设置的参数列表请参考 DSN 其他参数请参考上文通用基础配置项。

提供了 WS_URL 参数来表示连接到的服务器地址，其他参数同通用基础配置项。

创建消费者支持属性列表：

• useSSL：是否使用 SSL 连接，默认为 false。
• token：连接 TDengine cloud 的 token。
• ws.message.enableCompression：是否启用 WebSocket 压缩，默认为 false。
• ws.autoReconnect：是否自动重连，默认为 false。
• ws.reconnect.retry.count：重连次数，默认为 3。
• ws.reconnect.interval.ms：重连间隔毫秒时间，默认为 2000。

其他参数见上表。

• WebSocket 连接: 因为使用 dsn，不需要 td.connect.ip，td.connect.port，td.connect.user 和 td.connect.pass 四个配置项，其余同通用配置项。
• 原生连接: 同通用基础配置项。

不支持

WebSocket 连接

介绍各语言连接器使用 WebSocket 连接方式创建消费者。指定连接的服务器地址，设置自动提交，从最新消息开始消费，指定 group.id 和 client.id 等信息。有的语言的连接器还支持反序列化参数。

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C
• REST API

1. Properties config = new Properties();
2. config.setProperty("td.connect.type", "ws");
3. config.setProperty("bootstrap.servers", "localhost:6041");
4. config.setProperty("auto.offset.reset", "latest");
5. config.setProperty("msg.with.table.name", "true");
6. config.setProperty("enable.auto.commit", "true");
7. config.setProperty("auto.commit.interval.ms", "1000");
8. config.setProperty("group.id", "group1");
9. config.setProperty("client.id", "clinet1");
10. config.setProperty("td.connect.user", "root");
11. config.setProperty("td.connect.pass", "taosdata");
12. config.setProperty("value.deserializer", "com.taos.example.WsConsumerLoopFull$ResultDeserializer");
13. config.setProperty("value.deserializer.encoding", "UTF-8");

15. try {
16. TaosConsumer<ResultBean> consumer= new TaosConsumer<>(config);
17. System.out.printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s%n",
18. config.getProperty("bootstrap.servers"),
19. config.getProperty("group.id"),
20. config.getProperty("client.id"));
21. return consumer;
22. } catch (Exception ex) {
23. // please refer to the JDBC specifications for detailed exceptions info
24. System.out.printf("Failed to create websocket consumer, host: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
25. config.getProperty("bootstrap.servers"),
26. config.getProperty("group.id"),
27. config.getProperty("client.id"),
28. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
29. ex.getMessage());
30. // Print stack trace for context in examples. Use logging in production.
31. ex.printStackTrace();
32. throw ex;
33. }

查看源码

1. def create_consumer():
2. try:
3. consumer = taosws.Consumer(conf={
4. "td.connect.websocket.scheme": tdConnWsScheme,
5. "group.id": groupId,
6. "client.id": clientId,
7. "auto.offset.reset": autoOffsetReset,
8. "td.connect.ip": host,
9. "td.connect.port": port,
10. "enable.auto.commit": autoCommitState,
11. "auto.commit.interval.ms": autoCommitIntv,
12. })
13. print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}.");
14. return consumer;
15. except Exception as err:
16. print(f"Failed to create websocket consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.");
17. raise err

查看源码

1. // create consumer
2. wsUrl := "ws://127.0.0.1:6041"
3. groupID = "group1"
4. clientID = "client1"
5. host = "127.0.0.1"
6. consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
7. "ws.url":                  wsUrl,
8. "ws.message.channelLen":   uint(0),
9. "ws.message.timeout":      common.DefaultMessageTimeout,
10. "ws.message.writeWait":    common.DefaultWriteWait,
11. "td.connect.user":         "root",
12. "td.connect.pass":         "taosdata",
13. "auto.offset.reset":       "latest",
14. "msg.with.table.name":     "true",
15. "enable.auto.commit":      "true",
16. "auto.commit.interval.ms": "1000",
17. "group.id":                groupID,
18. "client.id":               clientID,
19. })
20. if err != nil {
21. log.Fatalf(
22. "Failed to create websocket consumer, host: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
23. host,
24. groupID,
25. clientID,
26. err.Error(),
27. )
28. }
29. log.Printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s\n", host, groupID, clientID)

查看源码

1. let dsn = "ws://localhost:6041".to_string();
2. println!("dsn: {}", dsn);
3. let mut dsn = Dsn::from_str(&dsn)?;

查看源码

1. let group_id = "group1".to_string();
2. let client_id = "client1".to_string();
3. dsn.params
4. .insert("auto.offset.reset".to_string(), "latest".to_string());
5. dsn.params
6. .insert("msg.with.table.name".to_string(), "true".to_string());
7. dsn.params
8. .insert("enable.auto.commit".to_string(), "true".to_string());
9. dsn.params
10. .insert("auto.commit.interval.ms".to_string(), "1000".to_string());
11. dsn.params.insert("group.id".to_string(), group_id.clone());
12. dsn.params
13. .insert("client.id".to_string(), client_id.clone());

15. let builder = TmqBuilder::from_dsn(&dsn)?;
16. let mut consumer = match builder.build().await {
17. Ok(consumer) => {
18. println!(
19. "Create consumer successfully, dsn: {}, groupId: {}, clientId: {}.",
20. dsn, group_id, client_id
21. );
22. consumer
23. }
24. Err(err) => {
25. eprintln!("Failed to create websocket consumer, dsn: {}, groupId: {}, clientId: {}, ErrMessage: {:?}", dsn, group_id, client_id, err);
26. return Err(err.into());
27. }
28. };

查看源码

1. const db = 'power';
2. const stable = 'meters';
3. const url = 'ws://localhost:6041';
4. const topic = 'topic_meters'
5. const topics = [topic];
6. const groupId = "group1";
7. const clientId = "client1";

9. async function createConsumer() {

11. let groupId = "group1";
12. let clientId = "client1";
13. let configMap = new Map([
14. [taos.TMQConstants.GROUP_ID, groupId],
15. [taos.TMQConstants.CLIENT_ID, clientId],
16. [taos.TMQConstants.CONNECT_USER, "root"],
17. [taos.TMQConstants.CONNECT_PASS, "taosdata"],
18. [taos.TMQConstants.AUTO_OFFSET_RESET, "latest"],
19. [taos.TMQConstants.WS_URL, url],
20. [taos.TMQConstants.ENABLE_AUTO_COMMIT, 'true'],
21. [taos.TMQConstants.AUTO_COMMIT_INTERVAL_MS, '1000']
22. ]);
23. try {
24. conn = await taos.tmqConnect(configMap);
25. console.log(`Create consumer successfully, host: ${url}, groupId: ${groupId}, clientId: ${clientId}`)
26. return conn;
27. } catch (err) {
28. console.error(`Failed to create websocket consumer, topic: ${topic}, groupId: ${groupId}, clientId: ${clientId}, ErrCode: ${err.code}, ErrMessage: ${err.message}`);
29. throw err;
30. }

32. }

查看源码

1. // consumer config
2. _host = "127.0.0.1";
3. _groupId = "group1";
4. _clientId = "client1";
5. var cfg = new Dictionary<string, string>()
6. {
7. { "td.connect.type", "WebSocket" },
8. { "td.connect.port", "6041" },
9. { "auto.offset.reset", "latest" },
10. { "msg.with.table.name", "true" },
11. { "enable.auto.commit", "true" },
12. { "auto.commit.interval.ms", "1000" },
13. { "group.id", _groupId },
14. { "client.id", _clientId },
15. { "td.connect.ip", _host },
16. { "td.connect.user", "root" },
17. { "td.connect.pass", "taosdata" },
18. };
19. IConsumer<Dictionary<string, object>> consumer = null!;
20. try
21. {
22. // create consumer
23. consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build();
24. Console.WriteLine(
25. $"Create consumer successfully, " +
26. $"host: {_host}, " +
27. $"groupId: {_groupId}, " +
28. $"clientId: {_clientId}");
29. }
30. catch (TDengineError e)
31. {
32. // handle TDengine error
33. Console.WriteLine(
34. $"Failed to create native consumer, " +
35. $"host: {_host}, " +
36. $"groupId: {_groupId}, " +
37. $"clientId: {_clientId}, " +
38. $"ErrCode: {e.Code}, " +
39. $"ErrMessage: {e.Error}");
40. throw;
41. }
42. catch (Exception e)
43. {
44. // handle other exceptions
45. Console.WriteLine(
46. $"Failed to create native consumer, " +
47. $"host: {_host}, " +
48. $"groupId: {_groupId}, " +
49. $"clientId: {_clientId}, " +
50. $"ErrMessage: {e.Message}");
51. throw;
52. }

查看源码

1. ws_tmq_t* build_consumer(const ConsumerConfig* config) {
2. ws_tmq_conf_res_t code;
3. ws_tmq_t*         tmq = NULL;

5. // create a configuration object
6. ws_tmq_conf_t* conf = ws_tmq_conf_new();

8. // set the configuration parameters
9. code = ws_tmq_conf_set(conf, "enable.auto.commit", config->enable_auto_commit);
10. if (WS_TMQ_CONF_OK != code) {
11. ws_tmq_conf_destroy(conf);
12. return NULL;
13. }
14. code = ws_tmq_conf_set(conf, "auto.commit.interval.ms", config->auto_commit_interval_ms);
15. if (WS_TMQ_CONF_OK != code) {
16. ws_tmq_conf_destroy(conf);
17. return NULL;
18. }
19. code = ws_tmq_conf_set(conf, "group.id", config->group_id);
20. if (WS_TMQ_CONF_OK != code) {
21. ws_tmq_conf_destroy(conf);
22. return NULL;
23. }
24. code = ws_tmq_conf_set(conf, "client.id", config->client_id);
25. if (WS_TMQ_CONF_OK != code) {
26. ws_tmq_conf_destroy(conf);
27. return NULL;
28. }

30. code = ws_tmq_conf_set(conf, "auto.offset.reset", config->auto_offset_reset);
31. if (WS_TMQ_CONF_OK != code) {
32. ws_tmq_conf_destroy(conf);
33. return NULL;
34. }

36. // create a consumer object
37. tmq = ws_tmq_consumer_new(conf, "taos://localhost:6041", NULL, 0);

39. _end:
40. // destroy the configuration object
41. ws_tmq_conf_destroy(conf);
42. return tmq;
43. }

查看源码

1. ws_tmq_t* tmq = build_consumer(&config);
2. if (NULL == tmq) {
3. fprintf(stderr, "Failed to create native consumer, host: %s, groupId: %s, , clientId: %s.\n",
4. config.td_connect_host, config.group_id, config.client_id);
5. return -1;
6. } else {
7. fprintf(stdout, "Create consumer successfully, host: %s, groupId: %s, clientId: %s.\n", config.td_connect_host,
8. config.group_id, config.client_id);
9. }

查看源码

调用 build_consumer 函数尝试获取消费者实例 tmq。成功则打印成功日志，失败则打印失败日志。

不支持

原生连接

介绍各语言连接器使用原生连接方式创建消费者。指定连接的服务器地址，设置自动提交，从最新消息开始消费，指定 group.id 和 client.id 等信息。有的语言的连接器还支持反序列化参数。

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C
• REST API

1. Properties config = new Properties();
2. config.setProperty("td.connect.type", "jni");
3. config.setProperty("bootstrap.servers", "localhost:6030");
4. config.setProperty("auto.offset.reset", "latest");
5. config.setProperty("msg.with.table.name", "true");
6. config.setProperty("enable.auto.commit", "true");
7. config.setProperty("auto.commit.interval.ms", "1000");
8. config.setProperty("group.id", "group1");
9. config.setProperty("client.id", "clinet1");
10. config.setProperty("td.connect.user", "root");
11. config.setProperty("td.connect.pass", "taosdata");
12. config.setProperty("value.deserializer", "com.taos.example.ConsumerLoopFull$ResultDeserializer");
13. config.setProperty("value.deserializer.encoding", "UTF-8");

15. try {
16. TaosConsumer<ResultBean> consumer= new TaosConsumer<>(config);
17. System.out.printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s%n",
18. config.getProperty("bootstrap.servers"),
19. config.getProperty("group.id"),
20. config.getProperty("client.id"));
21. return consumer;
22. } catch (Exception ex) {
23. // please refer to the JDBC specifications for detailed exceptions info
24. System.out.printf("Failed to create native consumer, host: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
25. config.getProperty("bootstrap.servers"),
26. config.getProperty("group.id"),
27. config.getProperty("client.id"),
28. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
29. ex.getMessage());
30. // Print stack trace for context in examples. Use logging in production.
31. ex.printStackTrace();
32. throw ex;
33. }

查看源码

1. from taos.tmq import Consumer

3. def create_consumer():
4. try:
5. consumer = Consumer(
6. {
7. "group.id": groupId,
8. "client.id": clientId,
9. "td.connect.user": user,
10. "td.connect.pass": password,
11. "enable.auto.commit": autoCommitState,
12. "auto.commit.interval.ms": autoCommitIntv,
13. "auto.offset.reset": autoOffsetReset,
14. "td.connect.ip": host,
15. "td.connect.port": str(port),
16. }
17. )
18. print(f"Create consumer successfully, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}")
19. return consumer
20. except Exception as err:
21. print(f"Failed to create native consumer, host: {host}:{port}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
22. raise err

查看源码

1. // create consumer
2. groupID = "group1"
3. clientID = "client1"
4. host = "127.0.0.1"
5. consumer, err := tmq.NewConsumer(&tmqcommon.ConfigMap{
6. "td.connect.user":         "root",
7. "td.connect.pass":         "taosdata",
8. "auto.offset.reset":       "latest",
9. "msg.with.table.name":     "true",
10. "enable.auto.commit":      "true",
11. "auto.commit.interval.ms": "1000",
12. "group.id":                groupID,
13. "client.id":               clientID,
14. })
15. if err != nil {
16. log.Fatalf(
17. "Failed to create native consumer, host: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
18. host,
19. groupID,
20. clientID,
21. err.Error(),
22. )
23. }
24. log.Printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s\n", host, groupID, clientID)

查看源码

1. let dsn = "taos://localhost:6030".to_string();
2. println!("dsn: {}", dsn);
3. let mut dsn = Dsn::from_str(&dsn)?;

查看源码

1. let group_id = "group1".to_string();
2. let client_id = "client1".to_string();
3. dsn.params
4. .insert("auto.offset.reset".to_string(), "latest".to_string());
5. dsn.params
6. .insert("msg.with.table.name".to_string(), "true".to_string());
7. dsn.params
8. .insert("enable.auto.commit".to_string(), "true".to_string());
9. dsn.params
10. .insert("auto.commit.interval.ms".to_string(), "1000".to_string());
11. dsn.params.insert("group.id".to_string(), group_id.clone());
12. dsn.params
13. .insert("client.id".to_string(), client_id.clone());

15. let builder = TmqBuilder::from_dsn(&dsn)?;
16. let mut consumer = match builder.build().await {
17. Ok(consumer) => {
18. println!(
19. "Create consumer successfully, dsn: {}, groupId: {}, clientId: {}.",
20. dsn, group_id, client_id
21. );
22. consumer
23. }
24. Err(err) => {
25. eprintln!("Failed to create native consumer, dsn: {}, groupId: {}, clientId: {}, ErrMessage: {:?}", dsn, group_id, client_id, err);
26. return Err(err.into());
27. }
28. };

查看源码

不支持

1. // consumer config
2. _host = "127.0.0.1";
3. _groupId = "group1";
4. _clientId = "client1";
5. var cfg = new Dictionary<string, string>()
6. {
7. { "td.connect.port", "6030" },
8. { "auto.offset.reset", "latest" },
9. { "msg.with.table.name", "true" },
10. { "enable.auto.commit", "true" },
11. { "auto.commit.interval.ms", "1000" },
12. { "group.id", _groupId },
13. { "client.id", _clientId },
14. { "td.connect.ip", _host },
15. { "td.connect.user", "root" },
16. { "td.connect.pass", "taosdata" },
17. };
18. IConsumer<Dictionary<string, object>> consumer = null!;
19. try
20. {
21. // create consumer
22. consumer = new ConsumerBuilder<Dictionary<string, object>>(cfg).Build();
23. Console.WriteLine(
24. $"Create consumer successfully, " +
25. $"host: {_host}, " +
26. $"groupId: {_groupId}, " +
27. $"clientId: {_clientId}");
28. }
29. catch (TDengineError e)
30. {
31. // handle TDengine error
32. Console.WriteLine(
33. $"Failed to create native consumer, " +
34. $"host: {_host}, " +
35. $"groupId: {_groupId}, " +
36. $"clientId: {_clientId}, " +
37. $"ErrCode: {e.Code}, " +
38. $"ErrMessage: {e.Error}");
39. throw;
40. }
41. catch (Exception e)
42. {
43. // handle other exceptions
44. Console.WriteLine(
45. $"Failed to create native consumer, " +
46. $"host: {_host}, " +
47. $"groupId: {_groupId}, " +
48. $"clientId: {_clientId}, " +
49. $"ErrMessage: {e.Message}");
50. throw;
51. }

查看源码

1. tmq_t* build_consumer(const ConsumerConfig* config) {
2. tmq_conf_res_t code;
3. tmq_t*         tmq = NULL;

5. // create a configuration object
6. tmq_conf_t* conf = tmq_conf_new();

8. // set the configuration parameters
9. code = tmq_conf_set(conf, "enable.auto.commit", config->enable_auto_commit);
10. if (TMQ_CONF_OK != code) {
11. tmq_conf_destroy(conf);
12. return NULL;
13. }
14. code = tmq_conf_set(conf, "auto.commit.interval.ms", config->auto_commit_interval_ms);
15. if (TMQ_CONF_OK != code) {
16. tmq_conf_destroy(conf);
17. return NULL;
18. }
19. code = tmq_conf_set(conf, "group.id", config->group_id);
20. if (TMQ_CONF_OK != code) {
21. tmq_conf_destroy(conf);
22. return NULL;
23. }
24. code = tmq_conf_set(conf, "client.id", config->client_id);
25. if (TMQ_CONF_OK != code) {
26. tmq_conf_destroy(conf);
27. return NULL;
28. }
29. code = tmq_conf_set(conf, "td.connect.ip", config->td_connect_host);
30. if (TMQ_CONF_OK != code) {
31. tmq_conf_destroy(conf);
32. return NULL;
33. }
34. code = tmq_conf_set(conf, "td.connect.port", config->td_connect_port);
35. if (TMQ_CONF_OK != code) {
36. tmq_conf_destroy(conf);
37. return NULL;
38. }
39. code = tmq_conf_set(conf, "td.connect.user", config->td_connect_user);
40. if (TMQ_CONF_OK != code) {
41. tmq_conf_destroy(conf);
42. return NULL;
43. }
44. code = tmq_conf_set(conf, "td.connect.pass", config->td_connect_pass);
45. if (TMQ_CONF_OK != code) {
46. tmq_conf_destroy(conf);
47. return NULL;
48. }
49. code = tmq_conf_set(conf, "auto.offset.reset", config->auto_offset_reset);
50. if (TMQ_CONF_OK != code) {
51. tmq_conf_destroy(conf);
52. return NULL;
53. }

55. // set the callback function for auto commit
56. tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
57. // create a consumer object
58. tmq = tmq_consumer_new(conf, NULL, 0);

60. _end:
61. // destroy the configuration object
62. tmq_conf_destroy(conf);
63. return tmq;
64. }

查看源码

1. tmq_t* tmq = build_consumer(&config);
2. if (NULL == tmq) {
3. fprintf(stderr, "Failed to create native consumer, host: %s, groupId: %s, , clientId: %s.\n",
4. config.td_connect_host, config.group_id, config.client_id);
5. return -1;
6. } else {
7. fprintf(stdout, "Create consumer successfully, host: %s, groupId: %s, clientId: %s.\n",
8. config.td_connect_host, config.group_id, config.client_id);
9. }

查看源码

调用 build_consumer 函数尝试获取消费者实例 tmq。成功则打印成功日志，失败则打印失败日志。

不支持

订阅消费数据

消费者订阅主题后，可以开始接收并处理这些主题中的消息。订阅消费数据的示例代码如下：

WebSocket 连接

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C
• REST API

1. List<String> topics = Collections.singletonList("topic_meters");
2. try {
3. // subscribe to the topics
4. consumer.subscribe(topics);
5. System.out.println("Subscribe topics successfully.");
6. for (int i = 0; i < 50; i++) {
7. // poll data
8. ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
9. for (ConsumerRecord<ResultBean> record : records) {
10. ResultBean bean = record.value();
11. // Add your data processing logic here
12. System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
13. }
14. }
15. } catch (Exception ex) {
16. // please refer to the JDBC specifications for detailed exceptions info
17. System.out.printf("Failed to poll data, topic: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
18. topics.get(0),
19. groupId,
20. clientId,
21. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
22. ex.getMessage());
23. // Print stack trace for context in examples. Use logging in production.
24. ex.printStackTrace();
25. throw ex;
26. }

查看源码

• subscribe 方法的参数含义为：订阅的主题列表（即名称），支持同时订阅多个主题。
• poll 每次调用获取一个消息，一个消息中可能包含多个记录。
• ResultBean 是我们自定义的一个内部类，其字段名和数据类型与列的名称和数据类型一一对应，这样根据 value.deserializer 属性对应的反序列化类可以反序列化出 ResultBean 类型的对象。

1. def subscribe(consumer):
2. try:
3. consumer.subscribe([topic])
4. print("Subscribe topics successfully")
5. for i in range(50):
6. records = consumer.poll(timeout=1.0)
7. if records:
8. for block in records:
9. for row in block:
10. print(f"data: {row}")

12. except Exception as err:
13. print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
14. raise err

查看源码

• subscribe 方法的参数含义为：订阅的主题列表（即名称），支持同时订阅多个主题。
• poll 每次调用获取一个消息，一个消息中可能包含多个记录。
• records 包含了多个 block 块， 每个块中可能包含多个记录。

1. topic = "topic_meters"
2. err = consumer.Subscribe(topic, nil)
3. if err != nil {
4. log.Fatalf(
5. "Failed to subscribe topic_meters, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
6. topic,
7. groupID,
8. clientID,
9. err.Error(),
10. )
11. }
12. log.Println("Subscribe topics successfully")
13. for i := 0; i < 50; i++ {
14. ev := consumer.Poll(100)
15. if ev != nil {
16. switch e := ev.(type) {
17. case *tmqcommon.DataMessage:
18. // process your data here
19. fmt.Printf("data:%v\n", e)
20. // commit offset
21. _, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition})
22. if err != nil {
23. log.Fatalf(
24. "Failed to commit offset, topic: %s, groupId: %s, clientId: %s, offset %s, ErrMessage: %s\n",
25. topic,
26. groupID,
27. clientID,
28. e.TopicPartition,
29. err.Error(),
30. )
31. }
32. log.Println("Commit offset manually successfully.")
33. case tmqcommon.Error:
34. log.Fatalf(
35. "Failed to poll data, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
36. topic,
37. groupID,
38. clientID,
39. e.Error(),
40. )
41. }
42. }
43. }

查看源码

1. let topic = "topic_meters";
2. match consumer.subscribe([topic]).await {
3. Ok(_) => println!("Subscribe topics successfully."),
4. Err(err) => {
5. eprintln!(
6. "Failed to subscribe topic: {}, groupId: {}, clientId: {}, ErrMessage: {:?}",
7. topic, group_id, client_id, err
8. );
9. return Err(err.into());
10. }
11. }

13. #[derive(Debug, serde::Deserialize)]
14. #[allow(dead_code)]
15. struct Record {
16. // deserialize timestamp to chrono::DateTime<Local>
17. ts: DateTime<Local>,
18. // float to f32
19. current: Option<f32>,
20. // int to i32
21. voltage: Option<i32>,
22. phase: Option<f32>,
23. groupid: i32,
24. // binary/varchar to String
25. location: String,
26. }

28. consumer
29. .stream_with_timeout(Timeout::from_secs(10))
30. .try_for_each(|(offset, message)| async move {
31. let topic = offset.topic();
32. // the vgroup id, like partition id in kafka.
33. let vgroup_id = offset.vgroup_id();
34. println!("* in vgroup id {vgroup_id} of topic {topic}\n");

36. if let Some(data) = message.into_data() {
37. while let Some(block) = data.fetch_raw_block().await? {
38. let records: Vec<Record> = block.deserialize().try_collect()?;
39. // Add your data processing logic here
40. println!("** read {} records: {:#?}\n", records.len(), records);
41. }
42. }
43. Ok(())
44. })
45. .await
46. .map_err(|e| {
47. eprintln!(
48. "Failed to poll data, topic: {}, groupId: {}, clientId: {}, ErrMessage: {:?}",
49. topic, group_id, client_id, e
50. );
51. e
52. })?;

查看源码

• 消费者可订阅一个或多个 TOPIC，一般建议一个消费者只订阅一个 TOPIC。
• TMQ 消息队列是一个 futures::Stream 类型，可以使用相应 API 对每个消息进行消费，并通过 .commit 进行已消费标记。
• Record 是我们自定义的一个结构体，其字段名和数据类型与列的名称和数据类型一一对应，这样可以通过 serde 反序列化出 Record 类型的对象。

1. const { sleep } = require("@tdengine/websocket");
2. const taos = require("@tdengine/websocket");

4. const db = 'power';
5. const stable = 'meters';
6. const url = 'ws://localhost:6041';
7. const topic = 'topic_meters'
8. const topics = [topic];
9. const groupId = "group1";
10. const clientId = "client1";

12. async function createConsumer() {
13. let groupId = "group1";
14. let clientId = "client1";
15. let configMap = new Map([
16. [taos.TMQConstants.GROUP_ID, groupId],
17. [taos.TMQConstants.CLIENT_ID, clientId],
18. [taos.TMQConstants.CONNECT_USER, "root"],
19. [taos.TMQConstants.CONNECT_PASS, "taosdata"],
20. [taos.TMQConstants.AUTO_OFFSET_RESET, "latest"],
21. [taos.TMQConstants.WS_URL, url],
22. [taos.TMQConstants.ENABLE_AUTO_COMMIT, 'true'],
23. [taos.TMQConstants.AUTO_COMMIT_INTERVAL_MS, '1000']
24. ]);
25. try {
26. conn = await taos.tmqConnect(configMap);
27. console.log(`Create consumer successfully, host: ${url}, groupId: ${groupId}, clientId: ${clientId}`)
28. return conn;
29. } catch (err) {
30. console.error(`Failed to create websocket consumer, topic: ${topic}, groupId: ${groupId}, clientId: ${clientId}, ErrCode: ${err.code}, ErrMessage: ${err.message}`);
31. throw err;
32. }

34. }

36. async function prepare() {
37. let conf = new taos.WSConfig('ws://192.168.1.98:6041');
38. conf.setUser('root');
39. conf.setPwd('taosdata');
40. conf.setDb('power');
41. const createDB = `CREATE DATABASE IF NOT EXISTS ${db}`;
42. const createStable = `CREATE STABLE IF NOT EXISTS ${db}.${stable} (ts timestamp, current float, voltage int, phase float) TAGS (location binary(64), groupId int);`;

44. let wsSql = await taos.sqlConnect(conf);
45. await wsSql.exec(createDB);
46. await wsSql.exec(createStable);

48. let createTopic = `CREATE TOPIC IF NOT EXISTS ${topics[0]} AS SELECT * FROM ${db}.${stable}`;
49. await wsSql.exec(createTopic);
50. await wsSql.close();
51. }

53. async function insert() {
54. let conf = new taos.WSConfig('ws://localhost:6041');
55. conf.setUser('root');
56. conf.setPwd('taosdata');
57. conf.setDb('power');
58. let wsSql = await taos.sqlConnect(conf);
59. for (let i = 0; i < 1; i++) {
60. await wsSql.exec(`INSERT INTO d1001 USING ${stable} (location, groupId) TAGS ("California.SanFrancisco", 3) VALUES (NOW, ${10 + i}, ${200 + i}, ${0.32 + i})`);
61. }
62. await wsSql.close();
63. }

65. async function subscribe(consumer) {
66. try {
67. await consumer.subscribe(['topic_meters']);
68. let res = new Map();
69. while (res.size == 0) {
70. res = await consumer.poll(100);
71. await consumer.commit();
72. }
73. let assignment = await consumer.assignment();
74. await consumer.seekToBeginning(assignment);
75. console.log("Assignment seek to beginning successfully");
76. } catch (err) {
77. console.error(`Failed to seek offset, topic: ${topic}, groupId: ${groupId}, clientId: ${clientId}, ErrCode: ${err.code}, ErrMessage: ${err.message}`);
78. throw err;
79. }
80. }

82. async function test() {
83. let consumer = null;
84. try {
85. await prepare();
86. consumer = await createConsumer();
87. const allPromises = [];
88. allPromises.push(subscribe(consumer));
89. allPromises.push(insert());
90. await Promise.all(allPromises);
91. await consumer.unsubscribe();
92. console.log("Consumer unsubscribed successfully.");
93. }
94. catch (err) {
95. console.error(`Failed to consumer, topic: ${topic}, groupId: ${groupId}, clientId: ${clientId}, ErrCode: ${err.code}, ErrMessage: ${err.message}`);
96. throw err;
97. }
98. finally {
99. if (consumer) {
100. await consumer.close();
101. console.log("Consumer closed successfully.");
102. }
103. taos.destroy();
104. }
105. }

107. test()

查看源码

1. _topic = "topic_meters";
2. try
3. {
4. // subscribe
5. consumer.Subscribe(new List<string>() { _topic });
6. Console.WriteLine("Subscribe topics successfully");
7. for (int i = 0; i < 50; i++)
8. {
9. // consume message with using block to ensure the result is disposed
10. using (var cr = consumer.Consume(100))
11. {
12. if (cr == null) continue;
13. foreach (var message in cr.Message)
14. {
15. // handle message
16. Console.WriteLine(
17. $"data: {{{((DateTime)message.Value["ts"]).ToString("yyyy-MM-dd HH:mm:ss.fff")}, " +
18. $"{message.Value["current"]}, {message.Value["voltage"]}, {message.Value["phase"]}}}");
19. }
20. }
21. }
22. }
23. catch (TDengineError e)
24. {
25. // handle TDengine error
26. Console.WriteLine(
27. $"Failed to poll data, " +
28. $"topic: {_topic}, " +
29. $"groupId: {_groupId}, " +
30. $"clientId: {_clientId}, " +
31. $"ErrCode: {e.Code}, " +
32. $"ErrMessage: {e.Error}");
33. throw;
34. }
35. catch (Exception e)
36. {
37. // handle other exceptions
38. Console.WriteLine($"Failed to poll data, " +
39. $"topic: {_topic}, " +
40. $"groupId: {_groupId}, " +
41. $"clientId: {_clientId}, " +
42. $"ErrMessage: {e.Message}");
43. throw;
44. }

查看源码

1. // build a topic list used to subscribe
2. ws_tmq_list_t* build_topic_list() {
3. // create a empty topic list
4. ws_tmq_list_t* topicList = ws_tmq_list_new();

6. // append topic name to the list
7. int32_t code = ws_tmq_list_append(topicList, topic_name);
8. if (code) {
9. // if failed, destroy the list and return NULL
10. ws_tmq_list_destroy(topicList);
11. fprintf(stderr,
12. "Failed to create topic_list, topic: %s, groupId: %s, clientId: %s, ErrCode: 0x%x, ErrMessage: %s.\n",
13. topic_name, config.group_id, config.client_id, code, ws_tmq_errstr(NULL));
14. return NULL;
15. }
16. // if success, return the list
17. return topicList;
18. }

查看源码

1. void basic_consume_loop(ws_tmq_t* tmq) {
2. int32_t totalRows = 0;   // total rows consumed
3. int32_t msgCnt = 0;      // total messages consumed
4. int32_t timeout = 5000;  // poll timeout

6. while (running) {
7. // poll message from TDengine
8. WS_RES* tmqmsg = ws_tmq_consumer_poll(tmq, timeout);
9. if (tmqmsg) {
10. msgCnt++;

12. // Add your data processing logic here
13. totalRows += msg_process(tmqmsg);

15. // free the message
16. ws_free_result(tmqmsg);
17. }
18. if (msgCnt > 50) {
19. // consume 50 messages and break
20. break;
21. }
22. }

24. // print the result: total messages and total rows consumed
25. fprintf(stdout, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
26. }

查看源码

1. int32_t msg_process(WS_RES* msg) {
2. int32_t     rows = 0;
3. const char* topicName = ws_tmq_get_topic_name(msg);
4. const char* dbName = ws_tmq_get_db_name(msg);
5. int32_t     vgroupId = ws_tmq_get_vgroup_id(msg);

7. while (true) {
8. // get one row data from message
9. WS_ROW row = ws_fetch_row(msg);
10. if (row == NULL) break;

12. // Add your data processing logic here

14. rows++;
15. }

17. return rows;
18. }

查看源码

1. ws_tmq_list_t* topic_list = build_topic_list();
2. if (NULL == topic_list) {
3. fprintf(stderr, "Failed to create topic_list, topic: %s, groupId: %s, clientId: %s.\n", topic_name, config.group_id,
4. config.client_id);
5. return -1;
6. }

8. if ((code = ws_tmq_subscribe(tmq, topic_list))) {
9. fprintf(stderr,
10. "Failed to subscribe topic_list, topic: %s, groupId: %s, clientId: %s, ErrCode: 0x%x, ErrMessage: %s.\n",
11. topic_name, config.group_id, config.client_id, code, ws_tmq_errstr(tmq));
12. } else {
13. fprintf(stdout, "Subscribe topics successfully.\n");
14. }

16. ws_tmq_list_destroy(topic_list);

18. basic_consume_loop(tmq);

查看源码

订阅消费数据步骤：

1. 调用 ws_build_topic_list 函数创建一个主题列表 topic_list。
2. 如果 topic_list 为 NULL，表示创建失败，函数返回 -1。
3. 使用 ws_tmq_subscribe 函数订阅 tmq 指定的主题列表。如果订阅失败，打印错误信息。
4. 销毁主题列表 topic_list 以释放资源。
5. 调用 basic_consume_loop 函数开始基本的消费循环，处理订阅的消息。

不支持

原生连接

• Java
• Python
• Go
• Rust
• Node.js
• C#
• C
• REST API

1. List<String> topics = Collections.singletonList("topic_meters");
2. try {
3. // subscribe to the topics
4. consumer.subscribe(topics);
5. System.out.println("Subscribe topics successfully.");
6. for (int i = 0; i < 50; i++) {
7. // poll data
8. ConsumerRecords<ResultBean> records = consumer.poll(Duration.ofMillis(100));
9. for (ConsumerRecord<ResultBean> record : records) {
10. ResultBean bean = record.value();
11. // Add your data processing logic here
12. System.out.println("data: " + JsonUtil.getObjectMapper().writeValueAsString(bean));
13. }
14. }
15. } catch (Exception ex) {
16. // please refer to the JDBC specifications for detailed exceptions info
17. System.out.printf("Failed to poll data, topic: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
18. topics.get(0),
19. groupId,
20. clientId,
21. ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
22. ex.getMessage());
23. // Print stack trace for context in examples. Use logging in production.
24. ex.printStackTrace();
25. throw ex;
26. }

查看源码

• subscribe 方法的参数含义为：订阅的主题列表（即名称），支持同时订阅多个主题。
• poll 每次调用获取一个消息，一个消息中可能包含多个记录。
• ResultBean 是我们自定义的一个内部类，其字段名和数据类型与列的名称和数据类型一一对应，这样根据 value.deserializer 属性对应的反序列化类可以反序列化出 ResultBean 类型的对象。

1. def subscribe(consumer):
2. try:
3. # subscribe to the topics
4. consumer.subscribe(["topic_meters"])
5. print("Subscribe topics successfully")
6. for i in range(50):
7. records = consumer.poll(1)
8. if records:
9. err = records.error()
10. if err is not None:
11. print(f"Poll data error, {err}")
12. raise err

14. val = records.value()
15. if val:
16. for block in val:
17. data = block.fetchall()
18. print(f"data: {data}")

20. except Exception as err:
21. print(f"Failed to poll data, topic: {topic}, groupId: {groupId}, clientId: {clientId}, ErrMessage:{err}.")
22. raise err

查看源码

• subscribe 方法的参数含义为：订阅的主题列表（即名称），支持同时订阅多个主题。
• poll 每次调用获取一个消息，一个消息中可能包含多个记录。
• records 包含了多个 block 块， 每个块中可能包含多个记录。

1. topic = "topic_meters"
2. err = consumer.Subscribe(topic, nil)
3. if err != nil {
4. log.Fatalf(
5. "Failed to subscribe topic_meters, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n",
6. topic,
7. groupID,
8. clientID,
9. err.Error(),
10. )
11. }
12. log.Println("Subscribe topics successfully")
13. for i := 0; i < 50; i++ {
14. ev := consumer.Poll(100)
15. if ev != nil {
16. switch e := ev.(type) {
17. case *tmqcommon.DataMessage:
18. // process your data here
19. fmt.Printf("data:%v\n", e)
20. // commit offset
21. _, err = consumer.CommitOffsets([]tmqcommon.TopicPartition{e.TopicPartition})
22. if err != nil {
23. log.Fatalf(
24. "Failed to commit offset, topic: %s, groupId: %s, clientId: %s, offset %s, ErrMessage: %s\n",
25. topic,
26. groupID,
27. clientID,
28. e.TopicPartition,
29. err.Error(),
30. )
31. }
32. log.Println("Commit offset manually successfully.")
33. case tmqcommon.Error:
34. log.Fatalf("Failed to poll data, topic: %s, groupId: %s, clientId: %s, ErrMessage: %s\n", topic, groupID, clientID, e.Error())
35. }
36. }
37. }

查看源码

1. let topic = "topic_meters";
2. match consumer.subscribe([topic]).await {
3. Ok(_) => println!("Subscribe topics successfully."),
4. Err(err) => {
5. eprintln!(
6. "Failed to subscribe topic: {}, groupId: {}, clientId: {}, ErrMessage: {:?}",
7. topic, group_id, client_id, err
8. );
9. return Err(err.into());
10. }
11. }

13. #[derive(Debug, serde::Deserialize)]
14. #[allow(dead_code)]
15. struct Record {
16. // deserialize timestamp to chrono::DateTime<Local>
17. ts: DateTime<Local>,
18. // float to f32
19. current: Option<f32>,
20. // int to i32
21. voltage: Option<i32>,
22. phase: Option<f32>,
23. groupid: i32,
24. // binary/varchar to String
25. location: String,
26. }

28. consumer
29. .stream_with_timeout(Timeout::from_secs(10))
30. .try_for_each(|(offset, message)| async move {
31. let topic = offset.topic();
32. // the vgroup id, like partition id in kafka.
33. let vgroup_id = offset.vgroup_id();
34. println!("* in vgroup id {vgroup_id} of topic {topic}\n");

36. if let Some(data) = message.into_data() {
37. while let Some(block) = data.fetch_raw_block().await? {
38. let records: Vec<Record> = block.deserialize().try_collect()?;
39. // Add your data processing logic here
40. println!("** read {} records: {:#?}\n", records.len(), records);
41. }
42. }
43. Ok(())
44. })
45. .await
46. .map_err(|e| {
47. eprintln!(
48. "Failed to poll data, topic: {}, groupId: {}, clientId: {}, ErrMessage: {:?}",
49. topic, group_id, client_id, e
50. );
51. e
52. })?;

查看源码

• 消费者可订阅一个或多个 TOPIC，一般建议一个消费者只订阅一个 TOPIC。
• TMQ 消息队列是一个 futures::Stream 类型，可以使用相应 API 对每个消息进行消费，并通过 .commit 进行已消费标记。
• Record 是我们自定义的一个结构体，其字段名和数据类型与列的名称和数据类型一一对应，这样可以通过 serde 反序列化出 Record 类型的对象。

不支持

1. _topic = "topic_meters";
2. try
3. {
4. // subscribe
5. consumer.Subscribe(new List<string>() { _topic });
6. Console.WriteLine("Subscribe topics successfully");
7. for (int i = 0; i < 50; i++)
8. {
9. // consume message with using block to ensure the result is disposed
10. using (var cr = consumer.Consume(100))
11. {
12. if (cr == null) continue;
13. foreach (var message in cr.Message)
14. {
15. // handle message
16. Console.WriteLine(
17. $"data: {{{((DateTime)message.Value["ts"]).ToString("yyyy-MM-dd HH:mm:ss.fff")}, " +
18. $"{message.Value["current"]}, {message.Value["voltage"]}, {message.Value["phase"]}}}");
19. }
20. }
21. }
22. }
23. catch (TDengineError e)
24. {
25. // handle TDengine error
26. Console.WriteLine(
27. $"Failed to poll data, " +
28. $"topic: {_topic}, " +
29. $"groupId: {_groupId}, " +
30. $"clientId: {_clientId}, " +
31. $"ErrCode: {e.Code}, " +
32. $"ErrMessage: {e.Error}");
33. throw;
34. }
35. catch (Exception e)
36. {
37. // handle other exceptions
38. Console.WriteLine(
39. $"Failed to poll data, " +
40. $"topic: {_topic}, " +
41. $"groupId: {_groupId}, " +
42. $"clientId: {_clientId}, " +
43. $"ErrMessage: {e.Message}");
44. throw;
45. }

查看源码

1. // build a topic list used to subscribe
2. tmq_list_t* build_topic_list() {
3. // create a empty topic list
4. tmq_list_t* topicList = tmq_list_new();

6. // append topic name to the list
7. int32_t code = tmq_list_append(topicList, topic_name);
8. if (code) {
9. // if failed, destroy the list and return NULL
10. tmq_list_destroy(topicList);
11. fprintf(stderr, "Failed to create topic_list, topic: %s, groupId: %s, clientId: %s, ErrCode: 0x%x, ErrMessage: %s.\n",
12. topic_name, config.group_id, config.client_id, code, tmq_err2str(code));
13. return NULL;
14. }
15. // if success, return the list
16. return topicList;
17. }

查看源码

1. void basic_consume_loop(tmq_t* tmq) {
2. int32_t totalRows = 0;   // total rows consumed
3. int32_t msgCnt = 0;      // total messages consumed
4. int32_t timeout = 5000;  // poll timeout

6. while (running) {
7. // poll message from TDengine
8. TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout);
9. if (tmqmsg) {
10. msgCnt++;

12. // Add your data processing logic here
13. totalRows += msg_process(tmqmsg);

15. // free the message
16. taos_free_result(tmqmsg);
17. }
18. if (msgCnt > 50) {
19. // consume 50 messages and break
20. break;
21. }
22. }

24. // print the result: total messages and total rows consumed
25. fprintf(stdout, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
26. }

查看源码

1. int32_t msg_process(TAOS_RES* msg) {
2. int32_t     rows = 0;
3. const char* topicName = tmq_get_topic_name(msg);
4. const char* dbName    = tmq_get_db_name(msg);
5. int32_t     vgroupId  = tmq_get_vgroup_id(msg);

7. while (true) {
8. // get one row data from message
9. TAOS_ROW row = taos_fetch_row(msg);
10. if (row == NULL) break;

12. // Add your data processing logic here

14. rows++;
15. }

17. return rows;
18. }

查看源码

``` 1. tmq_list_t* topic_list = build_topic_list(); 2. if (NULL == topic_list) { 3. fprintf(stderr, "Failed to create topic_list, topic: %s, groupId: %s, clientId: %s.\n", 4. topic_name, config.group_id, config.client_id); 5. return -1; 6. }

1. if ((code = tmq_subscribe(tmq, topic_list))) {
2. fprintf(stderr, "Failed to subscribe topic_list, topic: %s, groupId: %s, clientId: %s, ErrCode: 0x%x, ErrMessage: %

（本文超出站点单文档体积上限，已截断）