Manage Consumers

TDengine provides data subscription and consumption interfaces similar to those of message queue products. In many scenarios, using TDengine's time-series big data platform eliminates the need to integrate message queue products, thereby simplifying application design and reducing operational costs. This chapter introduces the relevant API for data subscription for various language connectors and how to use them. For basic knowledge of data subscription, please refer to Data Subscription.

Creating a Topic

You can use taos shell or refer to the Execute SQL chapter to execute the SQL for creating a topic: CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters

The above SQL will create a subscription named topic_meters. Each record in the messages retrieved using this subscription consists of the columns selected by this query statement SELECT ts, current, voltage, phase, groupid, location FROM meters.

note

In the TDengine connector implementation, there are the following limitations for subscription queries:

Query Limitations: Subscription queries can only use select statements and do not support other types of SQL, such as insert, update, or delete.
Raw Data Query: Subscription queries can only query raw data and cannot query aggregated or computed results.
Time Order Limit: Subscription queries can only query data in chronological order.

Creating a Consumer

The concept of a TDengine consumer is similar to that of Kafka. Consumers receive data streams by subscribing to topics. Consumers can configure various parameters, such as connection methods, server addresses, and automatic offset commits, to suit different data processing needs. Some language connectors' consumers also support advanced features such as automatic reconnection and data transmission compression to ensure efficient and stable data reception.

Creating Parameters

The parameters for creating consumers are numerous and flexible, supporting various connection types, offset commit methods, compression, reconnection, deserialization, and other features. The common basic configuration items applicable to all language connectors are shown in the following table:

Parameter Name	Type	Parameter Description	Remarks
`td.connect.ip`	string	Server's IP address
`td.connect.user`	string	Username
`td.connect.pass`	string	Password
`td.connect.port`	integer	Server's port number
`group.id`	string	Consumer group ID, shared consumption progress among the same group	Required. Maximum length: 192. A maximum of 100 consumer groups can be established for each topic.
`client.id`	string	Client ID	Maximum length: 192
`auto.offset.reset`	enum	Initial position for the consumer group subscription	`earliest`: default (version < 3.2.0.0); subscribe from the beginning; `latest`: default (version >= 3.2.0.0); only start from the latest data; `none`: cannot subscribe without a committed offset.
`enable.auto.commit`	boolean	Whether to enable automatic offset submission, true: automatically submit, the client application does not need to commit; false: the client application needs to commit itself	Default value is true
`auto.commit.interval.ms`	integer	Interval for automatically committing consumed offsets, in milliseconds	Default value is 5000
`msg.with.table.name`	boolean	Whether to allow parsing the table name from the message; not applicable for column subscriptions (for column subscriptions, tbname can be written as a column in the subquery) (this parameter is deprecated from version 3.2.0.0 and is always true)	Default is off
`enable.replay`	boolean	Whether to enable data replay functionality	Default is off
`session.timeout.ms`	integer	Timeout period after the consumer's heartbeat is lost; after timeout, the rebalance logic will be triggered, and if successful, the consumer will be deleted (supported from version 3.3.3.0)	Default value is 12000, value range [6000, 1800000]
`max.poll.interval.ms`	integer	Maximum time interval for the consumer to poll and fetch data; exceeding this time will be considered as the consumer being offline, triggering rebalance logic, and if successful, the consumer will be deleted (supported from version 3.3.3.0)	Default value is 300000, [1000, INT32_MAX]

Here are the creation parameters for various language connectors:

The parameters for creating a consumer in the Java connector are Properties. For the list of configurable parameters, please refer to Consumer. Other parameters can be referenced in the common basic configuration items above.

Provides the td.connect.websocket.scheme parameter to indicate the protocol type, with other parameters being the same as the common basic configuration items.

Supported property list for creating a consumer:

ws.url: WebSocket connection address.
ws.message.channelLen: WebSocket message channel cache length, default 0.
ws.message.timeout: WebSocket message timeout, default 5m.
ws.message.writeWait: WebSocket message write timeout, default 10s.
ws.message.enableCompression: Whether to enable WebSocket compression, default false.
ws.autoReconnect: Whether to automatically reconnect WebSocket, default false.
ws.reconnectIntervalMs: WebSocket reconnection interval time in milliseconds, default 2000.
ws.reconnectRetryCount: WebSocket reconnection retry count, default 3.

Other parameters are referenced in the table above.

Provides the WS_URL parameter to indicate the server address to connect to, with other parameters being the same as the common basic configuration items.

Supported property list for creating a consumer:

useSSL: Whether to use SSL for the connection, default false.
token: Token to connect to TDengine cloud.
ws.message.enableCompression: Whether to enable WebSocket compression, default false.
ws.autoReconnect: Whether to automatically reconnect, default false.
ws.reconnect.retry.count: Reconnection attempts, default 3.
ws.reconnect.interval.ms: Reconnection interval in milliseconds, default 2000.

Other parameters are referenced in the table above.

WebSocket Connection: Since DSN is used, there is no need for the configurations td.connect.ip, td.connect.port, td.connect.user, and td.connect.pass, with the rest being the same as the common configuration items.
Native Connection: Same as the common basic configuration items.

Websocket Connection

This section introduces how to create a consumer using WebSocket connection in various language connectors. Specify the server address to connect, set automatic commits, start consuming from the latest messages, and specify group.id and client.id information. Some language connectors also support deserialization parameters.

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

try {
    TaosConsumer<ResultBean> consumer= new TaosConsumer<>(config);
    System.out.printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s%n",
            config.getProperty("bootstrap.servers"),
            config.getProperty("group.id"),
            config.getProperty("client.id"));
    return consumer;
} catch (Exception ex) {
    // please refer to the JDBC specifications for detailed exceptions info
    System.out.printf("Failed to create websocket consumer, host: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
            config.getProperty("bootstrap.servers"),
            config.getProperty("group.id"),
            config.getProperty("client.id"),
            ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
            ex.getMessage());
    // Print stack trace for context in examples. Use logging in production.
    ex.printStackTrace();
    throw ex;
}

Creating a Topic​

Creating a Consumer​

Creating Parameters​

Websocket Connection​

Native Connection​

Subscribing to Consume Data​

Websocket Connection​

Native Connection​

Specifying the Subscription Offset​

Websocket Connection​

Native Connection​

Committing Offset​

Websocket Connection​

Native Connection​

Unsubscribing and Closing Consumption​

Websocket Connection​

Native Connection​

Complete Example​

Websocket Connection​

Native Connection​

Support and Feedback

Creating a Topic

Creating a Consumer

Creating Parameters

Websocket Connection

Native Connection

Subscribing to Consume Data

Websocket Connection

Native Connection

Specifying the Subscription Offset

Websocket Connection

Native Connection

Committing Offset

Websocket Connection

Native Connection

Unsubscribing and Closing Consumption

Websocket Connection

Native Connection

Complete Example

Websocket Connection

Native Connection