1.安装第三方包：
dubbo-python
2.新建2个文件
DubboClientNew.py
NacosMethods.py
#!usr/bin/python
#-*- coding:utf-8 _*-
"""
@author:xiana
@file: NacosMethods.py
@time: 2021/12/31 2:43 下午
"""
import json
from dubbo.client import NacosRegister
from dubbo.common.constants import DUBBO_NC_PROVIDERS
from dubbo.common.exceptions import RegisterException
class NacosMethods(NacosRegister):
def get_methods_from_nacos(self, interface, version):
"""
从nacos中根据interface获取到providers信息,methods信息
:param interface:
:param version:
:return:
"""
service = DUBBO_NC_PROVIDERS.format(interface, version)
providers = self.nc.get_service_list(timeout=self.timeout, group_name=self.group_name,
namespace_id=self.namespace_id)
if not providers or service not in providers:
raise RegisterException('no providers for service {}'.format(service))
self.nc.subscribe([], service_name=service)
services = self.nc.subscribed_local_manager.get_local_instances(service) or {}
self.close()
instance_r = None
for k, v in services.items():
service = v.__dict__
instance = service.get('instance')
instance_r = instance
if not isinstance(instance, list): instance = [instance]
hosts = []
for ins in instance:
host = '{}:{}'.format(ins.get('ip'), ins.get('port'))
hosts.append(host)
self.hosts[interface] = hosts
return json.loads(instance_r['metadata']['methodParamType'])
#!usr/bin/python
#-*- coding:utf-8 _*-
"""
@author:xiana
@file: DubboClientNew.py
@time: 2021/12/31 4:19 下午
"""
from dubbo.client import NacosRegister, DubboClient
from dubbo.codec.encoder import Object
from NacosMethods import NacosMethods
class DubboClientNew:
def __init__(self, host, username, password):
self.host = host
self.username = username
self.password = password
def callDubbo(self, className, methodName, *args, **param):
methods = NacosMethods(self.host, self.username, self.password)
.get_methods_from_nacos(className, ':')
nc = NacosRegister(self.host, self.username, self.password)
methodNames = methods.keys()
obj = ''
for method in methodNames:
if method == methodName:
obj = methods[method][0]
queryRequest = Object(obj)
if args != ():
queryRequest = args
elif param != {}:
for key in param.keys():
queryRequest[key] = param[key]
dubbo_cli = DubboClient(className, nacos_register=nc, version=':')
result = dubbo_cli.call(methodName, queryRequest)
return result
if __name__ == '__main__':
dubboClient = DubboClientNew('XXX.XXX.X.XX:8848', username='username', password='password')
param = {
"qty": 2,
"requestId": "1588011111",
"skuId": 48236429642
}
result = dubboClient.callDubbo('com.xxxx.service.xxxx.xxx.xxx', 'xxxxxxx',
**param)
print(result)
