Dubbo接口+python的接口测试举例（用Python自带的telnetlib库进行dubbo测试）
发布时间 2023-07-17 10:00:40作者: aha_baby
1、什么是dubbo
Dubbo来源于阿里巴巴集团某个团队研发出来的一个分布式服务框架，致力于提供高性能和透明化的RPC远程服务调用方案，以及SOA服务治理方案。简单的说，dubbo就是个服务框架，如果没有分布式的需求，其实是不需要用的，只有在分布式的时候，才有dubbo这样的分布式服务框架的需求，并且本质上是个服务调用的东东。
其核心部分包含:
1. 远程通讯: 提供对多种基于长连接的NIO框架抽象封装，包括多种线程模型，序列化，以及“请求-响应”模式的信息交换方式。
2. 集群容错: 提供基于接口方法的透明远程过程调用，包括多协议支持，以及软负载均衡，失败容错，地址路由，动态配置等集群支持。
3. 自动发现: 基于注册中心目录服务，使服务消费方能动态的查找服务提供方，使地址透明，使服务提供方可以平滑增加或减少机器。
Dubbo本身支持多种远程调用方式，例如Dubbo RPC（二进制序列化 + tcp协议）、http invoker（二进制序列化 + http协议）、hessian（二进制序列化 + http协议）、WebServices （文本序列化 + http协议）等。
2、Dubbo能做什么？
1.透明化的远程方法调用，就像调用本地方法一样调用远程方法，只需简单配置，没有任何API侵入。
2.软负载均衡及容错机制，可在内网替代F5等硬件负载均衡器，降低成本，减少单点。
3. 服务自动注册与发现，不再需要写死服务提供方地址，注册中心基于接口名查询服务提供者的IP地址，并且能够平滑添加或删除服务提供者。
3、dubbo的架构
Provider
暴露服务方称之为“服务提供者”。
Consumer
调用远程服务方称之为“服务消费者”。
Registry
服务注册与发现的中心目录服务称之为“服务注册中心”。
Monitor
统计服务的调用次数和调用时间的日志服务称之为“服务监控中心”。
调用关系说明：
0 服务容器负责启动，加载，运行服务提供者。
1. 服务提供者在启动时，向注册中心注册自己提供的服务。
2. 服务消费者在启动时，向注册中心订阅自己所需的服务。
3. 注册中心返回服务提供者地址列表给消费者，如果有变更，注册中心将基于长连接推送变更数据给消费者。
4. 服务消费者，从提供者地址列表中，基于软负载均衡算法，选一台提供者进行调用，如果调用失败，再选另一台调用。
5. 服务消费者和提供者，在内存中累计调用次数和调用时间，定时每分钟发送一次统计数据到监控中心。
以上均是dubbo框架的一些理论知识、那么了解了基础的知识，我们怎么来测试dubbo提供的接口呢。
遇到这个问题我也会咨询一些前辈，有人说java编写代码可以测试，但是需要的依赖多，有些人说基于python也是可以调用的，那么也是需要第三库来支持的。
我们可以去下载对应的python库，有的版本不支持python3 你可以下载下来源码，对源码中的一些代码格式调整到python3支持的格式就好。
我这里用的是用Python自带的telnetlib库，telnetlib是python3自带的库，存放在路径C:\Program Files\Python36\Lib，我们举个例子，看代码是否能够正常运行。
我们拿来了本博客原创的例子。
import json
import socket
import telnetlib
class Dubbo(telnetlib.Telnet):
prompt = 'dubbo>'
coding = 'utf-8'
def __init__(self, host=None, port=0,
timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
super().__init__(host, port)
self.write(b"\n")
def command(self, flag, str_=""):
data = self.read_until(flag.encode())
self.write(str_.encode() + b"\n")
return data
def invoke(self, service_name, method_name, arg):
command_str = "invoke {0}.{1}({2})".format(
service_name, method_name, arg) #这个地方的arg不能写成json.dumps(arg)，即不能转换成string型，提高复用性，在调2个或2个以上的接口的方法的参数时
self.command(Dubbo.prompt, command_str)
data = self.command(Dubbo.prompt, "")
data = data.decode(Dubbo.coding,errors='ignore').split('\n')[0].strip()
return data
def dotest1(self, conn,data): #含有泛型的json，看源码
result = conn.invoke("com.xxx.bluespar.star.service.BudgetService","recBudgetDetail", data)
#print("==========请求data参数为:==========" + str(data))
return result
if __name__ == '__main__':
# count = 10001
# for num in range(20190919200001, 20190919200006, 2):
# count += 100
conn = Dubbo('10.9.48.115', 28899) #这个是dubbo://的IP和端口
# T_E006_BUDGET_DETAIL
data1 = {
"budgetId": "HKYF201908000003",
"businessNo": "BN201908000003",
"sendDate": "2019-08-24 10:42:20",
"bill": "测试001",
"secondBill": "测试001_second",
"transName": "测试transName",
"startCustomerId": "982",
"startCustomerName": "苏州",
"startCityId": "320500",
"startCityName": "苏州",
"startAirportName": "无锡",
"endCustomerId": "522",
"endCustomerName": "金华",
"endCityId": "330700",
"endCityName": "金华",
"endAirportName": "广州",
"flightType": 1,
"flightNo": "GZ568888",
"secFlightNo": "GZ568888_001",
"pickLocationName": "广州",
"goodsTypeName": "广州",
"sendBillCount": 200,
"sendGrossWeight": 8935.66,
"sendCalWeight": 8955.55,
"pickBillCount": 1122.66,
"pickGrossWeight": 898.66,
"pickCalWeight": 55.88,
"computeWeight": 223.33,
"financeReviewWeight": 88.66,
"reviewWeight": 120.33,
"computeModeName": "结算",
"settleModeName": "结算",
"computePrice": 55.2,
"feeKindsName": "费用类型名",
"feeValue": 500.2,
"settleInCustId": "982",
"settleInCustName": "苏州",
"settleOutCustId": "522",
"settleOutCustName": "金华",
"remark": "备注一下"
}
result1 = conn.dotest1(conn, data1) #执行dotest_json_param测试用例（请求是json格式的）
print(result1)
其实，上述代码，就等同于我打开putty或者cmd命令提示符，输入如下命令：
> telnet 10.9.48.115 28899
> invoke com.xxx.bluespar.star.service.BudgetService.recBudgetDetail({"budgetId":"HKYF201908000003","businessNo":"BN201908000003","sendDate":"2019-08-24 10:42:20","bill":"测试001","secondBill":"测试001_second","transName":"测试transName","startCustomerId":"982","startCustomerName":"苏州","startCityId":"320500","startCityName":"苏州","startAirportName":"无锡","endCustomerId":"522","endCustomerName":"金华","endCityId":"330700","endCityName":"金华","endAirportName":"广州","flightType":1,"flightNo":"GZ568888","secFlightNo":"GZ568888_001","pickLocationName":"广州","goodsTypeName":"广州","sendBillCount":200,"sendGrossWeight":8935.66,"sendCalWeight":8955.55,"pickBillCount":1122.66,"pickGrossWeight":898.66,"pickCalWeight":55.88,"computeWeight":223.33,"financeReviewWeight":88.66,"reviewWeight":120.33,"computeModeName":"结算","settleModeName":"结算","computePrice":55.2,"feeKindsName":"费用类型名","feeValue":500.2,"settleInCustId":"982","settleInCustName":"苏州","settleOutCustId":"522","settleOutCustName":"金华","remark":"备注一下"})
Dubbo的接口测试其实也不是那么难，只要我们肯下功夫，我们一定能够做成功的。dubbo接口。开始学起来。
原文链接：https://blog.csdn.net/cyjs1988/article/details/82495136