Data subscription API

IoTDB provides powerful data subscription functionality, allowing users to access newly added data from IoTDB in real-time through subscription APIs. For detailed functional definitions and introductions：Data subscription

1 Core Steps

1. Create Topic: Create a Topic that includes the measurement points you wish to subscribe to.
2. Subscribe to Topic: Before a consumer subscribes to a topic, the topic must have been created, otherwise the subscription will fail. Consumers under the same consumer group will evenly distribute the data.
3. Consume Data: Only by explicitly subscribing to a specific topic will you receive data from that topic.
4. Unsubscribe: When a consumer is closed, it will exit the corresponding consumer group and cancel all existing subscriptions.

2 Detailed Steps

This section is used to illustrate the core development process and does not demonstrate all parameters and interfaces. For a comprehensive understanding of all features and parameters, please refer to: Java Native API

2.1 Create a Maven project

Create a Maven project and import the following dependencies（JDK >= 1.8, Maven >= 3.6）

1. <dependencies>
2. <dependency>
3. <groupId>org.apache.iotdb</groupId>
4. <artifactId>iotdb-session</artifactId>
5. <!-- The version number is the same as the database version number -->
6. <version>${project.version}</version>
7. </dependency>
8. </dependencies>

2.2 Code Example

2.2.1 Topic operations

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 Data Consume

Scenario-1: Subscribing to newly added real-time data in IoTDB (for scenarios such as dashboard or configuration display)

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. }

Scenario-2: Subscribing to newly added TsFiles (for scenarios such as regular data backup)

Prerequisite: The format of the topic to be consumed must be of the TsfileHandler type. For example：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. }

3 Java Native API Description

3.1 Parameter List

The consumer-related parameters can be set through the Properties parameter object. The specific parameters are as follows:

SubscriptionConsumer

Special configurations in SubscriptionPushConsumer ：

Special configurations in SubscriptionPullConsumer ：

3.2 Function List

Data subscription

SubscriptionPullConsumer

SubscriptionPushConsumer