最近工作中经常与各类dubbo服务打交道,而在测试框架选型的时候考虑到开发效率和团队背景,选择了Python作为基础语言,这样就带来一个问题,如何用Python来测试dubbo服务?以及如何用Python来模拟dubbo服务被其它dubbo消费者调用?

dubbo默认采用hessian2的编解码方式,没有找到现成的库,而转换为其它编码方式如jsonrpc等,在有些场景下也不可行,于是就决定自己撸一个,也就有了dubbo-py这个项目。

项目进展

当前这个项目还处于部分可用状态,但已经能满足我当前的需求了,如果您遇到问题,欢迎在github上给我提issue

主要功能的实现情况:

  • 大部分hessian2协议的编解码(我用到的那些吧)
  • Python DubboClient调用Java的Dubbo服务
  • Python DubboService可以注册到zookeeper上作为Dubbo Provider
  • 代码重构,让Python的rpc函数只需要关心调用参数与返回值(和普通函数一样),其它都交给library实现
  • Python DubboClient注册到zookeeper上作为Dubbo Consumer
  • 添加AsyncDubboService以支持asyncio
  • 更多地单测覆盖
  • 支持Python2.7

示例

from dubbo.codec.hessian2 import DubboResponse, JavaList
from dubbo.server import DubboService
from dubbo.client import DubboClient


def remote_max(*args):
    return max(args)


def remote_sum(lst):
    return sum(lst)


service = DubboService(12358, 'demo')
service.add_method('com.myservice.math', 'max', remote_max)
service.add_method('com.myservice.math', 'sum', remote_sum)
# service.register('127.0.0.1:2181', '1.0.0')  # register to zookeeper
service.start()  # service run in a daemon thread

client = DubboClient('127.0.0.1', 12358)
resp = client.send_request_and_return_response(service_name='com.myservice.math', method_name='max', args=[1, 2, 3, 4])
print(resp.data)  # 4
resp2 = client.send_request_and_return_response(service_name='com.myservice.math', method_name='sum', args=[JavaList([1, 2, 3, 4])])
print(resp2.data)  # 10

代码结构

dubbo-py的代码很简单,目前也没来得及好好整理,它要求Python3.6以上环境(对Python3.5和Python2.7的支持在计划中),依赖kazoo,借助它实现与zookeeper的交互。

  • dubbo
    • server 实现了DubboService类,用于模拟Dubbo Provider,感觉后续名字改成provider更合适
    • client 实现了DubboClient类,用于模拟Dubbo Consumer,感觉后续名字改成consumer更好
    • codec 协议编解码模块,目前只实现了hessian2协议的编解码,好像也只需要这个就够了
  • tst 单元测试,采用pytest框架