tencent cloud

Stream Compute Service

DokumentasiStream Compute ServiceSQL Developer GuideConnectorsTencent Cloud Message Queue

Tencent Cloud Message Queue

Unduh
Mode fokus
Ukuran font
Terakhir diperbarui: 2026-09-03 10:18:07
Diterjemahkan oleh AI

Introduction

Message Queue CMQ (Cloud Message Queue, hereinafter referred to as CMQ) is a distributed message queue system based on Tencent's self-developed messaging engine. It can serve as a data source (Source) and a data destination (Sink). Users can import streaming data into a specific Queue in CMQ. After the data is processed through Flink operators, they can output the results to another Queue in the same or a different CMQ instance.

Version Description

Flink Version
Description
1.13
Supported
1.14
Not supported.
1.16
Not supported.

Application Scope

CMQ can be used as a source table (Source) and as a destination table (Sink) for Tuple data streams. However, it does not currently support Upsert data streams.

DDL Definition

As a source

JSON format

CREATE TABLE `cmq_source_json_table` (
    `id` INT,
    `name`STRING,
    PRIMARY KEY (`id`) NOT ENFORCED	-- To deduplicate data, you must specify a primary key (PK) to distinguish different data records based on the key.
) WITH (
    'connector' = 'cmq',			   -- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',	-- The nameServer for the region where cmq is located
    'queue' = 'queue_name',			-- The queue name for cmq
    'secret-id' = 'xxxx',			  -- The account's secretId
    'secret-key' = 'xxxx',             -- The account's secretKey
    'sign-method' = 'HmacSHA1',        -- The signature method
    'format' = 'json',                 -- Define the data format (JSON).
    'json.fail-on-missing-field' = 'false',	 -- If this is set to 'false', no error will be reported when missing fields are encountered.
    'json.ignore-parse-errors' = 'true',   	 -- If this is set to 'true', any parse errors will be ignored.
    'batch-size' = '16',               -- The number of messages to consume in a batch
    'request-timeout' = '5000ms',  	-- The timeout duration for requests
    'polling-wait-timeout'= '10s', 	-- The wait time when no data can be obtained
    'key-alive-timeout'= '5min'    	-- The validity period for CMQ deduplication of messages containing a primary key
);

CSV format

CREATE TABLE `cmq_source_csv_table` (
    `id` int,
    `name` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED -- To deduplicate data, you must specify a primary key (PK) to distinguish different data records based on the key.
) WITH (
    'connector' = 'cmq', 			 -- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',	-- The nameServer for the region where cmq is located
    'queue' = 'queue_name',	 	-- The queue name for cmq
    'secret-id' = 'xxxx',		   -- The account's secretId
    'secret-key' = 'xxxx',          -- The account's secretKey
    'sign-method' = 'HmacSHA1',     -- The signature method
    'format' = 'csv',               -- Define the data format (CSV).
    'batch-size' = '16',            -- The number of messages to consume/send in a batch
    'request-timeout' = '5000ms',   -- The timeout duration for requests
    'polling-wait-timeout'= '10s',  -- The wait time when data cannot be obtained
    'key-alive-timeout'= '5min'     -- The validity period for CMQ deduplication of messages containing a primary key
);

As a data sink

JSON format

CREATE TABLE `cmq_sink_json_table` (
    `id` int,
    `name` STRING
) WITH (
    'connector' = 'cmq', 					 	-- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',    -- The nameServer for the region where cmq is located
    'queue' = 'queue_name',			 		-- The queue name for cmq
    'secret-id' = 'xxxx',				   	-- The account's secretId
    'secret-key' = 'xxxx',         		 	-- The account's secretKey
    'sign-method' = 'HmacSHA1',     			-- The signature method
    'format' = 'json',             		 	-- Defines the data format (JSON).
    'json.fail-on-missing-field' = 'false',	 -- If this is set to 'false', no error will be reported when missing fields are encountered.
    'json.ignore-parse-errors' = 'true',         -- If this is set to 'true', any parse errors will be ignored.
    'batch-size' = '16',                        -- The number of messages to send in a batch
    'request-timeout' = '5000ms',               -- The timeout duration for requests
    'retry-times' = '3',                        -- The number of retries for sending messages
    'max-block-timeout' = '0s'                  -- The maximum wait time for sending data in batches
);

CSV format

CREATE TABLE `cmq_sink_csv_table` (
    `id` int,
    `name` STRING
) WITH (
    'connector' = 'cmq', 					     -- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',    -- The nameServer for the region where cmq is located
    'queue' = 'queue_name',				     -- The queue name for cmq
    'secret-id' = 'xxxx',			           -- The account's secretId
    'secret-key' = 'xxxx',                      -- The account's secretKey
    'sign-method' = 'HmacSHA1',                 -- The signature method
    'format' = 'csv',                           -- Define the data format (CSV).
    'batch-size' = '16',                    	-- The number of messages to send in a batch
    'request-timeout' = '5000ms',               -- The timeout duration for requests
    'retry-times' = '3',                        -- The number of retries for sending messages
    'max-block-timeout' = '0s'                  -- The maximum wait time for sending data in batches
);

WITH Parameters

WITH Parameters

