背景
大部分測試dubbo接口，都是使用Jmeter工具進(jìn)行測試荐开，需要把jar包下載下來付翁，再根據(jù)dubbo接口的傳值方法進(jìn)行測試，對于這一過程及其不方便晃听，有時候會因?yàn)閐ubbo版本與Jmeter插件版本不兼容百侧，導(dǎo)致測試無法進(jìn)行下去，踩坑較多測試效率較低能扒；為此佣渴，本文分享一個dubbo接口測試小工具，大大提高效率初斑！
實(shí)現(xiàn)方案
Python + fastapi web框架 + Telnet庫
方案原理
dubbo提供了telent命令查看服務(wù)辛润，腳本模擬Telnet命令進(jìn)行dubbo接口測試，通過web框架對Telnet命令進(jìn)行包裝见秤，發(fā)送Telnet命令之后砂竖，解析數(shù)據(jù)，通過http接口返回
源碼分析
- Telnet分解（總結(jié)了dubbo接口大致傳參類型鹃答，如遇特殊的乎澄，可聯(lián)系我）
def invoke(self, service_name, method_name, arg):
#自定義對象傳參
if isinstance(arg, dict) and arg:
command_str = "invoke {0}.{1}({2})".format(
service_name, method_name, json.dumps(arg))
#集合對象傳參
elif isinstance(arg, list) and arg:
command_str = "invoke {0}.{1}({2})".format(
service_name, method_name, json.dumps(arg))
#無需對象傳值
elif isinstance(arg, dict) and not arg:
command_str = "invoke {0}.{1}()".format(
service_name, method_name)
#枚舉值類型傳參
else:
command_str = "invoke {0}.{1}({2})".format(
service_name, method_name, arg)
data = self.command(command_str)
try:
# 字節(jié)數(shù)據(jù)解碼 utf8
data = data.decode("utf-8").split('\n')[0].strip()
except BaseException:
# 字節(jié)數(shù)據(jù)解碼 gbk
data = data.decode("gbk").split('\n')[0].strip()
return data
- 模擬command控制臺提交
def command(self, str_=""):
# 模擬cmd控制臺 dubbo>invoke ...
if self.conn :
self.conn.write(str_.encode() + b'\n')
data = self.conn.read_until(self.prompt.encode())
return data
else:
return False
3.支持dubbo接口查詢（根據(jù)IP地址、端口测摔、服務(wù)名查詢到對應(yīng)方法名下的傳參類型）
def ls_invoke(self, service_name):
command_str = "ls -l {0}".format(service_name)
data = self.command(command_str)
if "No such service" in data.decode("utf-8"):
return False
else:
data = data.decode("utf-8").split('\n')
key = ['methodName', 'paramType','type']
dubbo_list = []
for i in range(0, len(data) - 1):
value = []
dubbo_name = data[i].strip().split(' ')[1]
method_name = re.findall(r"(.*?)[(]", dubbo_name)[0]
value.append(method_name)
paramType = re.findall(r"[(](.*?)[)]", dubbo_name)[0]
paramTypeList = paramType.split(',')
if len(paramTypeList) ==1:
paramTypeList = paramTypeList[0]
value.append(paramTypeList)
if 'java.lang' in paramType or 'java.math' in paramType:
value.append(0)
elif not paramType:
value.append(1)
elif 'List' in paramType:
value.append(2)
else:
value.append(3)
dubbo_list.append(dict(zip(key, value)))
return dubbo_list
4.view源碼--dubboList
@router.post('/dubboList', name='dubbo列表接口')
async def dubboList(data: DubboListBody):
host,port = data.url.split(":")
service_name = data.serviceName
method_name = data.methodName
conn = BmDubbo(host, port)
status = conn.command("")
#判斷是否連接成功
if status:
#傳入方法名置济，查詢對應(yīng)方法名的傳值類型
if method_name:
param_data = conn.param_data(service_name, method_name)
#判斷方法是否存在
if param_data:
res_data = {'responseCode': 200, 'responseMsg': "請求成功"}
dubbo_list = {'responseData': param_data}
res_data.update(dubbo_list)
return res_data
#不存在返回報(bào)錯
else:
return {'responseCode': 301, 'responseMsg': "找不到對應(yīng)的serviceName"}
#不傳，直接返回服務(wù)下所有的數(shù)據(jù)
else:
response_data = conn.ls_invoke(service_name)
if response_data:
res_data = {'responseCode': 200, 'responseMsg': "請求成功"}
dubbo_list = {'responseData':response_data}
res_data.update(dubbo_list)
return res_data
else:
return {'responseCode': 301, 'responseMsg': "找不到對應(yīng)的serviceName"}
#連接不成功返回報(bào)錯
else:
return {'responseCode': 302, 'responseMsg': "dubbo服務(wù)連接出錯"}
5.view源碼--dubboInvoke
@router.post('/dubbo', name='dubbo業(yè)務(wù)請求接口')
async def dubboInvoke(data: DubboInvokeBody):
host,port = data.url.split(":")
service_name = data.serviceName
method_name = data.methodName
boby = data.data
conn = BmDubbo(host, port)
status = conn.command("")
if status:
# 根據(jù)服務(wù)名和方法名锋八，返回param方法名和類型
param_data = conn.param_data(service_name, method_name)
if param_data:
type = param_data['type']
param = param_data['paramType']
# 傳參類型為枚舉值方法
if type == 0:
l_data = [v for v in boby.values()]
l_data = str(l_data)
boby = l_data[1:-1]
# 無需傳參
elif type == 1:
boby = boby
# 傳參類型為集合對象
elif type == 2:
for k, v in boby.items():
if isinstance(v, list):
boby = v
break
# 傳參類型為自定義對象
else:
boby.update({"class": param})
response_data = conn.invoke(service_name, method_name, boby)
try:
response_data = json.loads(response_data)
except Exception as e:
res_data = {'responseCode': 207, 'responseMsg': "dubbo接口請求異常"}
res_data.update({'responseData':response_data})
return res_data
return response_data
else:
return {'responseCode': 301, 'responseMsg': "找不到對應(yīng)的serviceName"}
else:
return {'responseCode': 302, 'responseMsg': "dubbo服務(wù)連接出錯"}
使用教程
- pip install requirements.txt
- 右鍵運(yùn)行main文件
- 訪問http://127.0.0.1:5000/api/xxx 即可開始測試dubbo接口
- 具體傳參可看dubbo接口文檔
- 基于fastapi,將dubbo接口轉(zhuǎn)換成便捷的http接口測試
- 優(yōu)點(diǎn)：等你來發(fā)掘
- 缺點(diǎn)/bug：等你來發(fā)掘
dubbo接口文檔
1.查詢服務(wù)名下的所有方法
接口地址
- 說明：根據(jù)dubbo接口地址和dubbo接口服務(wù)名浙于，查詢服務(wù)名下的所有方法
- 地址：
/api/dubboList - 方法：
POST
請求頭
| 序號 | 類型 | 值 | 說明 |
|---|---|---|---|
| 1 | Content-Type | application/json | JSON 格式 |
請求體
| 序號 | 鍵值 | 類型 | 說明 |
|---|---|---|---|
| 1 | url | String | dubbo接口地址，IP:端口 |
| 2 | serviceName | String | 對應(yīng)的服務(wù)名 |
| 3 | methodName | String | 服務(wù)名下對應(yīng)的方法名 |
請求體示例
{
"url": "xxx.xxx.xx.xx:20880",
"serviceName": "cn.com.xxx.mallerp.api.xxxx.xxxx"
}
返回體
| 序號 | 鍵值 | 類型 | 說明 |
|---|---|---|---|
| 1 | responseCode | Int | 返回code |
| 2 | responseMsg | String | 返回信息 |
| 3 | responseData | Array | data數(shù)組 |
| 4 | - type | int | 0-枚舉值挟纱，1-無需傳參<br />2- 集合對象路媚，3-自定義對象 |
| - paramType | string | Java傳值類型 | |
| - methodName | string | 方法名 |
返回值示例（成功）
{
"responseCode": 200,
"responseMsg": "請求成功",
"responseData": [
{
"methodName": "xxxxxx",
"paramType": "java.util.HashMap",
"type": 3
},
{
"methodName": "xxxxxx",
"paramType": [
"java.lang.String",
"java.lang.String",
"java.lang.String",
"java.lang.Integer",
"java.lang.Integer"
],
"type": 0
},
{
"methodName": "xxxxxx",
"paramType": "",
"type": 1
},
{
"methodName": "xxxxxx",
"paramType": "java.util.List",
"type": 2
}
]
}
返回值示例（失敗）
{
"responseCode": 500,
"responseMsg": "相應(yīng)的報(bào)錯信息"
}
2.dubbo接口-業(yè)務(wù)接口
接口地址
- 說明：根據(jù)dubbo接口地址和dubbo接口服務(wù)名樊销，方法名整慎，參數(shù)值實(shí)現(xiàn)dubbo接口邏輯
- 地址：
/api/dubbo - 方法：
POST
請求頭
| 序號 | 類型 | 值 | 說明 |
|---|---|---|---|
| 1 | Content-Type | application/json | JSON 格式 |
請求體
| 序號 | 鍵值 | 類型 | 說明 |
|---|---|---|---|
| 1 | url | string | dubbo接口地址，IP:端口 |
| 2 | serviceName | string | 服務(wù)名 |
| 3 | methodName | string | 方法名 |
| 4 | data | object | 傳值4種情況围苫，具體看示例 |
請求體示例 -- 原生對象或者自定義對象傳參
{
"url": "xxx.xxx.xx.xx:20880",
"serviceName": "cn.com.xxx.mallerp.api.xxxx.xxxx",
"methodName": "xxxxxx",
"data": { //data傳入對應(yīng)的業(yè)務(wù)json數(shù)據(jù)
"productStoreQueryDTOS": [
{
"productNoNumDTOList": [
{
"num": 13,
"productNo": "10000620"
},
{
"num": 13,
"productNo": "10000014"
}
],
"storeCode": "4401S1389"
}
]
}
}
請求體示例 -- 枚舉值類型傳參
{
"url": "xxx.xxx.xx.xx:20880",
"serviceName": "cn.com.xxx.mallerp.api.xxxx.xxxx",
"methodName": "login",
"data": { //格式為json裤园，順序必須按照dubbo接口枚舉值傳參順序，注意是否為int還是string
"account":"80563855",
"password":"3fd6ebe43dab8b6ce6d033a5da6e6ac5"
}
}
請求體示例 -- 方法名無需傳參
{
"url": "xxx.xxx.xx.xx:20880",
"serviceName": "cn.com.xxx.mallerp.api.xxxx.xxxx",
"methodName": "xxxxxx",
"data":{} //傳入空對象
}
請求體示例 --集合對象傳參
{
"url": "xxx.xxx.xx.xx:20880",
"serviceName": "cn.com.xxx.mallerp.api.xxxx.xxxx",
"methodName": "xxxxxx",
"data":{
"empList": [
"30000445",
"30000444"
]
} //傳入對象剂府，里面嵌套數(shù)組
}
返回值示例（成功）只展示其中一種
{
"responseData": "dubbo接口返回什么拧揽，就返回什么"
}
返回值示例（失敗）
{
"responseCode": 500,
"responseMsg": "相應(yīng)的報(bào)錯信息"
}
項(xiàng)目地址：dubbo_fastapi
