TDengine C# Connector

TDengine.Connector 是 TDengine 提供的 C# 语言连接器。C# 开发人员可以通过它开发存取 TDengine 集群数据的 C# 应用软件。

连接方式

TDengine.Connector 提供两种形式的连接器

• 原生连接，通过 TDengine 客户端驱动程序（taosc）原生连接 TDengine 实例，支持数据写入、查询、数据订阅、schemaless 接口和参数绑定接口等功能。
• WebSocket 连接，通过 taosAdapter 提供的 WebSocket 接口连接 TDengine 实例，WebSocket 连接实现的功能集合和原生连接有少量不同。（自 v3.0.1 起）

连接方式的详细介绍请参考：连接方式

兼容性

• TDengine.Connector 3.1.0 版本进行了完整的重构，不再兼容 3.0.2 及以前版本。3.0.2 文档请参考 nuget
• TDengine.Connector 3.x 不兼容 TDengine 2.x，如果在运行 TDengine 2.x 版本的环境下需要使用 C# 连接器请使用 TDengine.Connector 的 1.x 版本。

支持的平台

支持的平台和 TDengine 客户端驱动支持的平台一致。

warning

TDengine 不再支持 32 位 Windows 平台。

版本支持

Connector 版本	TDengine 版本	主要功能
3.1.3	3.2.1.0/3.1.1.18	支持 WebSocket 自动重连
3.1.2	3.2.1.0/3.1.1.18	修复 schemaless 资源释放
3.1.1	3.2.1.0/3.1.1.18	支持 varbinary 和 geometry 类型
3.1.0	3.2.1.0/3.1.1.18	WebSocket 使用原生实现

处理异常

TDengine.Connector 会抛出异常，应用程序需要处理异常。taosc 异常类型 TDengineError，包含错误码和错误信息，应用程序可以根据错误码和错误信息进行处理。 TDengine 其他功能模块的报错，请参考 错误码

数据类型映射

TDengine DataType	C# Type
TIMESTAMP	DateTime
TINYINT	sbyte
SMALLINT	short
INT	int
BIGINT	long
TINYINT UNSIGNED	byte
SMALLINT UNSIGNED	ushort
INT UNSIGNED	uint
BIGINT UNSIGNED	ulong
FLOAT	float
DOUBLE	double
BOOL	bool
BINARY	byte[]
NCHAR	string (utf-8编码)
JSON	byte[]
VARBINARY	byte[]
GEOMETRY	byte[]

注意：JSON 类型仅在 tag 中支持。 GEOMETRY类型是 little endian 字节序的二进制数据，符合 WKB 规范。详细信息请参考 数据类型 WKB规范请参考Well-Known Binary (WKB)

示例程序汇总

示例程序源码请参考：示例程序

API 参考

ADO.NET 驱动

TDengine.Data.Client 接口实现了 ADO.NET 驱动，支持连接 TDengine 数据库，进行数据操作。

参数规范

ConnectionStringBuilder 使用 key-value 对方式设置连接参数，key 为参数名，value 为参数值，不同参数之间使用分号 ; 分割。

例如：

"protocol=WebSocket;host=127.0.0.1;port=6041;useSSL=false"

原生连接

例如："host=127.0.0.1;port=6030;username=root;password=taosdata;protocol=Native;db=test"

支持的参数如下：

• host：TDengine 运行实例的地址。
• port：TDengine 运行实例的端口。
• username：连接的用户名。
• password：连接的密码。
• protocol：连接的协议，可选值为 Native 或 WebSocket，默认为 Native。
• db：连接的数据库。
• timezone：时区，默认为本地时区。

WebSocket 连接

例如："protocol=WebSocket;host=127.0.0.1;port=6041;useSSL=false;enableCompression=true;autoReconnect=true;reconnectIntervalMs=10;reconnectRetryCount=5"

支持的参数如下：

