数据订阅API

IoTDB 提供了强大的数据订阅功能，允许用户通过订阅 API 实时获取 IoTDB 新增的数据。详细的功能定义及介绍：数据订阅

1 核心步骤

1. 创建Topic：创建一个Topic，Topic中包含希望订阅的测点。
2. 订阅Topic：在 consumer 订阅 topic 前，topic 必须已经被创建，否则订阅会失败。同一个 consumer group 下的 consumers 会均分数据。
3. 消费数据：只有显式订阅了某个 topic，才会收到对应 topic 的数据。
4. 取消订阅： consumer close 时会退出对应的 consumer group，同时取消现存的所有订阅。

2 详细步骤

本章节用于说明开发的核心流程，并未演示所有的参数和接口，如需了解全部功能及参数请参见: 全量接口说明

2.1 创建maven项目

创建一个maven项目，并导入以下依赖（JDK >= 1.8, Maven >= 3.6）

1. <dependencies>
2. <dependency>
3. <groupId>org.apache.iotdb</groupId>
4. <artifactId>iotdb-session</artifactId>
5. <!-- 版本号与数据库版本号相同 -->
6. <version>${project.version}</version>
7. </dependency>
8. </dependencies>

2.2 代码案例

2.2.1 Topic操作

1. import java.util.Optional;
2. import java.util.Properties;
3. import java.util.Set;
4. import org.apache.iotdb.rpc.IoTDBConnectionException;
5. import org.apache.iotdb.rpc.StatementExecutionException;
6. import org.apache.iotdb.rpc.subscription.config.TopicConstant;
7. import org.apache.iotdb.session.subscription.SubscriptionSession;
8. import org.apache.iotdb.session.subscription.model.Topic;

10. public class DataConsumerExample {

12. public static void main(String[] args) throws IoTDBConnectionException, StatementExecutionException {
13. try (SubscriptionSession session = new SubscriptionSession("127.0.0.1", 6667, "root", "root", 67108864)) {
14. // 1. open session
15. session.open();

17. // 2. create a topic of all data
18. Properties sessionConfig = new Properties();
19. sessionConfig.put(TopicConstant.PATH_KEY, "root.**");

21. session.createTopic("allData", sessionConfig);

23. // 3. show all topics
24. Set<Topic> topics = session.getTopics();
25. System.out.println(topics);

27. // 4. show a specific topic
28. Optional<Topic> allData = session.getTopic("allData");
29. System.out.println(allData.get());
30. }
31. }
32. }

2.2.2 数据消费

场景-1: 订阅IoTDB中新增的实时数据（大屏或组态展示的场景）

1. import java.io.IOException;
2. import java.util.List;
3. import java.util.Properties;
4. import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
5. import org.apache.iotdb.rpc.subscription.config.TopicConstant;
6. import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer;
7. import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
8. import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
9. import org.apache.iotdb.session.subscription.payload.SubscriptionSessionDataSet;
10. import org.apache.tsfile.read.common.RowRecord;

12. public class DataConsumerExample {

14. public static void main(String[] args) throws IOException {

16. // 5. create a pull consumer, the subscription is automatically cancelled when the logic in the try resources is completed
17. Properties consumerConfig = new Properties();
18. consumerConfig.put(ConsumerConstant.CONSUMER_ID_KEY, "c1");
19. consumerConfig.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "cg1");
20. consumerConfig.put(ConsumerConstant.USERNAME_KEY, "root");
21. consumerConfig.put(ConsumerConstant.PASSWORD_KEY, "root");
22. try (SubscriptionPullConsumer pullConsumer = new SubscriptionPullConsumer(consumerConfig)) {
23. pullConsumer.open();
24. pullConsumer.subscribe("topic_all");
25. while (true) {
26. List<SubscriptionMessage> messages = pullConsumer.poll(10000);
27. for (final SubscriptionMessage message : messages) {
28. final short messageType = message.getMessageType();
29. if (SubscriptionMessageType.isValidatedMessageType(messageType)) {
30. for (final SubscriptionSessionDataSet dataSet : message.getSessionDataSetsHandler()) {
31. while (dataSet.hasNext()) {
32. final RowRecord record = dataSet.next();
33. System.out.println(record);
34. }
35. }
36. }
37. }
38. }
39. }
40. }
41. }

场景-2:订阅新增的 TsFile（定期数据备份的场景）

前提：需要被消费的topic的格式为TsfileHandler类型，举例：create topic topic_all_tsfile with ('path'='root.**','format'='TsFileHandler')

1. import java.io.IOException;
2. import java.util.List;
3. import java.util.Properties;
4. import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
5. import org.apache.iotdb.rpc.subscription.config.TopicConstant;
6. import org.apache.iotdb.session.subscription.consumer.SubscriptionPullConsumer;
7. import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;

10. public class DataConsumerExample {

12. public static void main(String[] args) throws IOException {
13. // 1. create a pull consumer, the subscription is automatically cancelled when the logic in the try resources is completed
14. Properties consumerConfig = new Properties();
15. consumerConfig.put(ConsumerConstant.CONSUMER_ID_KEY, "c1");
16. consumerConfig.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "cg1");
17. consumerConfig.put(ConsumerConstant.USERNAME_KEY, "root");
18. consumerConfig.put(ConsumerConstant.PASSWORD_KEY, "root");
19. consumerConfig.put(ConsumerConstant.FILE_SAVE_DIR_KEY, "/Users/iotdb/Downloads");
20. try (SubscriptionPullConsumer pullConsumer = new SubscriptionPullConsumer(consumerConfig)) {
21. pullConsumer.open();
22. pullConsumer.subscribe("topic_all_tsfile");
23. while (true) {
24. List<SubscriptionMessage> messages = pullConsumer.poll(10000);
25. for (final SubscriptionMessage message : messages) {
26. message.getTsFileHandler().copyFile("/Users/iotdb/Downloads/1.tsfile");
27. }
28. }
29. }
30. }
31. }

2 全量接口说明

2.1 参数列表

可通过Properties参数对象设置消费者相关参数，具体参数如下。

2.1.1 SubscriptionConsumer

SubscriptionPushConsumer 中的特殊配置：

SubscriptionPullConsumer 中的特殊配置：

函数列表

数据订阅

SubscriptionPullConsumer

SubscriptionPushConsumer