2023年7月10日发(作者：)

python实现dubbo接⼝的调⽤由于公司使⽤基于Java语⾔的Dubbo技术栈，⽽本⼈对Python技术栈更为熟悉。为了使不懂JAVA代码的同学也能进⾏Dubbo接⼝层的测试，总结⼀个通过python实现dubbo接⼝调⽤的实现⽅案。⼀、实现原理：⽽在Python中有⼀个第三⽅包 telnetlib，所以我们可以通过这个包来执⾏telnet命令，进⽽对dubbo接⼝进⾏调⽤通过上⾯官⽅⽂档截图，我们可以看到，当我们拿到dubbo服务的IP和端⼝号，就能去调⽤指定的dubbo接⼝了。下⾯，让我们⼀步步来实现⼆、dubbo架构：调⽤关系说明 服务容器负责启动，加载，运⾏服务提供者。 服务提供者在启动时，向注册中⼼注册⾃⼰提供的服务。服务消费者在启动时，向注册中⼼订阅⾃⼰所需的服务。注册中⼼返回服务提供者地址列表给消费者，如果有变更，注册中⼼将基于长连接推送变更数据给消费者。服务消费者，从提供者地址列表中，基于软负载均衡算法，选⼀台提供者进⾏调⽤，如果调⽤失败，再选另⼀台调⽤。服务消费者和提供者，在内存中累计调⽤次数和调⽤时间，定时每分钟发送⼀次统计数据到监控中⼼。 Dubbo架构具有以下⼏个特点，分别是连通性、健壮性、伸缩性、以及向未来架构的升级性。通过上⾯架构图我们可以类似 zookeeper 这样的服务注册中⼼找到对应的服务，所部署的机器和端⼝也通过dubbo-monitor上⾯进⾏查询三、python实现dubbo的调⽤通过上述收到查到到要调⽤的dubbo接⼝所处的服务器IP和端⼝，我们就可以通过python实现dubbo的调⽤了。详细代码如下：import reimport telnetlibimport timeimport onfig(level = ,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s')logger = ger(__name__)'''⽅法调⽤案例：conn = InvokeDubboApi('127.0.0.1:88888')data = { 'dubbo_service': 'e', 'dubbo_method': 'xxxxx', 'parameters': ({"age":41,"name":"tom"},"sh",564645,) }invoke = (_dubbo_api(data))()'''class TelnetClient(object): """通过telnet连接dubbo服务, 执⾏shell命令, 可⽤来调⽤dubbo接⼝ """ def __init__(self, server_host, server_port): = () _host = server_host _port = server_port # telnet登录主机 def connect_dubbo(self): try: ("telent连接dubbo服务端: telnet {} {} ……".format(_host, _port)) (_host, port=_port) return True except Exception as e: ('连接失败, 原因是: {}'.format(str(e))) return False # 执⾏传过来的命令，并输出其执⾏结果 def execute_command(self, command): # 执⾏命令 cmd = 'invoke {}n'.format(command).encode("utf-8") (cmd) # 初始化调⽤次数 invoke_count = 0 # 若调⽤⽆返回时，记录次数并重试 result = _very_eager().decode(encoding='utf-8').split('rn')[0] while result == '': (1) result = _very_eager().decode(encoding='utf-8').split('rn')[0] invoke_count += 1 if invoke_count>=5: ("调⽤dubbo接⼝超过五次，调⽤失败") return '调⽤dubbo接⼝失败' return result # 退出telnet def logout_host(self): (b"exitn") ("登出成功")class InvokeDubboApi(object): def __init__(self, content): #解析dubbo部署的ip和port try: dubboaddrre = e(r"([0-9]+.[0-9]+.[0-9]+.[0-9]+:[0-9]+)", re.I) result = (str(content)).group() server_host = (":")[0] server_port = (":")[1] ("获取到dubbo部署信息" + result) except Exception as e: raise Exception("获取dubbo部署信息失败：{}".format(e)) try: _client = TelnetClient(server_host, server_port) _flag = _t_dubbo() except Exception as e: ("invokedubboapi init error" + e) #调⽤dubbo接⼝ def invoke_dubbo_api(self, data): cmd = ("dubbo_service") + "." + ("dubbo_method") + "{}".format(("parameters")) ("调⽤命令是：{}".format(cmd)) resp = None try: if _flag: result= _e_command(cmd) ("接⼝响应是,result={}".format(resp)) return result else: ("登陆失败！") except Exception as e: raise Exception("调⽤接⼝异常, 接⼝响应是result={}, 异常信息为：{}".format(result, e)) () # 调⽤多个dubbo接⼝，注：确保所有接⼝是同⼀个ip和port def invoke_dubbo_apis(self,datas): summary = [] if isinstance(datas,list): for i in range(len(datas)): result = _dubbo_api(datas[i]) ({"data":datas[i],"result":result}) return summary else: return "请确认⼊参是list" def logout(self): __host()if __name__ == '__main__': data = { 'dubbo_service': 'rvice', 'dubbo_method': 'xxxxx', 'parameters': ({"id":"123456789","mobile":12456},) } i = InvokeDubboApi('127.0.0.1:110741') _dubbo_api(data) ()

请求结果：四、注意事项1、请求参数数据data中的参数字段parameters是⼀个元组，后⾯的 ‘,’ 不能少2、请求参数异常请求Dubbo接⼝如果填⼊的参数有误，会报 no such method 的错误，请检查⼀下参数是否正常3、当要批量请求时传⼊的参数必须是list，且需要同样的IP和端⼝。。