• host：TDengine 运行实例的地址。
• port：TDengine 运行实例的端口。
• username：连接的用户名。
• password：连接的密码。
• protocol：连接的协议，可选值为 Native 或 WebSocket，默认为 Native。
• db：连接的数据库。
• timezone：时区，默认为本地时区。
• connTimeout：连接超时时间，默认为 1 分钟。
• readTimeout：读取超时时间，默认为 5 分钟。
• writeTimeout：发送超时时间，默认为 10 秒。
• token：连接 TDengine cloud 的 token。
• useSSL：是否使用 SSL 连接，默认为 false。
• enableCompression：是否启用 WebSocket 压缩，默认为 false。
• autoReconnect：是否自动重连，默认为 false。
• reconnectRetryCount：重连次数，默认为 3。
• reconnectIntervalMs：重连间隔毫秒时间，默认为 2000。

接口说明

ConnectionStringBuilder 类提供了连接配置字符串的解析功能。

• public ConnectionStringBuilder(string connectionString)
  • 接口说明：ConnectionStringBuilder 构造函数。
  • 参数说明：
    • connectionString：连接配置字符串。

连接功能

C# 驱动支持创建 ADO.NET 连接，返回支持 ADO.NET 标准的 DbConnection 接口的对象，还提供了 ITDengineClient 接口，扩充了一些无模式写入接口。

标准接口

ADO.NET 连接支持的标准接口如下：

• public TDengineConnection(string connectionString)

  • 接口说明：TDengineConnection 构造函数。
  • 参数说明：
    • connectionString：连接配置字符串。
  • 异常：格式错误抛出 ArgumentException 异常。

• public void ChangeDatabase(string databaseName)

  • 接口说明：切换数据库。
  • 参数说明：
    • databaseName：数据库名。
  • 异常：执行失败抛出 TDengineError 异常。

• public void Close()

  • 接口说明：关闭连接。

• public void Open()

  • 接口说明：打开连接。
  • 异常：打开失败抛出 TDengineError 异常，WebSocket 连接可能存在网络异常须注意处理。

• public string ServerVersion

  • 接口说明：返回服务器版本。
  • 返回值：服务器版本字符串。
  • 异常：执行失败抛出 TDengineError 异常。

• public string DataSource

  • 接口说明：返回数据源。
  • 返回值：创建连接 host 配置。

• public string Database

  • 接口说明：返回连接数据库。
  • 返回值：创建连接 db 配置。

• public TDengineCommand(TDengineConnection connection)

  • 接口说明：TDengineCommand 构造函数。
  • 参数说明：
    • connection：TDengineConnection 对象。
  • 异常：执行失败抛出 TDengineError 异常。

• public void Prepare()

  • 接口说明：检查连接和命令文本，并准备命令执行。
  • 异常：未执行 open 或未设置 CommandText 抛出 InvalidOperationException 异常。

• public string CommandText

  • 接口说明：获取或设置命令文本。
  • 返回值：命令文本。

• public new virtual TDengineParameterCollection Parameters

  • 接口说明：获取参数集合。
  • 返回值：TDengineParameterCollection 对象。

无模式写入

• public static ITDengineClient Open(ConnectionStringBuilder builder)

  • 接口说明：打开连接。
  • 参数说明：
    • builder：连接配置。
  • 返回值：ITDengineClient 接口。
  • 异常：打开失败抛出 TDengineError 异常，WebSocket 连接可能存在网络异常须注意处理。

• void SchemalessInsert(string[] lines, TDengineSchemalessProtocol protocol,TDengineSchemalessPrecision precision, int ttl, long reqId)

  • 接口说明：无模式写入。
  • 参数说明：
    • lines：数据行数组。
    • protocol：数据协议，支持协议：TSDB_SML_LINE_PROTOCOL = 1 TSDB_SML_TELNET_PROTOCOL = 2 TSDB_SML_JSON_PROTOCOL = 3。
    • precision：时间精度，支持配置：TSDB_SML_TIMESTAMP_NOT_CONFIGURED = 0 TSDB_SML_TIMESTAMP_HOURS = 1 TSDB_SML_TIMESTAMP_MINUTES = 2 TSDB_SML_TIMESTAMP_SECONDS = 3 TSDB_SML_TIMESTAMP_MILLI_SECONDS = 4 TSDB_SML_TIMESTAMP_MICRO_SECONDS = 5 TSDB_SML_TIMESTAMP_NANO_SECONDS = 6。
    • ttl：数据过期时间，0表示不配置。
    • reqId：请求 ID。
  • 异常：执行失败抛出 TDengineError 异常。

执行 SQL

C# 驱动提供了符合 ADO.NET 标准的 DbCommand 接口，支持以下功能：

