Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
d538f9b8
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
d538f9b8
编写于
5月 12, 2022
作者:
H
Hui Li
提交者:
GitHub
5月 12, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #12381 from taosdata/test-v3.0/lihui
test: add test case for tmq
上级
24cc8a80
3309dd71
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
671 addition
and
531 deletion
+671
-531
tests/system-test/7-tmq/basic5.py
tests/system-test/7-tmq/basic5.py
+141
-31
tests/system-test/fulltest.sh
tests/system-test/fulltest.sh
+4
-0
tests/test/c/tmqSim.c
tests/test/c/tmqSim.c
+526
-500
未找到文件。
tests/system-test/7-tmq/basic5.py
浏览文件 @
d538f9b8
...
...
@@ -13,14 +13,12 @@ from util.dnodes import *
class
TDTestCase
:
hostname
=
socket
.
gethostname
()
rpcDebugFlagVal
=
'143'
clientCfgDict
=
{
'serverPort'
:
''
,
'firstEp'
:
''
,
'secondEp'
:
''
,
'rpcDebugFlag'
:
'135'
,
'fqdn'
:
''
}
clientCfgDict
[
"rpcDebugFlag"
]
=
rpcDebugFlagVal
updatecfgDict
=
{
'clientCfg'
:
{},
'serverPort'
:
''
,
'firstEp'
:
''
,
'secondEp'
:
''
,
'rpcDebugFlag'
:
'135'
,
'fqdn'
:
''
}
updatecfgDict
[
"rpcDebugFlag"
]
=
rpcDebugFlagVal
print
(
"===================: "
,
updatecfgDict
)
#rpcDebugFlagVal = '143'
#clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
#updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
#print ("===================: ", updatecfgDict)
def
init
(
self
,
conn
,
logSql
):
tdLog
.
debug
(
f
"start to excute
{
__file__
}
"
)
...
...
@@ -43,27 +41,35 @@ class TDTestCase:
break
return
buildPath
def
create_tables
(
self
,
dbName
,
vgroups
,
stbName
,
ctbNum
,
rowsPerTbl
):
tdSql
.
execute
(
"create database if not exists %s vgroups %d"
%
(
dbName
,
vgroups
))
tdSql
.
execute
(
"use %s"
%
dbName
)
tdSql
.
execute
(
"create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"
%
stbName
)
def
newcur
(
self
,
cfg
,
host
,
port
):
user
=
"root"
password
=
"taosdata"
con
=
taos
.
connect
(
host
=
host
,
user
=
user
,
password
=
password
,
config
=
cfg
,
port
=
port
)
cur
=
con
.
cursor
()
print
(
cur
)
return
cur
def
create_tables
(
self
,
tsql
,
dbName
,
vgroups
,
stbName
,
ctbNum
,
rowsPerTbl
):
tsql
.
execute
(
"create database if not exists %s vgroups %d"
%
(
dbName
,
vgroups
))
tsql
.
execute
(
"use %s"
%
dbName
)
tsql
.
execute
(
"create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"
%
stbName
)
pre_create
=
"create table"
sql
=
pre_create
#tdLog.debug("doing create one stable %s and %d child table in %s ..." %(stbname, count ,dbname))
for
i
in
range
(
ctbNum
):
sql
+=
" %s_%d using %s tags(%d)"
%
(
stbName
,
i
,
stbName
,
i
+
1
)
if
(
i
>
0
)
and
(
i
%
100
==
0
):
t
dS
ql
.
execute
(
sql
)
t
s
ql
.
execute
(
sql
)
sql
=
pre_create
if
sql
!=
pre_create
:
t
dS
ql
.
execute
(
sql
)
t
s
ql
.
execute
(
sql
)
tdLog
.
debug
(
"complete to create database[%s], stable[%s] and %d child tables"
%
(
dbName
,
stbName
,
ctbNum
))
return
def
insert_data
(
self
,
dbName
,
stbName
,
ctbNum
,
rowsPerTbl
,
startTs
):
def
insert_data
(
self
,
tsql
,
dbName
,
stbName
,
ctbNum
,
rowsPerTbl
,
batchNum
,
startTs
):
tdLog
.
debug
(
"start to insert data ............"
)
t
dS
ql
.
execute
(
"use %s"
%
dbName
)
t
s
ql
.
execute
(
"use %s"
%
dbName
)
pre_insert
=
"insert into "
sql
=
pre_insert
...
...
@@ -72,31 +78,38 @@ class TDTestCase:
sql
+=
" %s_%d values "
%
(
stbName
,
i
)
for
j
in
range
(
rowsPerTbl
):
sql
+=
"(%d, %d, 'tmqrow_%d') "
%
(
startTs
+
j
,
j
,
j
)
if
(
j
>
0
)
and
(
j
%
2000
==
0
):
tdSql
.
execute
(
sql
)
sql
=
"insert into %s_%d values "
%
(
stbName
,
i
)
if
(
j
>
0
)
and
((
j
%
batchNum
==
0
)
or
(
j
==
rowsPerTbl
-
1
)):
tsql
.
execute
(
sql
)
if
j
<
rowsPerTbl
-
1
:
sql
=
"insert into %s_%d values "
%
(
stbName
,
i
)
else
:
sql
=
"insert into "
#end sql
if
sql
!=
pre_insert
:
# print(sql)
print
(
"sql:%s"
%
sql
)
tdSql
.
execute
(
sql
)
#print("insert sql:%s"%sql)
tsql
.
execute
(
sql
)
tdLog
.
debug
(
"insert data ............ [OK]"
)
return
def
prepareEnv
(
self
,
**
parameterDict
):
print
(
"input parameters:"
)
print
(
parameterDict
)
self
.
create_tables
(
parameterDict
[
"dbName"
],
\
# create new connector for my thread
tsql
=
self
.
newcur
(
parameterDict
[
'cfg'
],
'localhost'
,
6030
)
self
.
create_tables
(
tsql
,
\
parameterDict
[
"dbName"
],
\
parameterDict
[
"vgroups"
],
\
parameterDict
[
"stbName"
],
\
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
])
self
.
insert_data
(
parameterDict
[
"dbName"
],
\
parameterDict
[
"stbName"
],
\
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
],
\
parameterDict
[
"startTs"
])
self
.
insert_data
(
tsql
,
\
parameterDict
[
"dbName"
],
\
parameterDict
[
"stbName"
],
\
parameterDict
[
"ctbNum"
],
\
parameterDict
[
"rowsPerTbl"
],
\
parameterDict
[
"batchNum"
],
\
parameterDict
[
"startTs"
])
return
def
run
(
self
):
...
...
@@ -113,17 +126,114 @@ class TDTestCase:
tdLog
.
printNoPrefix
(
"======== test scenario 1: "
)
tdLog
.
info
(
"step 1: create database, stb, ctb and insert data"
)
# create and start thread
parameterDict
=
{
'dbName'
:
'db'
,
\
parameterDict
=
{
'cfg'
:
''
,
\
'dbName'
:
'db'
,
\
'vgroups'
:
1
,
\
'stbName'
:
'stb'
,
\
'ctbNum'
:
10
,
\
'rowsPerTbl'
:
10
,
\
'rowsPerTbl'
:
10000
,
\
'batchNum'
:
10
,
\
'startTs'
:
1640966400000
}
# 2022-01-01 00:00:00.000
parameterDict
[
'cfg'
]
=
cfgPath
prepareEnvThread
=
threading
.
Thread
(
target
=
self
.
prepareEnv
,
kwargs
=
parameterDict
)
prepareEnvThread
.
start
()
time
.
sleep
(
2
)
# wait stb ready
while
1
:
#tdSql.query("show %s.stables"%parameterDict['dbName'])
tdSql
.
query
(
"show db.stables"
)
#print (self.queryResult)
#print (tdSql.getRows())
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"create topics from super table"
)
topicFromStb
=
'topic_stb_column'
topicFromCtb
=
'topic_ctb_column'
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s"
%
(
topicFromStb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
tdSql
.
execute
(
"create topic %s as select ts, c1, c2 from %s.%s_0"
%
(
topicFromCtb
,
parameterDict
[
'dbName'
],
parameterDict
[
'stbName'
]))
time
.
sleep
(
1
)
tdSql
.
query
(
"show topics"
)
print
(
"======================================"
)
#print (self.queryResult)
#tdSql.checkRows(2)
topic1
=
tdSql
.
getData
(
0
,
0
)
topic2
=
tdSql
.
getData
(
1
,
0
)
print
(
topic1
)
print
(
topic2
)
print
(
topicFromStb
)
print
(
topicFromCtb
)
#tdLog.info("show topics: %s, %s"%topic1, topic2)
#if topic1 != topicFromStb or topic1 != topicFromCtb:
# tdLog.exit("topic error1")
#if topic2 != topicFromStb or topic2 != topicFromCtb:
# tdLog.exit("topic error2")
tdLog
.
info
(
"create consume info table and consume result table"
)
cdbName
=
parameterDict
[
"dbName"
]
#tdSql.query("create database %s"%cdbName)
#tdSql.query("use %s"%cdbName)
tdSql
.
query
(
"create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)"
)
tdSql
.
query
(
"create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"
)
consumerId
=
0
expectmsgcnt
=
(
parameterDict
[
"rowsPerTbl"
]
/
parameterDict
[
"batchNum"
]
)
*
parameterDict
[
"ctbNum"
]
expectmsgcnt1
=
expectmsgcnt
+
parameterDict
[
"ctbNum"
]
topicList
=
topicFromStb
ifcheckdata
=
0
keyList
=
'group.id:cgrp1,\
enable.auto.commit:false,\
auto.commit.interval.ms:6000,\
auto.offset.reset:earliest'
sql
=
"insert into consumeinfo values "
sql
+=
"(now, %d, '%s', '%s', %d, %d)"
%
(
consumerId
,
topicList
,
keyList
,
expectmsgcnt1
,
ifcheckdata
)
tdSql
.
query
(
sql
)
tdLog
.
info
(
"check stb if there are data"
)
while
1
:
tdSql
.
query
(
"select count(*) from %s"
%
parameterDict
[
"stbName"
])
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
countOfStb
=
tdSql
.
getData
(
0
,
0
)
if
countOfStb
!=
0
:
tdLog
.
info
(
"count from stb: %d"
%
countOfStb
)
break
else
:
time
.
sleep
(
1
)
tdLog
.
info
(
"start consume processor"
)
pollDelay
=
5
showMsg
=
1
showRow
=
1
shellCmd
=
'nohup '
+
buildPath
+
'/build/bin/tmq_sim -c '
+
cfgPath
shellCmd
+=
" -y %d -d %s -g %d -r %d -w %s "
%
(
pollDelay
,
parameterDict
[
"dbName"
],
showMsg
,
showRow
,
cdbName
)
shellCmd
+=
"> /dev/null 2>&1 &"
tdLog
.
info
(
shellCmd
)
os
.
system
(
shellCmd
)
# wait for data ready
prepareEnvThread
.
join
()
prepareEnvThread
.
join
()
tdLog
.
info
(
"insert process end, and start to check consume result"
)
while
1
:
tdSql
.
query
(
"select * from consumeresult"
)
#tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
if
tdSql
.
getRows
()
==
1
:
break
else
:
time
.
sleep
(
5
)
expectrowcnt
=
parameterDict
[
"rowsPerTbl"
]
*
parameterDict
[
"ctbNum"
]
tdSql
.
checkData
(
0
,
1
,
consumerId
)
tdSql
.
checkData
(
0
,
2
,
expectmsgcnt
)
tdSql
.
checkData
(
0
,
3
,
expectrowcnt
)
tdLog
.
printNoPrefix
(
"======== test scenario 2: "
)
...
...
tests/system-test/fulltest.sh
浏览文件 @
d538f9b8
...
...
@@ -51,3 +51,7 @@ python3 ./test.py -f 2-query/arcsin.py
python3 ./test.py
-f
2-query/arccos.py
python3 ./test.py
-f
2-query/arctan.py
# python3 ./test.py -f 2-query/query_cols_tags_and_or.py
python3 ./test.py
-f
7-tmq/basic5.py
tests/test/c/tmqSim.c
浏览文件 @
d538f9b8
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include "taos.h"
#include "taoserror.h"
#include "tlog.h"
#define GREEN "\033[1;32m"
#define NC "\033[0m"
#define min(a, b) (((a) < (b)) ? (a) : (b))
#define MAX_SQL_STR_LEN (1024 * 1024)
#define MAX_ROW_STR_LEN (16 * 1024)
#define MAX_CONSUMER_THREAD_CNT (16)
typedef
struct
{
TdThread
thread
;
int32_t
consumerId
;
int32_t
ifCheckData
;
int64_t
expectMsgCnt
;
int64_t
consumeMsgCnt
;
int64_t
consumeRowCnt
;
int32_t
checkresult
;
char
topicString
[
1024
];
char
keyString
[
1024
];
int32_t
numOfTopic
;
char
topics
[
32
][
64
];
int32_t
numOfKey
;
char
key
[
32
][
64
];
char
value
[
32
][
64
];
tmq_t
*
tmq
;
tmq_list_t
*
topicList
;
}
SThreadInfo
;
typedef
struct
{
// input from argvs
char
cdbName
[
32
];
char
dbName
[
32
];
int32_t
showMsgFlag
;
int32_t
showRowFlag
;
int32_t
consumeDelay
;
// unit s
int32_t
numOfThread
;
SThreadInfo
stThreads
[
MAX_CONSUMER_THREAD_CNT
];
}
SConfInfo
;
static
SConfInfo
g_stConfInfo
;
TdFilePtr
g_fp
=
NULL
;
// char* g_pRowValue = NULL;
// TdFilePtr g_fp = NULL;
static
void
printHelp
()
{
char
indent
[
10
]
=
" "
;
printf
(
"Used to test the tmq feature with sim cases
\n
"
);
printf
(
"%s%s
\n
"
,
indent
,
"-c"
);
printf
(
"%s%s%s%s
\n
"
,
indent
,
indent
,
"Configuration directory, default is "
,
configDir
);
printf
(
"%s%s
\n
"
,
indent
,
"-d"
);
printf
(
"%s%s%s
\n
"
,
indent
,
indent
,
"The name of the database for cosumer, no default "
);
printf
(
"%s%s
\n
"
,
indent
,
"-g"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"showMsgFlag, default is "
,
g_stConfInfo
.
showMsgFlag
);
printf
(
"%s%s
\n
"
,
indent
,
"-r"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"showRowFlag, default is "
,
g_stConfInfo
.
showRowFlag
);
printf
(
"%s%s
\n
"
,
indent
,
"-y"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"consume delay, default is s"
,
g_stConfInfo
.
consumeDelay
);
exit
(
EXIT_SUCCESS
);
}
void
initLogFile
()
{
// FILE *fp = fopen(g_stConfInfo.resultFileName, "a");
char
file
[
256
];
sprintf
(
file
,
"%s/../log/tmqlog.txt"
,
configDir
);
TdFilePtr
pFile
=
taosOpenFile
(
file
,
TD_FILE_TEXT
|
TD_FILE_WRITE
|
TD_FILE_TRUNC
|
TD_FILE_STREAM
);
if
(
NULL
==
pFile
)
{
fprintf
(
stderr
,
"Failed to open %s for save result
\n
"
,
"./tmqlog.txt"
);
exit
(
-
1
);
}
g_fp
=
pFile
;
}
void
saveConfigToLogFile
()
{
time_t
tTime
=
taosGetTimestampSec
();
struct
tm
tm
=
*
taosLocalTime
(
&
tTime
,
NULL
);
taosFprintfFile
(
g_fp
,
"###################################################################
\n
"
);
taosFprintfFile
(
g_fp
,
"# configDir: %s
\n
"
,
configDir
);
taosFprintfFile
(
g_fp
,
"# dbName: %s
\n
"
,
g_stConfInfo
.
dbName
);
taosFprintfFile
(
g_fp
,
"# cdbName: %s
\n
"
,
g_stConfInfo
.
cdbName
);
taosFprintfFile
(
g_fp
,
"# showMsgFlag: %d
\n
"
,
g_stConfInfo
.
showMsgFlag
);
taosFprintfFile
(
g_fp
,
"# showRowFlag: %d
\n
"
,
g_stConfInfo
.
showRowFlag
);
taosFprintfFile
(
g_fp
,
"# consumeDelay: %d
\n
"
,
g_stConfInfo
.
consumeDelay
);
taosFprintfFile
(
g_fp
,
"# numOfThread: %d
\n
"
,
g_stConfInfo
.
numOfThread
);
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
taosFprintfFile
(
g_fp
,
"# consumer %d info:
\n
"
,
g_stConfInfo
.
stThreads
[
i
].
consumerId
);
taosFprintfFile
(
g_fp
,
" Topics: "
);
for
(
int
j
=
0
;
j
<
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
;
j
++
)
{
taosFprintfFile
(
g_fp
,
"%s, "
,
g_stConfInfo
.
stThreads
[
i
].
topics
[
j
]);
}
taosFprintfFile
(
g_fp
,
"
\n
"
);
taosFprintfFile
(
g_fp
,
" Key: "
);
for
(
int
k
=
0
;
k
<
g_stConfInfo
.
stThreads
[
i
].
numOfKey
;
k
++
)
{
taosFprintfFile
(
g_fp
,
"%s:%s, "
,
g_stConfInfo
.
stThreads
[
i
].
key
[
k
],
g_stConfInfo
.
stThreads
[
i
].
value
[
k
]);
}
taosFprintfFile
(
g_fp
,
"
\n
"
);
}
taosFprintfFile
(
g_fp
,
"# Test time: %d-%02d-%02d %02d:%02d:%02d
\n
"
,
tm
.
tm_year
+
1900
,
tm
.
tm_mon
+
1
,
tm
.
tm_mday
,
tm
.
tm_hour
,
tm
.
tm_min
,
tm
.
tm_sec
);
taosFprintfFile
(
g_fp
,
"###################################################################
\n
"
);
}
void
parseArgument
(
int32_t
argc
,
char
*
argv
[])
{
memset
(
&
g_stConfInfo
,
0
,
sizeof
(
SConfInfo
));
g_stConfInfo
.
showMsgFlag
=
0
;
g_stConfInfo
.
showRowFlag
=
0
;
g_stConfInfo
.
consumeDelay
=
5
;
for
(
int32_t
i
=
1
;
i
<
argc
;
i
++
)
{
if
(
strcmp
(
argv
[
i
],
"-h"
)
==
0
||
strcmp
(
argv
[
i
],
"--help"
)
==
0
)
{
printHelp
();
exit
(
0
);
}
else
if
(
strcmp
(
argv
[
i
],
"-d"
)
==
0
)
{
strcpy
(
g_stConfInfo
.
dbName
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-w"
)
==
0
)
{
strcpy
(
g_stConfInfo
.
cdbName
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-c"
)
==
0
)
{
strcpy
(
configDir
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-g"
)
==
0
)
{
g_stConfInfo
.
showMsgFlag
=
atol
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-r"
)
==
0
)
{
g_stConfInfo
.
showRowFlag
=
atol
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-y"
)
==
0
)
{
g_stConfInfo
.
consumeDelay
=
atol
(
argv
[
++
i
]);
}
else
{
printf
(
"%s unknow para: %s %s"
,
GREEN
,
argv
[
++
i
],
NC
);
exit
(
-
1
);
}
}
initLogFile
();
taosFprintfFile
(
g_fp
,
"====parseArgument() success
\n
"
);
#if 1
pPrint
(
"%s configDir:%s %s"
,
GREEN
,
configDir
,
NC
);
pPrint
(
"%s dbName:%s %s"
,
GREEN
,
g_stConfInfo
.
dbName
,
NC
);
pPrint
(
"%s cdbName:%s %s"
,
GREEN
,
g_stConfInfo
.
cdbName
,
NC
);
pPrint
(
"%s consumeDelay:%d %s"
,
GREEN
,
g_stConfInfo
.
consumeDelay
,
NC
);
pPrint
(
"%s showMsgFlag:%d %s"
,
GREEN
,
g_stConfInfo
.
showMsgFlag
,
NC
);
pPrint
(
"%s showRowFlag:%d %s"
,
GREEN
,
g_stConfInfo
.
showRowFlag
,
NC
);
#endif
}
void
splitStr
(
char
**
arr
,
char
*
str
,
const
char
*
del
)
{
char
*
s
=
strtok
(
str
,
del
);
while
(
s
!=
NULL
)
{
*
arr
++
=
s
;
s
=
strtok
(
NULL
,
del
);
}
}
void
ltrim
(
char
*
str
)
{
if
(
str
==
NULL
||
*
str
==
'\0'
)
{
return
;
}
int
len
=
0
;
char
*
p
=
str
;
while
(
*
p
!=
'\0'
&&
isspace
(
*
p
))
{
++
p
;
++
len
;
}
memmove
(
str
,
p
,
strlen
(
str
)
-
len
+
1
);
// return str;
}
static
int
running
=
1
;
static
int32_t
msg_process
(
TAOS_RES
*
msg
,
int64_t
msgIndex
,
int32_t
threadLable
)
{
char
buf
[
1024
];
int32_t
totalRows
=
0
;
// printf("topic: %s\n", tmq_get_topic_name(msg));
// printf("vg:%d\n", tmq_get_vgroup_id(msg));
taosFprintfFile
(
g_fp
,
"msg index:%"
PRId64
", threadLable: %d
\n
"
,
msgIndex
,
threadLable
);
taosFprintfFile
(
g_fp
,
"topic: %s, vgroupId: %d
\n
"
,
tmq_get_topic_name
(
msg
),
tmq_get_vgroup_id
(
msg
));
while
(
1
)
{
TAOS_ROW
row
=
taos_fetch_row
(
msg
);
if
(
row
==
NULL
)
break
;
if
(
0
!=
g_stConfInfo
.
showRowFlag
)
{
TAOS_FIELD
*
fields
=
taos_fetch_fields
(
msg
);
int32_t
numOfFields
=
taos_field_count
(
msg
);
taos_print_row
(
buf
,
row
,
fields
,
numOfFields
);
taosFprintfFile
(
g_fp
,
"rows[%d]: %s
\n
"
,
totalRows
,
buf
);
}
totalRows
++
;
}
return
totalRows
;
}
int
queryDB
(
TAOS
*
taos
,
char
*
command
)
{
TAOS_RES
*
pRes
=
taos_query
(
taos
,
command
);
int
code
=
taos_errno
(
pRes
);
// if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) {
if
(
code
!=
0
)
{
pError
(
"failed to reason:%s, sql: %s"
,
tstrerror
(
code
),
command
);
taos_free_result
(
pRes
);
return
-
1
;
}
taos_free_result
(
pRes
);
return
0
;
}
static
void
tmq_commit_cb_print
(
tmq_t
*
tmq
,
tmq_resp_err_t
resp
,
tmq_topic_vgroup_list_t
*
offsets
,
void
*
param
)
{
printf
(
"tmq_commit_cb_print() commit %d
\n
"
,
resp
);
}
void
build_consumer
(
SThreadInfo
*
pInfo
)
{
tmq_conf_t
*
conf
=
tmq_conf_new
();
// tmq_conf_set(conf, "td.connect.ip", "localhost");
// tmq_conf_set(conf, "td.connect.port", "6030");
tmq_conf_set
(
conf
,
"td.connect.user"
,
"root"
);
tmq_conf_set
(
conf
,
"td.connect.pass"
,
"taosdata"
);
tmq_conf_set
(
conf
,
"td.connect.db"
,
g_stConfInfo
.
dbName
);
tmq_conf_set_offset_commit_cb
(
conf
,
tmq_commit_cb_print
,
NULL
);
// tmq_conf_set(conf, "group.id", "cgrp1");
for
(
int32_t
i
=
0
;
i
<
pInfo
->
numOfKey
;
i
++
)
{
tmq_conf_set
(
conf
,
pInfo
->
key
[
i
],
pInfo
->
value
[
i
]);
}
// tmq_conf_set(conf, "client.id", "c-001");
// tmq_conf_set(conf, "enable.auto.commit", "true");
// tmq_conf_set(conf, "enable.auto.commit", "false");
// tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
// tmq_conf_set(conf, "auto.offset.reset", "none");
// tmq_conf_set(conf, "auto.offset.reset", "earliest");
// tmq_conf_set(conf, "auto.offset.reset", "latest");
pInfo
->
tmq
=
tmq_consumer_new
(
conf
,
NULL
,
0
);
return
;
}
void
build_topic_list
(
SThreadInfo
*
pInfo
)
{
pInfo
->
topicList
=
tmq_list_new
();
// tmq_list_append(topic_list, "test_stb_topic_1");
for
(
int32_t
i
=
0
;
i
<
pInfo
->
numOfTopic
;
i
++
)
{
tmq_list_append
(
pInfo
->
topicList
,
pInfo
->
topics
[
i
]);
}
return
;
}
int32_t
saveConsumeResult
(
SThreadInfo
*
pInfo
)
{
char
sqlStr
[
1024
]
=
{
0
};
TAOS
*
pConn
=
taos_connect
(
NULL
,
"root"
,
"taosdata"
,
NULL
,
0
);
assert
(
pConn
!=
NULL
);
// schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
sprintf
(
sqlStr
,
"insert into %s.consumeresult values (now, %d, %"
PRId64
", %"
PRId64
", %d)"
,
g_stConfInfo
.
cdbName
,
pInfo
->
consumerId
,
pInfo
->
consumeMsgCnt
,
pInfo
->
consumeRowCnt
,
pInfo
->
checkresult
);
TAOS_RES
*
pRes
=
taos_query
(
pConn
,
sqlStr
);
if
(
taos_errno
(
pRes
)
!=
0
)
{
printf
(
"error in save consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taos_free_result
(
pRes
);
exit
(
-
1
);
}
taos_free_result
(
pRes
);
return
0
;
}
void
loop_consume
(
SThreadInfo
*
pInfo
)
{
tmq_resp_err_t
err
;
int64_t
totalMsgs
=
0
;
int64_t
totalRows
=
0
;
while
(
running
)
{
TAOS_RES
*
tmqMsg
=
tmq_consumer_poll
(
pInfo
->
tmq
,
g_stConfInfo
.
consumeDelay
*
1000
);
if
(
tmqMsg
)
{
if
(
0
!=
g_stConfInfo
.
showMsgFlag
)
{
totalRows
+=
msg_process
(
tmqMsg
,
totalMsgs
,
pInfo
->
consumerId
);
}
taos_free_result
(
tmqMsg
);
totalMsgs
++
;
if
(
totalMsgs
>=
pInfo
->
expectMsgCnt
)
{
break
;
}
}
else
{
break
;
}
}
pInfo
->
consumeMsgCnt
=
totalMsgs
;
pInfo
->
consumeRowCnt
=
totalRows
;
taosFprintfFile
(
g_fp
,
"==== consumerId: %d, consumeMsgCnt: %"
PRId64
", consumeRowCnt: %"
PRId64
"
\n
"
,
pInfo
->
consumerId
,
pInfo
->
consumeMsgCnt
,
pInfo
->
consumeRowCnt
);
}
void
*
consumeThreadFunc
(
void
*
param
)
{
int32_t
totalMsgs
=
0
;
SThreadInfo
*
pInfo
=
(
SThreadInfo
*
)
param
;
build_consumer
(
pInfo
);
build_topic_list
(
pInfo
);
if
((
NULL
==
pInfo
->
tmq
)
||
(
NULL
==
pInfo
->
topicList
))
{
return
NULL
;
}
tmq_resp_err_t
err
=
tmq_subscribe
(
pInfo
->
tmq
,
pInfo
->
topicList
);
if
(
err
)
{
printf
(
"tmq_subscribe() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
exit
(
-
1
);
}
loop_consume
(
pInfo
);
tmq_commit
(
pInfo
->
tmq
,
NULL
,
0
);
err
=
tmq_unsubscribe
(
pInfo
->
tmq
);
if
(
err
)
{
printf
(
"tmq_unsubscribe() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
pInfo
->
consumeMsgCnt
=
-
1
;
return
NULL
;
}
err
=
tmq_consumer_close
(
pInfo
->
tmq
);
if
(
err
)
{
printf
(
"tmq_consumer_close() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
exit
(
-
1
);
}
pInfo
->
tmq
=
NULL
;
// save consume result into consumeresult table
saveConsumeResult
(
pInfo
);
return
NULL
;
}
void
parseConsumeInfo
()
{
char
*
token
;
const
char
delim
[
2
]
=
","
;
const
char
ch
=
':'
;
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
token
=
strtok
(
g_stConfInfo
.
stThreads
[
i
].
topicString
,
delim
);
while
(
token
!=
NULL
)
{
// printf("%s\n", token );
strcpy
(
g_stConfInfo
.
stThreads
[
i
].
topics
[
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
],
token
);
ltrim
(
g_stConfInfo
.
stThreads
[
i
].
topics
[
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
]);
// printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]);
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
++
;
token
=
strtok
(
NULL
,
delim
);
}
token
=
strtok
(
g_stConfInfo
.
stThreads
[
i
].
keyString
,
delim
);
while
(
token
!=
NULL
)
{
// printf("%s\n", token );
{
char
*
pstr
=
token
;
ltrim
(
pstr
);
char
*
ret
=
strchr
(
pstr
,
ch
);
memcpy
(
g_stConfInfo
.
stThreads
[
i
].
key
[
g_stConfInfo
.
stThreads
[
i
].
numOfKey
],
pstr
,
ret
-
pstr
);
strcpy
(
g_stConfInfo
.
stThreads
[
i
].
value
[
g_stConfInfo
.
stThreads
[
i
].
numOfKey
],
ret
+
1
);
// printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey],
// g_stConfInfo.value[g_stConfInfo.numOfKey]);
g_stConfInfo
.
stThreads
[
i
].
numOfKey
++
;
}
token
=
strtok
(
NULL
,
delim
);
}
}
}
int32_t
getConsumeInfo
()
{
char
sqlStr
[
1024
]
=
{
0
};
TAOS
*
pConn
=
taos_connect
(
NULL
,
"root"
,
"taosdata"
,
NULL
,
0
);
assert
(
pConn
!=
NULL
);
sprintf
(
sqlStr
,
"select * from %s.consumeinfo"
,
g_stConfInfo
.
cdbName
);
TAOS_RES
*
pRes
=
taos_query
(
pConn
,
sqlStr
);
if
(
taos_errno
(
pRes
)
!=
0
)
{
printf
(
"error in get consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taosFprintfFile
(
g_fp
,
"error in get consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taosCloseFile
(
&
g_fp
);
taos_free_result
(
pRes
);
exit
(
-
1
);
}
TAOS_ROW
row
=
NULL
;
int
num_fields
=
taos_num_fields
(
pRes
);
TAOS_FIELD
*
fields
=
taos_fetch_fields
(
pRes
);
// schema: ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint,
// ifcheckdata int
int32_t
numOfThread
=
0
;
while
((
row
=
taos_fetch_row
(
pRes
)))
{
int32_t
*
lengths
=
taos_fetch_lengths
(
pRes
);
for
(
int
i
=
0
;
i
<
num_fields
;
++
i
)
{
if
(
row
[
i
]
==
NULL
||
0
==
i
)
{
continue
;
}
if
((
1
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_INT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
consumerId
=
*
((
int32_t
*
)
row
[
i
]);
}
else
if
((
2
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
topicString
,
row
[
i
],
lengths
[
i
]);
}
else
if
((
3
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
keyString
,
row
[
i
],
lengths
[
i
]);
}
else
if
((
4
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BIGINT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
expectMsgCnt
=
*
((
int64_t
*
)
row
[
i
]);
}
else
if
((
5
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_INT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
ifCheckData
=
*
((
int32_t
*
)
row
[
i
]);
}
}
numOfThread
++
;
}
g_stConfInfo
.
numOfThread
=
numOfThread
;
taos_free_result
(
pRes
);
parseConsumeInfo
();
return
0
;
}
int
main
(
int32_t
argc
,
char
*
argv
[])
{
parseArgument
(
argc
,
argv
);
getConsumeInfo
();
saveConfigToLogFile
();
TdThreadAttr
thattr
;
taosThreadAttrInit
(
&
thattr
);
taosThreadAttrSetDetachState
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
// pthread_create one thread to consume
taosFprintfFile
(
g_fp
,
"==== create %d consume thread ====
\n
"
,
g_stConfInfo
.
numOfThread
);
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
++
i
)
{
taosThreadCreate
(
&
(
g_stConfInfo
.
stThreads
[
i
].
thread
),
&
thattr
,
consumeThreadFunc
,
(
void
*
)(
&
(
g_stConfInfo
.
stThreads
[
i
])));
}
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
taosThreadJoin
(
g_stConfInfo
.
stThreads
[
i
].
thread
,
NULL
);
}
// printf("consumer: %d, cosumer1: %d\n", totalMsgs, pInfo->consumeMsgCnt);
taosFprintfFile
(
g_fp
,
"==== close tmqlog ====
\n
"
);
taosCloseFile
(
&
g_fp
);
return
0
;
}
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include "taos.h"
#include "taoserror.h"
#include "tlog.h"
#define GREEN "\033[1;32m"
#define NC "\033[0m"
#define min(a, b) (((a) < (b)) ? (a) : (b))
#define MAX_SQL_STR_LEN (1024 * 1024)
#define MAX_ROW_STR_LEN (16 * 1024)
#define MAX_CONSUMER_THREAD_CNT (16)
typedef
struct
{
TdThread
thread
;
int32_t
consumerId
;
int32_t
autoCommitIntervalMs
;
// 1000 ms
char
autoCommit
[
8
];
// true, false
char
autoOffsetRest
[
16
];
// none, earliest, latest
int32_t
ifCheckData
;
int64_t
expectMsgCnt
;
int64_t
consumeMsgCnt
;
int64_t
consumeRowCnt
;
int32_t
checkresult
;
char
topicString
[
1024
];
char
keyString
[
1024
];
int32_t
numOfTopic
;
char
topics
[
32
][
64
];
int32_t
numOfKey
;
char
key
[
32
][
64
];
char
value
[
32
][
64
];
tmq_t
*
tmq
;
tmq_list_t
*
topicList
;
}
SThreadInfo
;
typedef
struct
{
// input from argvs
char
cdbName
[
32
];
char
dbName
[
32
];
int32_t
showMsgFlag
;
int32_t
showRowFlag
;
int32_t
consumeDelay
;
// unit s
int32_t
numOfThread
;
SThreadInfo
stThreads
[
MAX_CONSUMER_THREAD_CNT
];
}
SConfInfo
;
static
SConfInfo
g_stConfInfo
;
TdFilePtr
g_fp
=
NULL
;
// char* g_pRowValue = NULL;
// TdFilePtr g_fp = NULL;
static
void
printHelp
()
{
char
indent
[
10
]
=
" "
;
printf
(
"Used to test the tmq feature with sim cases
\n
"
);
printf
(
"%s%s
\n
"
,
indent
,
"-c"
);
printf
(
"%s%s%s%s
\n
"
,
indent
,
indent
,
"Configuration directory, default is "
,
configDir
);
printf
(
"%s%s
\n
"
,
indent
,
"-d"
);
printf
(
"%s%s%s
\n
"
,
indent
,
indent
,
"The name of the database for cosumer, no default "
);
printf
(
"%s%s
\n
"
,
indent
,
"-g"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"showMsgFlag, default is "
,
g_stConfInfo
.
showMsgFlag
);
printf
(
"%s%s
\n
"
,
indent
,
"-r"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"showRowFlag, default is "
,
g_stConfInfo
.
showRowFlag
);
printf
(
"%s%s
\n
"
,
indent
,
"-y"
);
printf
(
"%s%s%s%d
\n
"
,
indent
,
indent
,
"consume delay, default is s"
,
g_stConfInfo
.
consumeDelay
);
exit
(
EXIT_SUCCESS
);
}
void
initLogFile
()
{
// FILE *fp = fopen(g_stConfInfo.resultFileName, "a");
char
file
[
256
];
sprintf
(
file
,
"%s/../log/tmqlog.txt"
,
configDir
);
TdFilePtr
pFile
=
taosOpenFile
(
file
,
TD_FILE_TEXT
|
TD_FILE_WRITE
|
TD_FILE_TRUNC
|
TD_FILE_STREAM
);
if
(
NULL
==
pFile
)
{
fprintf
(
stderr
,
"Failed to open %s for save result
\n
"
,
"./tmqlog.txt"
);
exit
(
-
1
);
}
g_fp
=
pFile
;
}
void
saveConfigToLogFile
()
{
time_t
tTime
=
taosGetTimestampSec
();
struct
tm
tm
=
*
taosLocalTime
(
&
tTime
,
NULL
);
taosFprintfFile
(
g_fp
,
"###################################################################
\n
"
);
taosFprintfFile
(
g_fp
,
"# configDir: %s
\n
"
,
configDir
);
taosFprintfFile
(
g_fp
,
"# dbName: %s
\n
"
,
g_stConfInfo
.
dbName
);
taosFprintfFile
(
g_fp
,
"# cdbName: %s
\n
"
,
g_stConfInfo
.
cdbName
);
taosFprintfFile
(
g_fp
,
"# showMsgFlag: %d
\n
"
,
g_stConfInfo
.
showMsgFlag
);
taosFprintfFile
(
g_fp
,
"# showRowFlag: %d
\n
"
,
g_stConfInfo
.
showRowFlag
);
taosFprintfFile
(
g_fp
,
"# consumeDelay: %d
\n
"
,
g_stConfInfo
.
consumeDelay
);
taosFprintfFile
(
g_fp
,
"# numOfThread: %d
\n
"
,
g_stConfInfo
.
numOfThread
);
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
taosFprintfFile
(
g_fp
,
"# consumer %d info:
\n
"
,
g_stConfInfo
.
stThreads
[
i
].
consumerId
);
taosFprintfFile
(
g_fp
,
" auto commit: %s
\n
"
,
g_stConfInfo
.
stThreads
[
i
].
autoCommit
);
taosFprintfFile
(
g_fp
,
" auto commit interval ms: %d
\n
"
,
g_stConfInfo
.
stThreads
[
i
].
autoCommitIntervalMs
);
taosFprintfFile
(
g_fp
,
" auto offset rest: %s
\n
"
,
g_stConfInfo
.
stThreads
[
i
].
autoOffsetRest
);
taosFprintfFile
(
g_fp
,
" Topics: "
);
for
(
int
j
=
0
;
j
<
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
;
j
++
)
{
taosFprintfFile
(
g_fp
,
"%s, "
,
g_stConfInfo
.
stThreads
[
i
].
topics
[
j
]);
}
taosFprintfFile
(
g_fp
,
"
\n
"
);
taosFprintfFile
(
g_fp
,
" Key: "
);
for
(
int
k
=
0
;
k
<
g_stConfInfo
.
stThreads
[
i
].
numOfKey
;
k
++
)
{
taosFprintfFile
(
g_fp
,
"%s:%s, "
,
g_stConfInfo
.
stThreads
[
i
].
key
[
k
],
g_stConfInfo
.
stThreads
[
i
].
value
[
k
]);
}
taosFprintfFile
(
g_fp
,
"
\n
"
);
}
taosFprintfFile
(
g_fp
,
"# Test time: %d-%02d-%02d %02d:%02d:%02d
\n
"
,
tm
.
tm_year
+
1900
,
tm
.
tm_mon
+
1
,
tm
.
tm_mday
,
tm
.
tm_hour
,
tm
.
tm_min
,
tm
.
tm_sec
);
taosFprintfFile
(
g_fp
,
"###################################################################
\n
"
);
}
void
parseArgument
(
int32_t
argc
,
char
*
argv
[])
{
memset
(
&
g_stConfInfo
,
0
,
sizeof
(
SConfInfo
));
g_stConfInfo
.
showMsgFlag
=
0
;
g_stConfInfo
.
showRowFlag
=
0
;
g_stConfInfo
.
consumeDelay
=
5
;
for
(
int32_t
i
=
1
;
i
<
argc
;
i
++
)
{
if
(
strcmp
(
argv
[
i
],
"-h"
)
==
0
||
strcmp
(
argv
[
i
],
"--help"
)
==
0
)
{
printHelp
();
exit
(
0
);
}
else
if
(
strcmp
(
argv
[
i
],
"-d"
)
==
0
)
{
strcpy
(
g_stConfInfo
.
dbName
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-w"
)
==
0
)
{
strcpy
(
g_stConfInfo
.
cdbName
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-c"
)
==
0
)
{
strcpy
(
configDir
,
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-g"
)
==
0
)
{
g_stConfInfo
.
showMsgFlag
=
atol
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-r"
)
==
0
)
{
g_stConfInfo
.
showRowFlag
=
atol
(
argv
[
++
i
]);
}
else
if
(
strcmp
(
argv
[
i
],
"-y"
)
==
0
)
{
g_stConfInfo
.
consumeDelay
=
atol
(
argv
[
++
i
]);
}
else
{
printf
(
"%s unknow para: %s %s"
,
GREEN
,
argv
[
++
i
],
NC
);
exit
(
-
1
);
}
}
initLogFile
();
taosFprintfFile
(
g_fp
,
"====parseArgument() success
\n
"
);
#if 1
pPrint
(
"%s configDir:%s %s"
,
GREEN
,
configDir
,
NC
);
pPrint
(
"%s dbName:%s %s"
,
GREEN
,
g_stConfInfo
.
dbName
,
NC
);
pPrint
(
"%s cdbName:%s %s"
,
GREEN
,
g_stConfInfo
.
cdbName
,
NC
);
pPrint
(
"%s consumeDelay:%d %s"
,
GREEN
,
g_stConfInfo
.
consumeDelay
,
NC
);
pPrint
(
"%s showMsgFlag:%d %s"
,
GREEN
,
g_stConfInfo
.
showMsgFlag
,
NC
);
pPrint
(
"%s showRowFlag:%d %s"
,
GREEN
,
g_stConfInfo
.
showRowFlag
,
NC
);
#endif
}
void
splitStr
(
char
**
arr
,
char
*
str
,
const
char
*
del
)
{
char
*
s
=
strtok
(
str
,
del
);
while
(
s
!=
NULL
)
{
*
arr
++
=
s
;
s
=
strtok
(
NULL
,
del
);
}
}
void
ltrim
(
char
*
str
)
{
if
(
str
==
NULL
||
*
str
==
'\0'
)
{
return
;
}
int
len
=
0
;
char
*
p
=
str
;
while
(
*
p
!=
'\0'
&&
isspace
(
*
p
))
{
++
p
;
++
len
;
}
memmove
(
str
,
p
,
strlen
(
str
)
-
len
+
1
);
// return str;
}
static
int
running
=
1
;
static
int32_t
msg_process
(
TAOS_RES
*
msg
,
int64_t
msgIndex
,
int32_t
threadLable
)
{
char
buf
[
1024
];
int32_t
totalRows
=
0
;
// printf("topic: %s\n", tmq_get_topic_name(msg));
// printf("vg:%d\n", tmq_get_vgroup_id(msg));
taosFprintfFile
(
g_fp
,
"msg index:%"
PRId64
", threadLable: %d
\n
"
,
msgIndex
,
threadLable
);
taosFprintfFile
(
g_fp
,
"topic: %s, vgroupId: %d
\n
"
,
tmq_get_topic_name
(
msg
),
tmq_get_vgroup_id
(
msg
));
while
(
1
)
{
TAOS_ROW
row
=
taos_fetch_row
(
msg
);
if
(
row
==
NULL
)
break
;
if
(
0
!=
g_stConfInfo
.
showRowFlag
)
{
TAOS_FIELD
*
fields
=
taos_fetch_fields
(
msg
);
int32_t
numOfFields
=
taos_field_count
(
msg
);
taos_print_row
(
buf
,
row
,
fields
,
numOfFields
);
taosFprintfFile
(
g_fp
,
"rows[%d]: %s
\n
"
,
totalRows
,
buf
);
}
totalRows
++
;
}
return
totalRows
;
}
int
queryDB
(
TAOS
*
taos
,
char
*
command
)
{
TAOS_RES
*
pRes
=
taos_query
(
taos
,
command
);
int
code
=
taos_errno
(
pRes
);
// if ((code != 0) && (code != TSDB_CODE_RPC_AUTH_REQUIRED)) {
if
(
code
!=
0
)
{
pError
(
"failed to reason:%s, sql: %s"
,
tstrerror
(
code
),
command
);
taos_free_result
(
pRes
);
return
-
1
;
}
taos_free_result
(
pRes
);
return
0
;
}
static
void
tmq_commit_cb_print
(
tmq_t
*
tmq
,
tmq_resp_err_t
resp
,
tmq_topic_vgroup_list_t
*
offsets
,
void
*
param
)
{
printf
(
"tmq_commit_cb_print() commit %d
\n
"
,
resp
);
}
void
build_consumer
(
SThreadInfo
*
pInfo
)
{
tmq_conf_t
*
conf
=
tmq_conf_new
();
// tmq_conf_set(conf, "td.connect.ip", "localhost");
// tmq_conf_set(conf, "td.connect.port", "6030");
tmq_conf_set
(
conf
,
"td.connect.user"
,
"root"
);
tmq_conf_set
(
conf
,
"td.connect.pass"
,
"taosdata"
);
//tmq_conf_set(conf, "td.connect.db", g_stConfInfo.dbName);
tmq_conf_set_offset_commit_cb
(
conf
,
tmq_commit_cb_print
,
NULL
);
// tmq_conf_set(conf, "group.id", "cgrp1");
for
(
int32_t
i
=
0
;
i
<
pInfo
->
numOfKey
;
i
++
)
{
tmq_conf_set
(
conf
,
pInfo
->
key
[
i
],
pInfo
->
value
[
i
]);
}
// tmq_conf_set(conf, "client.id", "c-001");
// tmq_conf_set(conf, "enable.auto.commit", "true");
// tmq_conf_set(conf, "enable.auto.commit", "false");
// tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
// tmq_conf_set(conf, "auto.offset.reset", "none");
// tmq_conf_set(conf, "auto.offset.reset", "earliest");
// tmq_conf_set(conf, "auto.offset.reset", "latest");
pInfo
->
tmq
=
tmq_consumer_new
(
conf
,
NULL
,
0
);
tmq_conf_destroy
(
conf
);
return
;
}
void
build_topic_list
(
SThreadInfo
*
pInfo
)
{
pInfo
->
topicList
=
tmq_list_new
();
// tmq_list_append(topic_list, "test_stb_topic_1");
for
(
int32_t
i
=
0
;
i
<
pInfo
->
numOfTopic
;
i
++
)
{
tmq_list_append
(
pInfo
->
topicList
,
pInfo
->
topics
[
i
]);
}
return
;
}
int32_t
saveConsumeResult
(
SThreadInfo
*
pInfo
)
{
char
sqlStr
[
1024
]
=
{
0
};
TAOS
*
pConn
=
taos_connect
(
NULL
,
"root"
,
"taosdata"
,
NULL
,
0
);
assert
(
pConn
!=
NULL
);
// schema: ts timestamp, consumerid int, consummsgcnt bigint, checkresult int
sprintf
(
sqlStr
,
"insert into %s.consumeresult values (now, %d, %"
PRId64
", %"
PRId64
", %d)"
,
g_stConfInfo
.
cdbName
,
pInfo
->
consumerId
,
pInfo
->
consumeMsgCnt
,
pInfo
->
consumeRowCnt
,
pInfo
->
checkresult
);
TAOS_RES
*
pRes
=
taos_query
(
pConn
,
sqlStr
);
if
(
taos_errno
(
pRes
)
!=
0
)
{
printf
(
"error in save consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taos_free_result
(
pRes
);
exit
(
-
1
);
}
taos_free_result
(
pRes
);
return
0
;
}
void
loop_consume
(
SThreadInfo
*
pInfo
)
{
tmq_resp_err_t
err
;
int64_t
totalMsgs
=
0
;
int64_t
totalRows
=
0
;
while
(
running
)
{
TAOS_RES
*
tmqMsg
=
tmq_consumer_poll
(
pInfo
->
tmq
,
g_stConfInfo
.
consumeDelay
*
1000
);
if
(
tmqMsg
)
{
if
(
0
!=
g_stConfInfo
.
showMsgFlag
)
{
totalRows
+=
msg_process
(
tmqMsg
,
totalMsgs
,
pInfo
->
consumerId
);
}
taos_free_result
(
tmqMsg
);
totalMsgs
++
;
if
(
totalMsgs
>=
pInfo
->
expectMsgCnt
)
{
taosFprintfFile
(
g_fp
,
"==== totalMsgs >= pInfo->expectMsgCnt, so break
\n
"
);
break
;
}
}
else
{
taosFprintfFile
(
g_fp
,
"==== delay over time, so break
\n
"
);
break
;
}
}
pInfo
->
consumeMsgCnt
=
totalMsgs
;
pInfo
->
consumeRowCnt
=
totalRows
;
taosFprintfFile
(
g_fp
,
"==== consumerId: %d, consumeMsgCnt: %"
PRId64
", consumeRowCnt: %"
PRId64
"
\n
"
,
pInfo
->
consumerId
,
pInfo
->
consumeMsgCnt
,
pInfo
->
consumeRowCnt
);
}
void
*
consumeThreadFunc
(
void
*
param
)
{
int32_t
totalMsgs
=
0
;
SThreadInfo
*
pInfo
=
(
SThreadInfo
*
)
param
;
build_consumer
(
pInfo
);
build_topic_list
(
pInfo
);
if
((
NULL
==
pInfo
->
tmq
)
||
(
NULL
==
pInfo
->
topicList
))
{
return
NULL
;
}
tmq_resp_err_t
err
=
tmq_subscribe
(
pInfo
->
tmq
,
pInfo
->
topicList
);
if
(
err
)
{
printf
(
"tmq_subscribe() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
exit
(
-
1
);
}
tmq_list_destroy
(
pInfo
->
topicList
);
pInfo
->
topicList
=
NULL
;
loop_consume
(
pInfo
);
tmq_commit
(
pInfo
->
tmq
,
NULL
,
0
);
err
=
tmq_unsubscribe
(
pInfo
->
tmq
);
if
(
err
)
{
printf
(
"tmq_unsubscribe() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
pInfo
->
consumeMsgCnt
=
-
1
;
return
NULL
;
}
err
=
tmq_consumer_close
(
pInfo
->
tmq
);
if
(
err
)
{
printf
(
"tmq_consumer_close() fail, reason: %s
\n
"
,
tmq_err2str
(
err
));
exit
(
-
1
);
}
pInfo
->
tmq
=
NULL
;
// save consume result into consumeresult table
saveConsumeResult
(
pInfo
);
return
NULL
;
}
void
parseConsumeInfo
()
{
char
*
token
;
const
char
delim
[
2
]
=
","
;
const
char
ch
=
':'
;
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
token
=
strtok
(
g_stConfInfo
.
stThreads
[
i
].
topicString
,
delim
);
while
(
token
!=
NULL
)
{
// printf("%s\n", token );
strcpy
(
g_stConfInfo
.
stThreads
[
i
].
topics
[
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
],
token
);
ltrim
(
g_stConfInfo
.
stThreads
[
i
].
topics
[
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
]);
// printf("%s\n", g_stConfInfo.topics[g_stConfInfo.numOfTopic]);
g_stConfInfo
.
stThreads
[
i
].
numOfTopic
++
;
token
=
strtok
(
NULL
,
delim
);
}
token
=
strtok
(
g_stConfInfo
.
stThreads
[
i
].
keyString
,
delim
);
while
(
token
!=
NULL
)
{
// printf("%s\n", token );
{
char
*
pstr
=
token
;
ltrim
(
pstr
);
char
*
ret
=
strchr
(
pstr
,
ch
);
memcpy
(
g_stConfInfo
.
stThreads
[
i
].
key
[
g_stConfInfo
.
stThreads
[
i
].
numOfKey
],
pstr
,
ret
-
pstr
);
strcpy
(
g_stConfInfo
.
stThreads
[
i
].
value
[
g_stConfInfo
.
stThreads
[
i
].
numOfKey
],
ret
+
1
);
// printf("key: %s, value: %s\n", g_stConfInfo.key[g_stConfInfo.numOfKey],
// g_stConfInfo.value[g_stConfInfo.numOfKey]);
g_stConfInfo
.
stThreads
[
i
].
numOfKey
++
;
}
token
=
strtok
(
NULL
,
delim
);
}
}
}
int32_t
getConsumeInfo
()
{
char
sqlStr
[
1024
]
=
{
0
};
TAOS
*
pConn
=
taos_connect
(
NULL
,
"root"
,
"taosdata"
,
NULL
,
0
);
assert
(
pConn
!=
NULL
);
sprintf
(
sqlStr
,
"select * from %s.consumeinfo"
,
g_stConfInfo
.
cdbName
);
TAOS_RES
*
pRes
=
taos_query
(
pConn
,
sqlStr
);
if
(
taos_errno
(
pRes
)
!=
0
)
{
printf
(
"error in get consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taosFprintfFile
(
g_fp
,
"error in get consumeinfo, reason:%s
\n
"
,
taos_errstr
(
pRes
));
taosCloseFile
(
&
g_fp
);
taos_free_result
(
pRes
);
exit
(
-
1
);
}
TAOS_ROW
row
=
NULL
;
int
num_fields
=
taos_num_fields
(
pRes
);
TAOS_FIELD
*
fields
=
taos_fetch_fields
(
pRes
);
// schema: ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint,
// ifcheckdata int
int32_t
numOfThread
=
0
;
while
((
row
=
taos_fetch_row
(
pRes
)))
{
int32_t
*
lengths
=
taos_fetch_lengths
(
pRes
);
// set default value
g_stConfInfo
.
stThreads
[
numOfThread
].
autoCommitIntervalMs
=
5000
;
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
autoCommit
,
"true"
,
strlen
(
"true"
));
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
autoOffsetRest
,
"earlieast"
,
strlen
(
"earlieast"
));
for
(
int
i
=
0
;
i
<
num_fields
;
++
i
)
{
if
(
row
[
i
]
==
NULL
||
0
==
i
)
{
continue
;
}
if
((
1
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_INT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
consumerId
=
*
((
int32_t
*
)
row
[
i
]);
}
else
if
((
2
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
topicString
,
row
[
i
],
lengths
[
i
]);
}
else
if
((
3
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
keyString
,
row
[
i
],
lengths
[
i
]);
}
else
if
((
4
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BIGINT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
expectMsgCnt
=
*
((
int64_t
*
)
row
[
i
]);
}
else
if
((
5
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_INT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
ifCheckData
=
*
((
int32_t
*
)
row
[
i
]);
}
else
if
((
6
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
autoCommit
,
row
[
i
],
lengths
[
i
]);
}
else
if
((
7
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_INT
))
{
g_stConfInfo
.
stThreads
[
numOfThread
].
autoCommitIntervalMs
=
*
((
int32_t
*
)
row
[
i
]);
}
else
if
((
8
==
i
)
&&
(
fields
[
i
].
type
==
TSDB_DATA_TYPE_BINARY
))
{
memcpy
(
g_stConfInfo
.
stThreads
[
numOfThread
].
autoOffsetRest
,
row
[
i
],
lengths
[
i
]);
}
}
numOfThread
++
;
}
g_stConfInfo
.
numOfThread
=
numOfThread
;
taos_free_result
(
pRes
);
parseConsumeInfo
();
return
0
;
}
int
main
(
int32_t
argc
,
char
*
argv
[])
{
parseArgument
(
argc
,
argv
);
getConsumeInfo
();
saveConfigToLogFile
();
TdThreadAttr
thattr
;
taosThreadAttrInit
(
&
thattr
);
taosThreadAttrSetDetachState
(
&
thattr
,
PTHREAD_CREATE_JOINABLE
);
// pthread_create one thread to consume
taosFprintfFile
(
g_fp
,
"==== create %d consume thread ====
\n
"
,
g_stConfInfo
.
numOfThread
);
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
++
i
)
{
taosThreadCreate
(
&
(
g_stConfInfo
.
stThreads
[
i
].
thread
),
&
thattr
,
consumeThreadFunc
,
(
void
*
)(
&
(
g_stConfInfo
.
stThreads
[
i
])));
}
for
(
int32_t
i
=
0
;
i
<
g_stConfInfo
.
numOfThread
;
i
++
)
{
taosThreadJoin
(
g_stConfInfo
.
stThreads
[
i
].
thread
,
NULL
);
}
// printf("consumer: %d, cosumer1: %d\n", totalMsgs, pInfo->consumeMsgCnt);
taosFprintfFile
(
g_fp
,
"==== close tmqlog ====
\n
"
);
taosCloseFile
(
&
g_fp
);
return
0
;
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录