由于公司使用基于Java语言的
Dubbo技术栈，而本人对Python技术栈更为熟悉。为了使不懂JAVA代码的同学也能进行Dubbo接口层的测试，总结一个通过python实现dubbo接口调用的实现方案。
根据Dubbo官方文档中提到的：dubbo可以通过telnet命令进行服务治理，可以通过telnet链接dubbo服务，再通过invoke方法调用dubbo接口
详情见http://dubbo.apache.org/zh-cn/docs/user/references/telnet.html
而在Python中有一个第三方包 telnetlib，所以我们可以通过这个包来执行telnet命令，进而对dubbo接口进行调用
通过上面官方文档截图，我们可以看到，当我们拿到dubbo服务的IP和端口号，就能去调用指定的dubbo接口了。下面，让我们一步步来实现
调用关系说明 服务容器负责启动，加载，运行服务提供者。 服务提供者在启动时，向注册中心注册自己提供的服务。
服务消费者在启动时，向注册中心订阅自己所需的服务。
注册中心返回服务提供者地址列表给消费者，如果有变更，注册中心将基于长连接推送变更数据给消费者。
服务消费者，从提供者地址列表中，基于软负载均衡算法，选一台提供者进行调用，如果调用失败，再选另一台调用。
服务消费者和提供者，在内存中累计调用次数和调用时间，定时每分钟发送一次统计数据到监控中心。 Dubbo
架构具有以下几个特点，分别是连通性、健壮性、伸缩性、以及向未来架构的升级性。
通过上面架构图我们可以类似 zookeeper 这样的服务注册中心找到对应的服务，所部署的机器和端口
也通过dubbo-monitor上面进行查询
通过上述收到查到到要调用的dubbo接口所处的服务器IP和端口，我们就可以通过python实现dubbo的调用了。详细代码如下：
import re
import telnetlib
import time
import logging
logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
'''
方法调用案例：
conn = InvokeDubboApi('127.0.0.1:88888')
data = {
'dubbo_service': 'xxx.xxx.xx.xxxx.xxxx.xxxx.Service',
'dubbo_method': 'xxxxx',
'parameters': ({"age":41,"name":"tom"},"sh",564645,)
}
invoke = json.loads(conn.invoke_dubbo_api(data))
conn.logout()
'''
class TelnetClient(object):
"""通过telnet连接dubbo服务, 执行shell命令, 可用来调用dubbo接口
"""
def __init__(self, server_host, server_port):
self.conn = telnetlib.Telnet()
self.server_host = server_host
self.server_port = server_port
# telnet登录主机
def connect_dubbo(self):
try:
logging.info("telent连接dubbo服务端: telnet {} {} ……".format(self.server_host, self.server_port))
self.conn.open(self.server_host, port=self.server_port)
return True
except Exception as e:
logging.info('连接失败, 原因是: {}'.format(str(e)))
return False
# 执行传过来的命令，并输出其执行结果
def execute_command(self, command):
# 执行命令
cmd = 'invoke {}\n'.format(command).encode("utf-8")
self.conn.write(cmd)
# 初始化调用次数
invoke_count = 0
# 若调用无返回时，记录次数并重试
result = self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]
while result == '':
time.sleep(1)
result = self.conn.read_very_eager().decode(encoding='utf-8').split('\r\n')[0]
invoke_count += 1
if invoke_count>=5:
logging.info("调用dubbo接口超过五次，调用失败")
return '调用dubbo接口失败'
return result
# 退出telnet
def logout_host(self):
self.conn.write(b"exit\n")
logging.info("登出成功")
class InvokeDubboApi(object):
def __init__(self, content):
#解析dubbo部署的ip和port
try:
dubboaddrre = re.compile(r"([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+)", re.I)
result = dubboaddrre.search(str(content)).group()
server_host = result.split(":")[0]
server_port = result.split(":")[1]
logging.info("获取到dubbo部署信息" + result)
except Exception as e:
raise Exception("获取dubbo部署信息失败：{}".format(e))
try:
self.telnet_client = TelnetClient(server_host, server_port)
self.login_flag = self.telnet_client.connect_dubbo()
except Exception as e:
logging.info("invokedubboapi init error" + e)
#调用dubbo接口
def invoke_dubbo_api(self, data):
cmd = data.get("dubbo_service") + "." + data.get("dubbo_method") + "{}".format(data.get("parameters"))
logging.info("调用命令是：{}".format(cmd))
resp = None
try:
if self.login_flag:
result= self.telnet_client.execute_command(cmd)
logging.info("接口响应是,result={}".format(resp))
return result
else:
logging.info("登陆失败！")
except Exception as e:
raise Exception("调用接口异常, 接口响应是result={}, 异常信息为：{}".format(result, e))
self.logout()
# 调用多个dubbo接口，注：确保所有接口是同一个ip和port
def invoke_dubbo_apis(self,datas):
summary = []
if isinstance(datas,list):
for i in range(len(datas)):
result = self.invoke_dubbo_api(datas[i])
summary.append({"data":datas[i],"result":result})
return summary
else:
return "请确认入参是list"
def logout(self):
self.telnet_client.logout_host()
if __name__ == '__main__':
data = {
'dubbo_service': 'xxx.xxx.xx.xxxx.xxxx.xxxxService',
'dubbo_method': 'xxxxx',
'parameters': ({"id":"123456789","mobile":12456},)
}
i = InvokeDubboApi('127.0.0.1:110741')
i.invoke_dubbo_api(data)
i.logout()
请求结果：
1、请求参数
数据data中的参数字段parameters是一个元组，后面的 ‘,’ 不能少
2、请求参数异常
请求Dubbo接口如果填入的参数有误，会报 no such method 的错误，请检查一下参数是否正常
3、当要批量请求时
传入的参数必须是list，且需要同样的IP和端口。。