1. 执行 SQL 语句：执行静态 SQL 语句，并返回其生成的结果对象。
2. 查询执行：可以执行返回数据集的查询（SELECT 语句）。
3. 更新执行：可以执行影响行数的 SQL 语句，如 INSERT、UPDATE、DELETE 等。
4. 获取结果：可以获取查询执行后返回的结果集（ResultSet 对象），并遍历查询返回的数据。
5. 获取更新计数：对于非查询 SQL 语句，可以获取执行后影响的行数。
6. 关闭资源：提供了关闭的方法，以释放数据库资源。

另外 C# 驱动还提供了用于请求链路跟踪的扩展接口。

标准接口

• public int ExecuteNonQuery()

  • 接口说明：执行 SQL 语句，返回受影响的行数。
  • 返回值：受影响的行数。
  • 异常：执行失败抛出 TDengineError 异常。

• public object ExecuteScalar()

  • 接口说明：执行查询，并返回查询结果的第一行第一列。
  • 返回值：查询结果的第一行第一列。
  • 异常：执行失败抛出 TDengineError 异常。

• public DbDataReader ExecuteReader()

  • 接口说明：执行查询，并返回查询结果的数据读取器。
  • 返回值：查询结果的数据读取器。
  • 异常：执行失败抛出 TDengineError 异常。

• public void Dispose();

  • 接口说明：释放资源。

扩展接口

扩展接口主要用于请求链路跟踪。

• IRows Query(string query, long reqId)

  • 接口说明：执行查询，返回查询结果。
  • 参数说明：
    • query：查询语句。
    • reqId：请求 ID。
  • 返回值：查询结果。
  • 异常：执行失败抛出 TDengineError 异常。

• long Exec(string query, long reqId)

  • 接口说明：执行 SQL 语句。
  • 参数说明：
    • query：SQL 语句。
    • reqId：请求 ID。
  • 返回值：受影响的行数。
  • 异常：执行失败抛出 TDengineError 异常。

结果获取

C# 驱动提供了符合 ADO.NET 标准的 DbDataReader 接口，提供了用于读取结果集中元数据和数据的方法。

结果集

DbDataReader 接口提供了以下方法获取结果集：

• public bool GetBoolean(int ordinal)

  • 接口说明：获取指定列的布尔值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：布尔值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public byte GetByte(int ordinal)

  • 接口说明：获取指定列的字节值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：字节值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public long GetBytes(int ordinal, long dataOffset, byte[] buffer, int bufferOffset, int length)

  • 接口说明：获取指定列的字节值。
  • 参数说明：
    • ordinal：列索引。
    • dataOffset：数据偏移量。
    • buffer：缓冲区。
    • bufferOffset：缓冲区偏移量。
    • length：长度。
  • 返回值：字节值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public char GetChar(int ordinal)

  • 接口说明：获取指定列的字符值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：字符值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public long GetChars(int ordinal, long dataOffset, char[] buffer, int bufferOffset, int length)

  • 接口说明：获取指定列的字符值。
  • 参数说明：
    • ordinal：列索引。
    • dataOffset：数据偏移量。
    • buffer：缓冲区。
    • bufferOffset：缓冲区偏移量。
    • length：长度。
  • 返回值：字符值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public DateTime GetDateTime(int ordinal)

  • 接口说明：获取指定列的日期时间值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：日期时间值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public double GetDouble(int ordinal)

  • 接口说明：获取指定列的双精度浮点数值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：双精度浮点数值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public float GetFloat(int ordinal)

  • 接口说明：获取指定列的单精度浮点数值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：单精度浮点数值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public short GetInt16(int ordinal)

  • 接口说明：获取指定列的 16 位整数值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：16 位整数值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public int GetInt32(int ordinal)

  • 接口说明：获取指定列的 32 位整数值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：32 位整数值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public long GetInt64(int ordinal)

  • 接口说明：获取指定列的 64 位整数值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：64 位整数值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public string GetString(int ordinal)

  • 接口说明：获取指定列的字符串值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：字符串值。
  • 异常：类型不对应抛出 InvalidCastException 异常。

• public object GetValue(int ordinal)

  • 接口说明：获取指定列的值。
  • 参数说明：
    • ordinal：列索引。
  • 返回值： 结果对象。

