diff --git a/docs/examples/c/tmq.c b/docs/examples/c/tmq.c deleted file mode 100644 index 15ab4fcfc9..0000000000 --- a/docs/examples/c/tmq.c +++ /dev/null @@ -1,333 +0,0 @@ -/* - * Copyright (c) 2019 TAOS Data, Inc. - * - * This program is free software: you can use, redistribute, and/or modify - * it under the terms of the GNU Affero General Public License, version 3 - * or later ("AGPL"), as published by the Free Software Foundation. - * - * This program is distributed in the hope that it will be useful, but WITHOUT - * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or - * FITNESS FOR A PARTICULAR PURPOSE. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see . - */ - -#include -#include -#include -#include -#include -#include "taos.h" - -static int running = 1; -const char* topic_name = "topicname"; - -static int32_t msg_process(TAOS_RES* msg) { - char buf[1024]; - int32_t rows = 0; - - const char* topicName = tmq_get_topic_name(msg); - const char* dbName = tmq_get_db_name(msg); - int32_t vgroupId = tmq_get_vgroup_id(msg); - - printf("topic: %s\n", topicName); - printf("db: %s\n", dbName); - printf("vgroup id: %d\n", vgroupId); - - while (1) { - TAOS_ROW row = taos_fetch_row(msg); - if (row == NULL) break; - - TAOS_FIELD* fields = taos_fetch_fields(msg); - int32_t numOfFields = taos_field_count(msg); - // int32_t* length = taos_fetch_lengths(msg); - int32_t precision = taos_result_precision(msg); - rows++; - taos_print_row(buf, row, fields, numOfFields); - printf("precision: %d, row content: %s\n", precision, buf); - } - - return rows; -} - -static int32_t init_env() { - TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); - if (pConn == NULL) { - return -1; - } - - TAOS_RES* pRes; - // drop database if exists - printf("create database\n"); - pRes = taos_query(pConn, "drop topic topicname"); - if (taos_errno(pRes) != 0) { - printf("error in drop topicname, reason:%s\n", taos_errstr(pRes)); - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "drop database if exists tmqdb"); - if (taos_errno(pRes) != 0) { - printf("error in drop tmqdb, reason:%s\n", taos_errstr(pRes)); - } - taos_free_result(pRes); - - // create database - pRes = taos_query(pConn, "create database tmqdb precision 'ns' WAL_RETENTION_PERIOD 3600"); - if (taos_errno(pRes) != 0) { - printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - // create super table - printf("create super table\n"); - pRes = taos_query( - pConn, "create table tmqdb.stb (ts timestamp, c1 int, c2 float, c3 varchar(16)) tags(t1 int, t3 varchar(16))"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table stb, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - // create sub tables - printf("create sub tables\n"); - pRes = taos_query(pConn, "create table tmqdb.ctb0 using tmqdb.stb tags(0, 'subtable0')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table ctb0, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table tmqdb.ctb1 using tmqdb.stb tags(1, 'subtable1')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table ctb1, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table tmqdb.ctb2 using tmqdb.stb tags(2, 'subtable2')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table ctb2, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create table tmqdb.ctb3 using tmqdb.stb tags(3, 'subtable3')"); - if (taos_errno(pRes) != 0) { - printf("failed to create super table ctb3, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - // insert data - printf("insert data into sub tables\n"); - pRes = taos_query(pConn, "insert into tmqdb.ctb0 values(now, 0, 0, 'a0')(now+1s, 0, 0, 'a00')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into tmqdb.ctb1 values(now, 1, 1, 'a1')(now+1s, 11, 11, 'a11')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into tmqdb.ctb2 values(now, 2, 2, 'a1')(now+1s, 22, 22, 'a22')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "insert into tmqdb.ctb3 values(now, 3, 3, 'a1')(now+1s, 33, 33, 'a33')"); - if (taos_errno(pRes) != 0) { - printf("failed to insert into ctb0, reason:%s\n", taos_errstr(pRes)); - goto END; - } - taos_free_result(pRes); - taos_close(pConn); - return 0; - -END: - taos_free_result(pRes); - taos_close(pConn); - return -1; -} - -int32_t create_topic() { - printf("create topic\n"); - TAOS_RES* pRes; - TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); - if (pConn == NULL) { - return -1; - } - - pRes = taos_query(pConn, "use tmqdb"); - if (taos_errno(pRes) != 0) { - printf("error in use tmqdb, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - pRes = taos_query(pConn, "create topic topicname as select ts, c1, c2, c3, tbname from tmqdb.stb where c1 > 1"); - if (taos_errno(pRes) != 0) { - printf("failed to create topic topicname, reason:%s\n", taos_errstr(pRes)); - return -1; - } - taos_free_result(pRes); - - taos_close(pConn); - return 0; -} - -void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { - printf("tmq_commit_cb_print() code: %d, tmq: %p, param: %p\n", code, tmq, param); -} - -tmq_t* build_consumer() { - tmq_conf_res_t code; - tmq_t* tmq = NULL; - - tmq_conf_t* conf = tmq_conf_new(); - code = tmq_conf_set(conf, "enable.auto.commit", "true"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "group.id", "cgrpName"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "client.id", "user defined name"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "td.connect.user", "root"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "td.connect.pass", "taosdata"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - code = tmq_conf_set(conf, "auto.offset.reset", "earliest"); - if (TMQ_CONF_OK != code) { - tmq_conf_destroy(conf); - return NULL; - } - - tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); - tmq = tmq_consumer_new(conf, NULL, 0); - -_end: - tmq_conf_destroy(conf); - return tmq; -} - -tmq_list_t* build_topic_list() { - tmq_list_t* topicList = tmq_list_new(); - int32_t code = tmq_list_append(topicList, topic_name); - if (code) { - tmq_list_destroy(topicList); - return NULL; - } - return topicList; -} - -void basic_consume_loop(tmq_t* tmq) { - int32_t totalRows = 0; - int32_t msgCnt = 0; - int32_t timeout = 5000; - while (running) { - TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout); - if (tmqmsg) { - msgCnt++; - totalRows += msg_process(tmqmsg); - taos_free_result(tmqmsg); - } else { - break; - } - } - - fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows); -} - -void consume_repeatly(tmq_t* tmq) { - int32_t numOfAssignment = 0; - tmq_topic_assignment* pAssign = NULL; - - int32_t code = tmq_get_topic_assignment(tmq, topic_name, &pAssign, &numOfAssignment); - if (code != 0) { - fprintf(stderr, "failed to get assignment, reason:%s", tmq_err2str(code)); - } - - // seek to the earliest offset - for(int32_t i = 0; i < numOfAssignment; ++i) { - tmq_topic_assignment* p = &pAssign[i]; - - code = tmq_offset_seek(tmq, topic_name, p->vgId, p->begin); - if (code != 0) { - fprintf(stderr, "failed to seek to %d, reason:%s", (int)p->begin, tmq_err2str(code)); - } - } - - tmq_free_assignment(pAssign); - - // let's do it again - basic_consume_loop(tmq); -} - -int main(int argc, char* argv[]) { - int32_t code; - - if (init_env() < 0) { - return -1; - } - - if (create_topic() < 0) { - return -1; - } - - tmq_t* tmq = build_consumer(); - if (NULL == tmq) { - fprintf(stderr, "build_consumer() fail!\n"); - return -1; - } - - tmq_list_t* topic_list = build_topic_list(); - if (NULL == topic_list) { - return -1; - } - - if ((code = tmq_subscribe(tmq, topic_list))) { - fprintf(stderr, "Failed to tmq_subscribe(): %s\n", tmq_err2str(code)); - } - - tmq_list_destroy(topic_list); - - basic_consume_loop(tmq); - - consume_repeatly(tmq); - - code = tmq_consumer_close(tmq); - if (code) { - fprintf(stderr, "Failed to close consumer: %s\n", tmq_err2str(code)); - } else { - fprintf(stderr, "Consumer closed\n"); - } - - return 0; -} diff --git a/docs/examples/c/tmq_demo.c b/docs/examples/c/tmq_demo.c new file mode 100644 index 0000000000..a33896f670 --- /dev/null +++ b/docs/examples/c/tmq_demo.c @@ -0,0 +1,412 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * This program is free software: you can use, redistribute, and/or modify + * it under the terms of the GNU Affero General Public License, version 3 + * or later ("AGPL"), as published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +#include +#include +#include +#include +#include +#include +#include +#include "taos.h" + +volatile int thread_stop = 0; +static int running = 1; +const char* topic_name = "topic_meters"; + +void* prepare_data(void* arg) { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return NULL; + } + + TAOS_RES* pRes; + int i = 1; + + while (!thread_stop) { + char buf[200] = {0}; + i++; + snprintf( + buf, sizeof(buf), + "INSERT INTO power.d1001 USING power.meters TAGS(2,'California.SanFrancisco') VALUES (NOW + %da, 10.30000, " + "219, 0.31000)", + i); + + pRes = taos_query(pConn, buf); + if (taos_errno(pRes) != 0) { + printf("error in insert data to power.meters, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); + sleep(1); + } + printf("prepare data thread exit\n"); + return NULL; +} + +// ANCHOR: msg_process +static int32_t msg_process(TAOS_RES* msg) { + char buf[1024]; // buf to store the row content + int32_t rows = 0; + const char* topicName = tmq_get_topic_name(msg); + const char* dbName = tmq_get_db_name(msg); + int32_t vgroupId = tmq_get_vgroup_id(msg); + + printf("topic: %s\n", topicName); + printf("db: %s\n", dbName); + printf("vgroup id: %d\n", vgroupId); + + while (1) { + // get one row data from message + TAOS_ROW row = taos_fetch_row(msg); + if (row == NULL) break; + + // get the field information + TAOS_FIELD* fields = taos_fetch_fields(msg); + // get the number of fields + int32_t numOfFields = taos_field_count(msg); + // get the precision of the result + int32_t precision = taos_result_precision(msg); + rows++; + // print the row content + if (taos_print_row(buf, row, fields, numOfFields) < 0) { + printf("failed to print row\n"); + break; + } + // print the precision and row content to the console + printf("precision: %d, row content: %s\n", precision, buf); + } + + return rows; +} +// ANCHOR_END: msg_process + +static int32_t init_env() { + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + TAOS_RES* pRes; + // drop database if exists + printf("create database\n"); + pRes = taos_query(pConn, "drop topic if exists topic_meters"); + if (taos_errno(pRes) != 0) { + printf("error in drop topic_meters, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop database if exists power"); + if (taos_errno(pRes) != 0) { + printf("error in drop power, reason:%s\n", taos_errstr(pRes)); + } + taos_free_result(pRes); + + // create database + pRes = taos_query(pConn, "create database power precision 'ms' WAL_RETENTION_PERIOD 3600"); + if (taos_errno(pRes) != 0) { + printf("error in create tmqdb, reason:%s\n", taos_errstr(pRes)); + goto END; + } + taos_free_result(pRes); + + // create super table + printf("create super table\n"); + pRes = taos_query( + pConn, + "CREATE STABLE IF NOT EXISTS power.meters (ts TIMESTAMP, current FLOAT, voltage INT, phase FLOAT) TAGS " + "(groupId INT, location BINARY(24))"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table meters, reason:%s\n", taos_errstr(pRes)); + goto END; + } + + taos_free_result(pRes); + taos_close(pConn); + return 0; + +END: + taos_free_result(pRes); + taos_close(pConn); + return -1; +} + +int32_t create_topic() { + printf("create topic\n"); + TAOS_RES* pRes; + TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0); + if (pConn == NULL) { + return -1; + } + + pRes = taos_query(pConn, "use power"); + if (taos_errno(pRes) != 0) { + printf("error in use tmqdb, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query( + pConn, + "CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters"); + if (taos_errno(pRes) != 0) { + printf("failed to create topic topic_meters, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + taos_close(pConn); + return 0; +} + +void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) { + printf("tmq_commit_cb_print() code: %d, tmq: %p, param: %p\n", code, tmq, param); +} + +// ANCHOR: create_consumer_1 +tmq_t* build_consumer() { + tmq_conf_res_t code; + tmq_t* tmq = NULL; + + // create a configuration object + tmq_conf_t* conf = tmq_conf_new(); + + // set the configuration parameters + code = tmq_conf_set(conf, "enable.auto.commit", "true"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "group.id", "group1"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "client.id", "client1"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "td.connect.user", "root"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "td.connect.pass", "taosdata"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + code = tmq_conf_set(conf, "auto.offset.reset", "latest"); + if (TMQ_CONF_OK != code) { + tmq_conf_destroy(conf); + return NULL; + } + + // set the callback function for auto commit + tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL); + // create a consumer object + tmq = tmq_consumer_new(conf, NULL, 0); + +_end: + // destroy the configuration object + tmq_conf_destroy(conf); + return tmq; +} +// ANCHOR_END: create_consumer_1 + +// ANCHOR: build_topic_list +// build a topic list used to subscribe +tmq_list_t* build_topic_list() { + // create a empty topic list + tmq_list_t* topicList = tmq_list_new(); + const char* topic_name = "topic_meters"; + + // append topic name to the list + int32_t code = tmq_list_append(topicList, topic_name); + if (code) { + // if failed, destroy the list and return NULL + tmq_list_destroy(topicList); + return NULL; + } + // if success, return the list + return topicList; +} +// ANCHOR_END: build_topic_list + +// ANCHOR: basic_consume_loop +void basic_consume_loop(tmq_t* tmq) { + int32_t totalRows = 0; // total rows consumed + int32_t msgCnt = 0; // total messages consumed + int32_t timeout = 5000; // poll timeout + + while (running) { + // poll message from TDengine + TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout); + if (tmqmsg) { + msgCnt++; + // process the message + totalRows += msg_process(tmqmsg); + // free the message + taos_free_result(tmqmsg); + } + if (msgCnt > 10) { + // consume 10 messages and break + break; + } + } + + // print the result: total messages and total rows consumed + fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows); +} +// ANCHOR_END: basic_consume_loop + +// ANCHOR: consume_repeatly +void consume_repeatly(tmq_t* tmq) { + int32_t numOfAssignment = 0; + tmq_topic_assignment* pAssign = NULL; + + // get the topic assignment + int32_t code = tmq_get_topic_assignment(tmq, topic_name, &pAssign, &numOfAssignment); + if (code != 0 || pAssign == NULL || numOfAssignment == 0) { + fprintf(stderr, "failed to get assignment, reason:%s", tmq_err2str(code)); + return; + } + + // seek to the earliest offset + for (int32_t i = 0; i < numOfAssignment; ++i) { + tmq_topic_assignment* p = &pAssign[i]; + + code = tmq_offset_seek(tmq, topic_name, p->vgId, p->begin); + if (code != 0) { + fprintf(stderr, "failed to seek to %d, reason:%s", (int)p->begin, tmq_err2str(code)); + } + } + + // free the assignment array + tmq_free_assignment(pAssign); + + // let's consume the messages again + basic_consume_loop(tmq); +} +// ANCHOR_END: consume_repeatly + +// ANCHOR: manual_commit +void manual_commit(tmq_t* tmq) { + int32_t totalRows = 0; // total rows consumed + int32_t msgCnt = 0; // total messages consumed + int32_t timeout = 5000; // poll timeout + + while (running) { + // poll message from TDengine + TAOS_RES* tmqmsg = tmq_consumer_poll(tmq, timeout); + if (tmqmsg) { + msgCnt++; + // process the message + totalRows += msg_process(tmqmsg); + // commit the message + int32_t code = tmq_commit_sync(tmq, tmqmsg); + + if (code) { + fprintf(stderr, "Failed to commit message: %s\n", tmq_err2str(code)); + // free the message + taos_free_result(tmqmsg); + break; + } + // free the message + taos_free_result(tmqmsg); + } + if (msgCnt > 10) { + // consume 10 messages and break + break; + } + } + + // print the result: total messages and total rows consumed + fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows); +} +// ANCHOR_END: manual_commit + +int main(int argc, char* argv[]) { + int32_t code; + pthread_t thread_id; + + if (init_env() < 0) { + return -1; + } + + if (create_topic() < 0) { + return -1; + } + + if (pthread_create(&thread_id, NULL, &prepare_data, NULL)) { + fprintf(stderr, "create thread failed\n"); + return 1; + } + + // ANCHOR: create_consumer_2 + tmq_t* tmq = build_consumer(); + if (NULL == tmq) { + fprintf(stderr, "build consumer to localhost fail!\n"); + return -1; + } + printf("build consumer to localhost successfully \n"); + + // ANCHOR_END: create_consumer_2 + + // ANCHOR: subscribe_3 + tmq_list_t* topic_list = build_topic_list(); + if (NULL == topic_list) { + return -1; + } + + if ((code = tmq_subscribe(tmq, topic_list))) { + fprintf(stderr, "Failed to tmq_subscribe(): %s\n", tmq_err2str(code)); + } + + tmq_list_destroy(topic_list); + + basic_consume_loop(tmq); + // ANCHOR_END: subscribe_3 + + consume_repeatly(tmq); + + manual_commit(tmq); + + // ANCHOR: unsubscribe_and_close + // unsubscribe the topic + code = tmq_unsubscribe(tmq); + if (code) { + fprintf(stderr, "Failed to tmq_unsubscribe(): %s\n", tmq_err2str(code)); + } + fprintf(stderr, "Unsubscribed consumer successfully.\n"); + // close the consumer + code = tmq_consumer_close(tmq); + if (code) { + fprintf(stderr, "Failed to close consumer: %s\n", tmq_err2str(code)); + } else { + fprintf(stderr, "Consumer closed successfully.\n"); + } + // ANCHOR_END: unsubscribe_and_close + + thread_stop = 1; + pthread_join(thread_id, NULL); + + return 0; +} diff --git a/docs/zh/08-develop/07-tmq.md b/docs/zh/08-develop/07-tmq.md index fa6770a29f..ba6c24884e 100644 --- a/docs/zh/08-develop/07-tmq.md +++ b/docs/zh/08-develop/07-tmq.md @@ -93,6 +93,8 @@ Rust 连接器创建消费者的参数为 DSN， 可以设置的参数列表请 +同通用基础配置项。 + @@ -193,7 +195,16 @@ Rust 连接器创建消费者的参数为 DSN， 可以设置的参数列表请 - + +```c +{{#include docs/examples/c/tmq_demo.c:create_consumer_1}} +``` + +```c +{{#include docs/examples/c/tmq_demo.c:create_consumer_2}} +``` + +调用 `build_consumer` 函数尝试获取消费者实例 `tmq`。成功则打印成功日志，失败则打印失败日志。 @@ -294,7 +305,29 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur - +```c +{{#include docs/examples/c/tmq_demo.c:build_topic_list}} +``` + +```c +{{#include docs/examples/c/tmq_demo.c:basic_consume_loop}} +``` + +```c +{{#include docs/examples/c/tmq_demo.c:msg_process}} +``` + +```c +{{#include docs/examples/c/tmq_demo.c:subscribe_3}} +``` + +订阅消费数据步骤： + 1. 调用 `build_topic_list` 函数创建一个主题列表 `topic_list`。 + 2. 如果 `topic_list` 为 `NULL`，表示创建失败，函数返回 `-1`。 + 3. 使用 `tmq_subscribe` 函数订阅 `tmq` 指定的主题列表。如果订阅失败，打印错误信息。 + 4. 销毁主题列表 `topic_list` 以释放资源。 + 5. 调用 `basic_consume_loop` 函数开始基本的消费循环，处理订阅的消息。 + @@ -406,6 +439,16 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur +```c +{{#include docs/examples/c/tmq_demo.c:consume_repeatly}} +``` + +1. 通过 `tmq_get_topic_assignment` 函数获取特定主题的分配信息，包括分配的数量和具体分配详情。 +2. 如果获取分配信息失败，则打印错误信息并返回。 +3. 对于每个分配，使用 `tmq_offset_seek` 函数将消费者的偏移量设置到最早的偏移量。 +4. 如果设置偏移量失败，则打印错误信息。 +5. 释放分配信息数组以释放资源。 +6. 调用 `basic_consume_loop` 函数开始新的的消费循环，处理消息。 @@ -495,10 +538,6 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur 可以通过 `consumer.commit` 方法来手工提交消费进度。 - - - - ```csharp {{#include docs/examples/csharp/subscribe/Program.cs:commit_offset}} @@ -506,7 +545,12 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur - +```c +{{#include docs/examples/c/tmq_demo.c:manual_commit}} +``` + +可以通过 `tmq_commit_sync` 函数来手工提交消费进度。 + @@ -589,10 +633,6 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur **注意**：消费者取消订阅后无法重用，如果想订阅新的 `topic`， 请重新创建消费者。 - - - - ```csharp {{#include docs/examples/csharp/subscribe/Program.cs:close}} @@ -600,7 +640,9 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur - +```c +{{#include docs/examples/c/tmq_demo.c:unsubscribe_and_close}} +``` @@ -693,10 +735,6 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur ``` - - - - ```csharp {{#include docs/examples/csharp/subscribe/Program.cs}} @@ -704,6 +742,11 @@ TMQ 消息队列是一个 [futures::Stream](https://docs.rs/futures/latest/futur - +
+完整原生连接代码示例 +```c +{{#include docs/examples/c/tmq_demo.c}} +``` +