python通过nacos请求dubbo接口

python通过nacos请求dubbo接口

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)