• public int GetValues(object[] values)

  • 接口说明：获取所有列的值。
  • 参数说明：
    • values：值数组。
  • 返回值：值数量。

• public bool IsDBNull(int ordinal)

  • 接口说明：判断指定列是否为 NULL。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：是否为 NULL。

• public int RecordsAffected

  • 接口说明：获取受影响的行数。
  • 返回值：受影响的行数。

• public bool HasRows

  • 接口说明：结果是否有行数据。
  • 返回值：结果是否有行数据。

• public bool Read()

  • 接口说明：读取下一行。
  • 返回值：是否读取成功。

• public IEnumerator GetEnumerator()

  • 接口说明：获取枚举器。
  • 返回值：枚举器。

• public void Close()

  • 接口说明：关闭结果集。

结果集元数据

DbDataReader 接口提供了以下方法获取结果集元数据：

• public DataTable GetSchemaTable()

  • 接口说明：获取结果集元数据。
  • 返回值：结果集元数据。

• public string GetDataTypeName(int ordinal)

  • 接口说明：获取指定列的数据类型名称。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：数据类型名称。

• public Type GetFieldType(int ordinal)

  • 接口说明：获取指定列的数据类型。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：数据类型。

• public string GetName(int ordinal)

  • 接口说明：获取指定列的名称。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：列名称。

• public int GetFieldSize(int ordinal)

  • 接口说明：获取指定列的大小。
  • 参数说明：
    • ordinal：列索引。
  • 返回值：列大小。

• public int GetOrdinal(string name)

  • 接口说明：获取指定列的索引。
  • 参数说明：
    • name：列名称。
  • 返回值：列索引。

• public int FieldCount

  • 接口说明：获取列数。
  • 返回值：列数。

参数绑定

TDengineCommand 类支持参数绑定。

标准接口

TDengineCommand 类继承了 DbCommand 接口，支持以下功能：

• public string CommandText

  • 接口说明：获取或设置命令文本，支持参数绑定。
  • 返回值：命令文本。

• public new virtual TDengineParameterCollection Parameters

  • 接口说明：获取参数集合。
  • 返回值：TDengineParameterCollection 对象。

参数元数据

TDengineParameterCollection 继承了 DbParameterCollection 接口，支持以下功能：

• public int Add(object value)

  • 接口说明：添加参数。
  • 参数说明：
    • value：参数值。
  • 返回值：参数索引。

• public void Clear()

  • 接口说明：清空参数。

• public bool Contains(object value)

  • 接口说明：是否包含参数。
  • 参数说明：
    • value：参数值。
  • 返回值：是否包含参数。

• public int IndexOf(object value)

  • 接口说明：获取参数索引。
  • 参数说明：
    • value：参数值。
  • 返回值：参数索引。

• public void Insert(int index, object value)

  • 接口说明：插入参数。
  • 参数说明：
    • index：索引。
    • value：参数值。

• public void Remove(object value)

  • 接口说明：移除参数。
  • 参数说明：
    • value：参数值。

• public void RemoveAt(int index)

  • 接口说明：移除参数。
  • 参数说明：
    • index：索引。

• public void RemoveAt(string parameterName)

  • 接口说明：移除参数。
  • 参数说明：
    • parameterName：参数名。

• public int Count

  • 接口说明：获取参数数量。
  • 返回值：参数数量。

• public int IndexOf(string parameterName)

  • 接口说明：获取参数索引。
  • 参数说明：
    • parameterName：参数名。
  • 返回值：参数索引。

• public bool Contains(string value)

  • 接口说明：是否包含参数。
  • 参数说明：
    • value：参数名。
  • 返回值：是否包含参数。

• public void CopyTo(Array array, int index)

  • 接口说明：复制参数。
  • 参数说明：
    • array：目标数组。
    • index：索引。

• public IEnumerator GetEnumerator()

  • 接口说明：获取枚举器。
  • 返回值：枚举器。

• public void AddRange(Array values)

  • 接口说明：添加参数。
  • 参数说明：
    • values：参数数组。

TDengineParameter 继承了 DbParameter 接口，支持以下功能：