Parameter Value
Required
Default Value
Description
connector
Yes
N/A
Must be set to 'cmq'.
hosts
Yes
N/A
The nameServer for the queue's region. For the specific address, refer to the TCP SDK.
queue
Yes
N/A
The queue name corresponding to cmq.
secret-id
Yes
N/A
The account's secretId.
secret-key
Yes
N/A
The account's secretKey.
sign-method
No
HmacSHA1
The account's signature method.
format
Yes
N/A
The input/output format for CMQ messages. Currently supported formats include 'csv' and 'json'.
batch-size
No
16
The number of messages to send/receive in a batch.
request-timeout
No
5000ms
The request timeout duration.
polling-wait-timeout
No
10s
The wait time when no data can be obtained.
key-alive-timeout
No
60s
The validity period for CMQ deduplication of messages containing a primary key. This parameter is designed to prevent the same data from being consumed repeatedly, but it does not guarantee global uniqueness.
retry-times
No
3
The number of retries for sending messages.
max-block-timeout
No
0s
The maximum wait time for sending data in batches. A value of '0s' indicates no waiting, and data is sent immediately when available.

WITH parameters for JSON

Parameter Value
Required
Default Value
Description
json.fail-on-missing-field
No
false
If this is true, the job will fail in case of missing parameters. If this is false (default), the missing parameters will be set to null and the job will continue to be executed.
json.ignore-parse-errors
No
false
If this is true, when there is a parse error, the field will be set to null and the job will continue to be executed. If this is false, the job will fail in case of a parse error.
json.timestamp-format.standard
No
SQL
Specify the format for JSON timestamp fields. The default is SQL (format: yyyy-MM-dd HH:mm:ss.s{optional precision}). You can also choose ISO-8601, with the format yyyy-MM-ddTHH:mm:ss.s{optional precision}.

WITH parameters for CSV

Parameter Value
Required
Default Value
Description
csv.field-delimiter
No
,
The CSV field delimiter, which is a half-width comma by default.
csv.line-delimiter
No
U&'\\000A'
Specify the CSV row delimiter. The default is the newline character \\n, which must be represented as U&'\\000A' in SQL. If you need to use the carriage return character \\r, it must be represented as U&'\\000D' in SQL.
csv.disable-quote-character
No
false
Whether to disable quote characters. If this is true, 'csv.quote-character' cannot be used.
csv.quote-character
No
''
The quote character for field enclosure. Content inside quotes is treated as a whole. The default is ''.
csv.ignore-parse-errors
No
false
Whether to ignore parse errors. If this is true, fields will be set to null in case of parse failure.
csv.allow-comments
No
false
Whether to ignore comment lines that start with # and output them as empty lines (if this is true, make sure you set csv.ignore-parse-errors to true as well).
csv.array-element-delimiter
No
;
The delimiter for array elements. The default is ;.
csv.escape-character
No
N/A
The escape character. By default, escape characters are disabled.
csv.null-literal
No
N/A
The string that will be seen as null.

Sample Code

CREATE TABLE `cmq_source_json_table` (
    `id` int,
    `name` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED	-- To deduplicate data, you must specify a primary key (PK) to distinguish different data records based on the key.
) WITH (
    'connector' = 'cmq',			   -- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',	-- The nameServer for the region where cmq is located
    'queue' = 'queue_name',			-- The queue name for cmq
    'secret-id' = 'xxxx',			  -- The account's secretId
    'secret-key' = 'xxxx',             -- The account's secretKey
    'sign-method' = 'HmacSHA1',        -- The signature method
    'format' = 'json',                 -- Define the data format (JSON).
    'json.fail-on-missing-field' = 'false',	 -- If this is set to 'false', no error will be reported when missing fields are encountered.
    'json.ignore-parse-errors' = 'true',   	 -- If this is set to 'true', any parse errors will be ignored.
    'batch-size' = '16',               -- The number of messages to consume in a batch
    'request-timeout' = '5000ms',  	-- The timeout duration for requests
    'polling-wait-timeout'= '10s', 	-- The wait time when no data can be obtained
    'key-alive-timeout'= '5min'    	-- The validity period for CMQ deduplication of messages containing a primary key
);
CREATE TABLE `cmq_sink_json_table` (
    `id` int,
    `name` STRING
) WITH (
    'connector' = 'cmq', 					 	-- Must be 'cmq'
    'hosts' = 'http://cmq-nameserver-vpc-gz.api.tencentyun.com',    -- The nameServer for the region where cmq is located
    'queue' = 'queue_name',			 		-- The queue name for cmq
    'secret-id' = 'xxxx',				   	-- The account's secretId
    'secret-key' = 'xxxx',         		 	-- The account's secretKey
    'sign-method' = 'HmacSHA1',     			-- The signature method
    'format' = 'json',             		 	-- Defines the data format (JSON).
    'json.fail-on-missing-field' = 'false',	 -- If this is set to 'false', no error will be reported when missing fields are encountered.
    'json.ignore-parse-errors' = 'true',         -- If this is set to 'true', any parse errors will be ignored.
    'batch-size' = '16',                        -- The number of messages to send in a batch
    'request-timeout' = '5000ms',               -- The timeout duration for requests
    'retry-times' = '3',                        -- The number of retries for sending messages
    'max-block-timeout' = '0s'                  -- The maximum wait time for sending data in batches
);
insert into cmq_sink_json_table select * from cmq_source_json_table;

Must-Knows

When using CMQ as a data source (Source), consider the following points:
1. To deduplicate data, you can set a primary key (PRIMARY KEY) to specify the deduplication key. Data deduplication can be achieved within a specified time window, which is user-configurable. Note that the longer this time window is set, the more memory is consumed.
2. It is strongly recommended to set the CMQ message invisibility duration to be greater than the Flink task's checkpoint interval. Otherwise, consumed messages may be consumed again, which impacts message processing performance.
Topik sebelumnya: Upsert KafkaTopik selanjutnya: TDMQ for RabbitMQ

Bantuan dan Dukungan

Apakah halaman ini membantu?

masukan

Search
    Enter keywords to search documentation