From 5b3f85ed3ba569a3cf867603d6747f991b5f5636 Mon Sep 17 00:00:00 2001 From: xleili Date: Tue, 16 Aug 2022 15:22:09 +0800 Subject: [PATCH 1/4] docs(driver):update C# 3.0 tmq doc in develop/tmq.mdx --- docs/zh/07-develop/07-tmq.mdx | 90 +++++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index f36f76fd85..957a8bbdda 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -87,6 +87,31 @@ void commitSync() throws SQLException; void close() throws SQLException; ``` + + + +```C# +ConsumerBuilder(IEnumerable> config) + +virtual IConsumer Build() + +Consumer(ConsumerBuilder builder) + +void Subscribe(IEnumerable topics) + +void Subscribe(string topic) + +ConsumeResult Consume(int millisecondsTimeout) + +List Subscription() + +void Unsubscribe() + +void Commit(ConsumeResult consumerResult) + +void Close() +``` + @@ -229,6 +254,30 @@ public class MetersDeserializer extends ReferenceDeserializer { } ``` + + + +```C# +using TDengineTMQ; + +// 根据需要，设置消费组 (GourpId)、自动提交 (EnableAutoCommit)、 +// 自动提交时间间隔 (AutoCommitIntervalMs)、用户名 (TDConnectUser)、密码 (TDConnectPasswd) 等参数 +var cfg = new ConsumerConfig + { + EnableAutoCommit = "true" + AutoCommitIntervalMs = "1000" + GourpId = "TDengine-TMQ-C#", + TDConnectUser = "root", + TDConnectPasswd = "taosdata", + AutoOffsetReset = "earliest" + MsgWithTableName = "true", + TDConnectIp = "127.0.0.1", + TDConnectPort = "6030" + }; + +var consumer = new ConsumerBuilder(cfg).Build(); +``` + @@ -262,6 +311,18 @@ consumer.subscribe(topics); + + +```C# +// 创建订阅 topics 列表 +List topics = new List(); +topics.add("tmq_topic"); +// 启动订阅 +consumer.Subscribe(topics); +``` + + + ## 消费 @@ -296,6 +357,23 @@ while(running){ + + +```C# +// 消费数据 +while (true) +{ + var consumerRes = consumer.Consume(100); + // process ConsumeResult + ProcessMsg(consumerRes); + consumer.Commit(consumerRes); +} +``` + + + + + ## 结束消费 消费结束后，应当取消订阅。 @@ -322,6 +400,18 @@ consumer.unsubscribe(); consumer.close(); ``` + + + + +```C# +// 取消订阅 +consumer.Unsubscribe(); + +// 关闭消费 +consumer.Close(); +``` + From 0db3fe1e1e617e03435238e327a31e98c803b72d Mon Sep 17 00:00:00 2001 From: xleili Date: Tue, 16 Aug 2022 15:39:22 +0800 Subject: [PATCH 2/4] docs(driver):update markdown syntax mistake caused by merge --- docs/zh/07-develop/07-tmq.mdx | 105 ++++++++++++++++++---------------- 1 file changed, 56 insertions(+), 49 deletions(-) diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index 985b6a6ff2..fb308c5c40 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -89,6 +89,26 @@ void close() throws SQLException; + + +```go +func NewConsumer(conf *Config) (*Consumer, error) + +func (c *Consumer) Close() error + +func (c *Consumer) Commit(ctx context.Context, message unsafe.Pointer) error + +func (c *Consumer) FreeMessage(message unsafe.Pointer) + +func (c *Consumer) Poll(timeout time.Duration) (*Result, error) + +func (c *Consumer) Subscribe(topics []string) error + +func (c *Consumer) Unsubscribe() error + +``` + + ```C# @@ -111,27 +131,10 @@ void Unsubscribe() void Commit(ConsumeResult consumerResult) void Close() - - - - -```go -func NewConsumer(conf *Config) (*Consumer, error) - -func (c *Consumer) Close() error - -func (c *Consumer) Commit(ctx context.Context, message unsafe.Pointer) error - -func (c *Consumer) FreeMessage(message unsafe.Pointer) - -func (c *Consumer) Poll(timeout time.Duration) (*Result, error) - -func (c *Consumer) Subscribe(topics []string) error - -func (c *Consumer) Unsubscribe() error ``` + ## 写入数据 @@ -275,29 +278,6 @@ public class MetersDeserializer extends ReferenceDeserializer { - - -```C# -using TDengineTMQ; - -// 根据需要，设置消费组 (GourpId)、自动提交 (EnableAutoCommit)、 -// 自动提交时间间隔 (AutoCommitIntervalMs)、用户名 (TDConnectUser)、密码 (TDConnectPasswd) 等参数 -var cfg = new ConsumerConfig - { - EnableAutoCommit = "true" - AutoCommitIntervalMs = "1000" - GourpId = "TDengine-TMQ-C#", - TDConnectUser = "root", - TDConnectPasswd = "taosdata", - AutoOffsetReset = "earliest" - MsgWithTableName = "true", - TDConnectIp = "127.0.0.1", - TDConnectPort = "6030" - }; - -var consumer = new ConsumerBuilder(cfg).Build(); - - ```go @@ -348,6 +328,32 @@ if err != nil { ``` + + + +```C# +using TDengineTMQ; + +// 根据需要，设置消费组 (GourpId)、自动提交 (EnableAutoCommit)、 +// 自动提交时间间隔 (AutoCommitIntervalMs)、用户名 (TDConnectUser)、密码 (TDConnectPasswd) 等参数 +var cfg = new ConsumerConfig + { + EnableAutoCommit = "true" + AutoCommitIntervalMs = "1000" + GourpId = "TDengine-TMQ-C#", + TDConnectUser = "root", + TDConnectPasswd = "taosdata", + AutoOffsetReset = "earliest" + MsgWithTableName = "true", + TDConnectIp = "127.0.0.1", + TDConnectPort = "6030" + }; + +var consumer = new ConsumerBuilder(cfg).Build(); + +``` + + 上述配置中包括 consumer group ID，如果多个 consumer 指定的 consumer group ID 一样，则自动形成一个 consumer group，共享消费进度。 @@ -500,6 +506,15 @@ consumer.close(); + + +```go +consumer.Close() + +``` + + + ```C# @@ -510,14 +525,6 @@ consumer.Unsubscribe(); consumer.Close(); - - -```go -consumer.Close() - -``` - - ## 删除 *topic* From a0525216c0c3a6d49696b76b7f03f36f2b0e5bf5 Mon Sep 17 00:00:00 2001 From: xleili Date: Tue, 16 Aug 2022 15:44:51 +0800 Subject: [PATCH 3/4] docs(driver):update C# tmq document --- docs/zh/07-develop/07-tmq.mdx | 5 ----- 1 file changed, 5 deletions(-) diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index fb308c5c40..c37a3c4737 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -88,7 +88,6 @@ void close() throws SQLException; ``` - ```go @@ -105,7 +104,6 @@ func (c *Consumer) Poll(timeout time.Duration) (*Result, error) func (c *Consumer) Subscribe(topics []string) error func (c *Consumer) Unsubscribe() error - ``` @@ -277,7 +275,6 @@ public class MetersDeserializer extends ReferenceDeserializer { ``` - ```go @@ -505,7 +502,6 @@ consumer.close(); ``` - ```go @@ -514,7 +510,6 @@ consumer.Close() ``` - ```C# From 093ff2bd54b50f41895151773872b9c1ee73069c Mon Sep 17 00:00:00 2001 From: xleili Date: Tue, 16 Aug 2022 15:46:07 +0800 Subject: [PATCH 4/4] docs(driver):update TMQ markdown doc --- docs/zh/07-develop/07-tmq.mdx | 1 - 1 file changed, 1 deletion(-) diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index c37a3c4737..0945f69fcf 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -506,7 +506,6 @@ consumer.close(); ```go consumer.Close() - ```