• public TDengineParameter(string name, object value)

  • 接口说明：TDengineParameter 构造函数。
  • 参数说明：
    • name：参数名，需要以 @ 开头，如 @0、@1、@2 等。
    • value：参数值，需要 C# 列类型与 TDengine 列类型一一对应。

• public string ParameterName

  • 接口说明：获取或设置参数名。
  • 返回值：参数名。

• public object Value

  • 接口说明：获取或设置参数值。
  • 返回值：参数值。

扩展接口

ITDengineClient 接口提供了扩展的参数绑定接口。

• IStmt StmtInit(long reqId)
  • 接口说明：初始化 statement 对象。
  • 参数说明：
    • reqId：请求 ID。
  • 返回值：实现 IStmt 接口的对象。
  • 异常：执行失败抛出 TDengineError 异常。

IStmt 接口提供了扩展的参数绑定接口。

• void Prepare(string query)

  • 接口说明：准备 statement。
  • 参数说明：
    • query：查询语句。
  • 异常：执行失败抛出 TDengineError 异常。

• bool IsInsert()

  • 接口说明：判断是否为插入语句。
  • 返回值：是否为插入语句。
  • 异常：执行失败抛出 TDengineError 异常。

• void SetTableName(string tableName)

  • 接口说明：设置表名。
  • 参数说明：
    • tableName：表名。
  • 异常：执行失败抛出 TDengineError 异常。

• void SetTags(object[] tags)

  • 接口说明：设置标签。
  • 参数说明：
    • tags：标签数组。
  • 异常：执行失败抛出 TDengineError 异常。

• TaosFieldE[] GetTagFields()

  • 接口说明：获取标签属性。
  • 返回值：标签属性数组。
  • 异常：执行失败抛出 TDengineError 异常。

• TaosFieldE[] GetColFields()

  • 接口说明：获取列属性。
  • 返回值：列属性数组。
  • 异常：执行失败抛出 TDengineError 异常。

• void BindRow(object[] row)

  • 接口说明：绑定行。
  • 参数说明：
    • row：行数据数组。
  • 异常：执行失败抛出 TDengineError 异常。

• void BindColumn( TaosFieldE[] fields,params Array[] arrays)

  • 接口说明：绑定全部列。
  • 参数说明：
    • fields：字段属性数组。
    • arrays：多列数据数组。
  • 异常：执行失败抛出 TDengineError 异常。

• void AddBatch()

  • 接口说明：添加批处理。
  • 异常：执行失败抛出 TDengineError 异常。

• void Exec()

  • 接口说明：执行参数绑定。
  • 异常：执行失败抛出 TDengineError 异常。

• long Affected()

  • 接口说明：获取受影响的行数。
  • 返回值：受影响的行数。
  • 异常：执行失败抛出 TDengineError 异常。

• IRows Result()

  • 接口说明：获取结果。
  • 返回值：结果对象。
  • 异常：执行失败抛出 TDengineError 异常。

数据订阅

ConsumerBuilder 类提供了消费者构建相关接口，ConsumeResult 类提供了消费结果相关接口，TopicPartitionOffset 类提供了分区偏移量相关接口。ReferenceDeserializer 和 DictionaryDeserializer 提供了反序列化的支持。

消费者

• public ConsumerBuilder(IEnumerable<KeyValuePair<string, string>> config)
  • 接口说明：ConsumerBuilder 构造函数。
  • 参数说明：
    • config：消费配置。

创建消费者支持属性列表：

• useSSL：是否使用 SSL 连接，默认为 false。
• token：连接 TDengine cloud 的 token。
• ws.message.enableCompression：是否启用 WebSocket 压缩，默认为 false。
• ws.autoReconnect：是否自动重连，默认为 false。
• ws.reconnect.retry.count：重连次数，默认为 3。
• ws.reconnect.interval.ms：重连间隔毫秒时间，默认为 2000。

其他参数请参考：Consumer 参数列表， 注意TDengine服务端自 3.2.0.0 版本开始消息订阅中的 auto.offset.reset 默认值发生变化。

• public IConsumer<TValue> Build()
  • 接口说明：构建消费者。
  • 返回值：消费者对象。

IConsumer 接口提供了消费者相关 API：

• ConsumeResult<TValue> Consume(int millisecondsTimeout)

  • 接口说明：消费消息。
  • 参数说明：
    • millisecondsTimeout：毫秒超时时间。
  • 返回值：消费结果。
  • 异常：执行失败抛出 TDengineError 异常。

