python telnet invoke 调用dubbo服务

telnet invoke 调用dubbo服务"/>

python telnet invoke 调用dubbo服务

python telnet invoke 调用dubbo服务

import json
import telnetlib# python3 没有试过
class Dubbo(telnetlib.Telnet):prompt = 'dubbo>'coding = 'utf-8'def __init__(self, host=None, port=0):super().__init__(host, port)self.write(b'\n')def command(self, flag, str_=""):data = self.read_until(flag.encode())self.write(str_.encode() + b"\n")return datadef do(self, arg):command_str = argself.command(Dubbo.prompt, command_str)data = self.command(Dubbo.prompt, command_str)data = data.decode(Dubbo.coding, errors='ignore').split('\n')[0].strip()return datalist = ['invoke aa.Service.add(6720126)',
'invoke aa.Service.add(20958353)']	if __name__ == '__main__':conn = Dubbo('127.0.0.1', 18107)for a in list:do = conn.do(a)loads = json.loads(do)print(loads)

import telnetlib
# python2 批量没有成功 批量只成功了前几个命令
def run_telnet(host, port, commands):tn = telnetlib.Telnet(host, port, timeout=10)tn.set_debuglevel(1)for command in commands:print(command)tn.write(command.encode('utf-8') + b'\n')tn.read_until(b'\r\n$')tn.close()
def telnet(host, port, command):tn = telnetlib.Telnet(host, port, timeout=10)tn.set_debuglevel(1)print(command)tn.write(command.encode('utf-8'))tn.read_until(command.encode('utf-8'))tn.close()if __name__ == '__main__':command_list = ['aa.Service.add(24823541)']	for command in command_list:run_telnet( command)