• List<TopicPartition> Assignment { get; }

  • 接口说明：获取分配信息。
  • 返回值：分配信息。
  • 异常：执行失败抛出 TDengineError 异常。

• List<string> Subscription()

  • 接口说明：获取订阅的主题。
  • 返回值：主题列表。
  • 异常：执行失败抛出 TDengineError 异常。

• void Subscribe(IEnumerable<string> topic)

  • 接口说明：订阅主题列表。
  • 参数说明：
    • topic：主题列表。
  • 异常：执行失败抛出 TDengineError 异常。

• void Subscribe(string topic)

  • 接口说明：订阅单个主题。
  • 参数说明：
    • topic：主题。
  • 异常：执行失败抛出 TDengineError 异常。

• void Unsubscribe()

  • 接口说明：取消订阅。
  • 异常：执行失败抛出 TDengineError 异常。

• void Commit(ConsumeResult<TValue> consumerResult)

  • 接口说明：提交消费结果。
  • 参数说明：
    • consumerResult：消费结果。
  • 异常：执行失败抛出 TDengineError 异常。

• List<TopicPartitionOffset> Commit()

  • 接口说明：提交全部消费结果。
  • 返回值：分区偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• void Commit(IEnumerable<TopicPartitionOffset> offsets)

  • 接口说明：提交消费结果。
  • 参数说明：
    • offsets：分区偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• void Seek(TopicPartitionOffset tpo)

  • 接口说明：跳转到分区偏移量。
  • 参数说明：
    • tpo：分区偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• List<TopicPartitionOffset> Committed(TimeSpan timeout)

  • 接口说明：获取分区偏移量。
  • 参数说明：
    • timeout：超时时间(未使用)。
  • 返回值：分区偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• List<TopicPartitionOffset> Committed(IEnumerable<TopicPartition> partitions, TimeSpan timeout)

  • 接口说明：获取指定分区偏移量。
  • 参数说明：
    • partitions：分区列表。
    • timeout：超时时间(未使用)。
  • 返回值：分区偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• Offset Position(TopicPartition partition)

  • 接口说明：获取消费位置。
  • 参数说明：
    • partition：分区。
  • 返回值：偏移量。
  • 异常：执行失败抛出 TDengineError 异常。

• void Close()

  • 接口说明：关闭消费者。

消费记录

ConsumeResult 类提供了消费结果相关接口：

• public List<TmqMessage<TValue>> Message
  • 接口说明：获取消息列表。
  • 返回值：消息列表。

TmqMessage 类提供了消息具体内容：

    public class TmqMessage<TValue>
    {
        public string TableName { get; set; }
        public TValue Value { get; set; }
    }

• TableName：表名
• Value：消息内容

分区信息

从 ConsumeResult 获取 TopicPartitionOffset：

public TopicPartitionOffset TopicPartitionOffset

TopicPartitionOffset 类提供了获取分区信息的接口：

• public string Topic { get; }

  • 接口说明：获取主题。
  • 返回值：主题。

• public Partition Partition { get; }

  • 接口说明：获取分区。
  • 返回值：分区。

• public Offset Offset { get; }

  • 接口说明：获取偏移量。
  • 返回值：偏移量。

• public TopicPartition TopicPartition

  • 接口说明：获取主题分区。
  • 返回值：主题分区。

• public string ToString()

  • 接口说明：转换为字符串。
  • 返回值：字符串信息。

偏移量元数据

Offset 类提供了偏移量相关接口：

• public long Value
  • 接口说明：获取偏移量值。
  • 返回值：偏移量值。

反序列化

C# 驱动提供了两个反序列化类：ReferenceDeserializer 和 DictionaryDeserializer。它们都实现了 IDeserializer 接口。

ReferenceDeserializer 用来将消费到的一条记录反序列化为一个对象，需要保证对象类的属性名与消费到的数据的列名能够对应，且类型能够匹配。

DictionaryDeserializer 则会将消费到的一行数据反序列化为一个 Dictionary<string, object> 对象，其 key 为列名，值为对象。

ReferenceDeserializer 和 DictionaryDeserializer 的接口不会被用户直接调用，请参考使用样例。

附录

更多示例程序。