简介 接着上一篇继续看一下如何并发测试以及并发测试的过程中，可能遇到的问题，在这里宏哥把宏哥遇到的和小伙伴或者童鞋们，一起分享一下。 Appium端口检测 问题思考 经过前面学习，我们已经能够使用python启动appium服务，但是启动Appium服务之前必须保证对应的端口没有被占用，否则会出现如下报错： error: Couldn't start Appium REST http interface listener. Requested port is already in use. Please make sure there's no other instance of Appium running already. 针对以上这种情况，我们在启动appium服务前该如何检测端口是否可用呢？对于被占用的端口我们又该如何释放？ 需求分析 1.自动检测端口是否被占用 2.如果端口被占用则自动关闭对应端口的进程 端口检测 端口检测需要使用到socket模块来校验端口是否被占用。 python socket模块官方文档 什么是socket？ 网络上的两个程序通过一个双向的通信连接实现数据的交换，这个连接的一端称为一个socket。建立网络通信连接至少要一对端口号(socket)。 socket本质是编程接口(API)，对TCP/IP的封装，TCP/IP也要提供可供程序员做网络开发所用的接口，这就是Socket编程接口；HTTP是轿车，提供了封装或者显示数据的具体形式；Socket是发动机，提供了网络通信的能力。 例如当你用浏览器打开我要博客园主页时，你的浏览器会创建一个socket并命令它去连接博客园的服务器主机，服务器也对客户端的请求创建一个socket进行监听。两端使用各自的socket来发送和接收信息。在socket通信的时候，每个socket都被绑定到一个特定的IP地址和端口。 补充资料： 网络工程师视频教程（自己网上搜一下哈） 代码实现 参考代码 check_port.py 复制代码 1 # coding=utf-8 2 # 1.先设置编码，utf-8可支持中英文，如上，一般放在第一行 3 4 # 2.注释：包括记录创建时间，创建人，项目名称。 5 ''' 6 Created on 2019-9-15 7 @author: 北京-宏哥 QQ交流群：707699217 8 Project:学习和使用appium自动化测试-并发测试 9 ''' 10 # 3.导入模块 11 import socket 12 13 14 def check_port(host, port): 15 """检测指定的端口是否被占用""" 16 17 # 创建socket对象 18 19 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 20 21 try: 22 23 s.connect((host, port)) 24 25 s.shutdown(2) 26 27 except OSError as msg: 28 29 print('port %s is available! ' % port) 30 31 print(msg) 32 33 return True 34 35 else: 36 37 print('port %s already be in use !' % port) 38 39 return False 40 41 42 if __name__ == '__main__': 43 host = '127.0.0.1' 44 45 port = 4723 46 47 check_port(host, port) 复制代码 方法 shutdown(self, flag)：禁止在一个Socket上进行数据的接收与发送。利用shutdown()函数使socket双向数据传输变为单向数据传输。shutdown()需要一个单独的参数， 该参数表示了如何关闭socket 参数 0表示禁止将来读； 1表示禁止将来写 2表示禁止将来读和写。 当端口不可以使用时，运行上边代码，控制台输出如下：此使说明服务端已经开启这个端口服务，所以不可用。 这个端口不可用是由于我用命令行启动这个端口的appium服务 将appium服务关闭后， 端口可以使用时，运行上边代码，控制台输出如下：此使说明服务端没有开启这个端口服务，所以可用。 端口释放 如果端口被占用，则需要释放该端口。那么怎么样去释放被占用的端口呢？ 代码实现 参考代码 check_port.py 复制代码 1 # coding=utf-8 2 # 1.先设置编码，utf-8可支持中英文，如上，一般放在第一行 3 4 # 2.注释：包括记录创建时间，创建人，项目名称。 5 ''' 6 Created on 2019-9-15 7 @author: 北京-宏哥 QQ交流群：707699217 8 Project:学习和使用appium自动化测试-并发测试 9 ''' 10 # 3.导入模块 11 import os 12 13 def release_port(port): 14 """释放指定的端口""" 15 16 # 查找对应端口的pid 17 cmd_find = 'netstat -aon | findstr %s' % port 18 print(cmd_find) 19 # 返回命令执行后的结果 20 result = os.popen(cmd_find).read() 21 print(result) 22 if str(port) and 'LISTENING' in result: 23 24 # 获取端口对应的pid进程 25 i = result.index('LISTENING') 26 start = i + len('LISTENING') + 7 27 end = result.index('\n') 28 pid = result[start:end] 29 # 关闭被占用端口的pid 30 cmd_kill = 'taskkill -f -pid %s' % pid 31 print(cmd_kill) 32 os.popen(cmd_kill) 33 else: 34 35 print('port %s is available !' % port) 36 37 if __name__ == '__main__': 38 host = '127.0.0.1' 39 40 port = 4723 41 42 # check_port(host,port) 43 release_port(port) 复制代码 appium服务端口4723未启动时，控制台显示： appium服务端口4723启动时，控制台显示： Appium并发测试综合实践 测试场景 并发启动2个appium服务，再并发启动2台设备测试考研帮App 2个appium服务，端口配置如下： Appium服务器端口：4723，bp端口为4724 Appium服务器端口：4725，bp端口为4726 2台设备： 设备1：127.0.0.1:62001（夜神模拟器） 设备2：emulator-5554（AVD模拟器） 测试app：考研帮Andriod版 场景分析 其实就是将前面所讲的两部分组合起来，先启动appium服务，再分配设备启动app。 代码实现 参考代码 appium_devices_sync.py 复制代码 1 # coding=utf-8 2 # 1.先设置编码，utf-8可支持中英文，如上，一般放在第一行 3 4 # 2.注释：包括记录创建时间，创建人，项目名称。 5 ''' 6 Created on 2019-9-15 7 @author: 北京-宏哥 QQ交流群：707699217 8 Project:学习和使用appium自动化测试-并发测试 9 ''' 10 # 3.导入模块 11 appium_devices_sync.py 12 13 from appium_sync.multi_appium import appium_start 14 15 from appium_sync.multi_devices import appium_desired 16 17 from appium_sync.check_port import * 18 19 from time import sleep 20 21 import multiprocessing 22 23 devices_list = ['emulator-5554', '127.0.0.1:62001'] 24 25 26 def start_appium_action(host, port): 27 '''检测端口是否被占用，如果没有被占用则启动appium服务''' 28 29 if check_port(host, port): 30 31 appium_start(host, port) 32 33 return True 34 35 else: 36 37 print('appium %s start failed!' % port) 38 39 return False 40 41 42 def start_devices_action(udid, port): 43 '''先检测appium服务是否启动成功，启动成功则再启动App,否则释放端口''' 44 45 host = '127.0.0.1' 46 47 if start_appium_action(host, port): 48 49 appium_desired(udid, port) 50 51 else: 52 53 release_port(port) 54 55 56 def appium_start_sync(): 57 '''并发启动appium服务''' 58 59 print('====appium_start_sync=====') 60 61 # 构建appium进程组 62 63 appium_process = [] 64 65 # 加载appium进程 66 67 for i in range(len(devices_list)): 68 host = '127.0.0.1' 69 70 port = 4723 + 2 * i 71 72 appium = multiprocessing.Process(target=start_appium_action, args=(host, port)) 73 74 appium_process.append(appium) 75 76 # 启动appium服务 77 78 for appium in appium_process: 79 appium.start() 80 81 for appium in appium_process: 82 appium.join() 83 84 sleep(5) 85 86 87 def devices_start_sync(): 88 '''并发启动设备''' 89 90 print('===devices_start_sync===') 91 92 # 定义desired进程组 93 94 desired_process = [] 95 96 # 加载desired进程 97 98 for i in range(len(devices_list)): 99 port = 4723 + 2 * i 100 101 desired = multiprocessing.Process(target=start_devices_action, args=(devices_list[i], port)) 102 103 desired_process.append(desired) 104 105 # 并发启动App 106 107 for desired in desired_process: 108 desired.start() 109 110 for desired in desired_process: 111 desired.join() 112 113 114 if __name__ == '__main__': 115 appium_start_sync() 116 117 devices_start_sync() 复制代码 补充资料：谈谈TCP中的TIME_WAIT 运行代码控制台输出如下日志，这是怎么回事了？？？ 这个是因为宏哥一开始用cmd命令窗口启动了appium，所以会出现下边的样子。 再次运行代码控制台输出如下日志，这又是怎么回事了？？？ 这个是因为第一步启动appium服务已经将端口4723和4725两个端口占用了，第二步appium服务连接设备再次使用的还是同样的端口，所以才会出现如下错误，这个是代码里的bug。宏哥考考你们能不能自己找到修改。 修改bug后，再次运行代码再看一下，如下就正常了，说明你找到bug并已经修改好了。 并发用例执行 测试场景 再上面的场景基础之上，并发启动设备后然后执行跳过引导页面操作。 代码实现 参考代码 kyb_test.py 复制代码 # coding=utf-8 # 1.先设置编码，utf-8可支持中英文，如上，一般放在第一行 # 2.注释：包括记录创建时间，创建人，项目名称。 ''' Created on 2019-9-15 @author: 北京-宏哥 QQ交流群：707699217 Project:学习和使用appium自动化测试-并发测试 ''' # 3.导入模块 from selenium.common.exceptions import NoSuchElementException class KybTest(object): def __init__(self,driver): self.driver=driver def check_cancelBtn(self): print('check cancelBtn') try: cancelBtn = self.driver.find_element_by_id('android:id/button2') except NoSuchElementException: print('no cancelBtn') else: cancelBtn.click() def check_skipBtn(self): print('check skipBtn') try: skipBtn = self.driver.find_element_by_id('com.tal.kaoyan:id/tv_skip') except NoSuchElementException: print('no skipBtn') else: skipBtn.click() def skip_update_guide(self): self.check_cancelBtn() self.check_skipBtn() 复制代码 将执行的用例集成到 multi_devices.py 代码实现 参考代码 multi_devices.py 复制代码 # coding=utf-8 # 1.先设置编码，utf-8可支持中英文，如上，一般放在第一行 # 2.注释：包括记录创建时间，创建人，项目名称。 ''' Created on 2019-9-14 @author: 北京-宏哥 QQ交流群：707699217 Project:学习和使用appium自动化测试-并发测试 ''' # 3.导入模块 from appium import webdriver import yaml from time import ctime from kyb_test import KybTest with open('desired_caps.yaml', 'r')as file: data = yaml.load(file, Loader=yaml.FullLoader) devices_list = ['emulator-5554','127.0.0.1:62001' ] def appium_desired(udid, port): desired_caps = {} desired_caps['platformName'] = data['platformName'] desired_caps['platformVersion'] = data['platformVersion'] desired_caps['deviceName'] = data['deviceName'] desired_caps['udid'] = udid desired_caps['app'] = data['app'] desired_caps['appPackage'] = data['appPackage'] desired_caps['appActivity'] = data['appActivity'] desired_caps['noReset'] = data['noReset'] print('appium port: %s start run %s at %s' % (port, udid, ctime())) driver = webdriver.Remote('http://' + str(data['ip']) + ':' + str(port) + '/wd/hub', desired_caps) driver.implicitly_wait(5) k = KybTest(driver) k.skip_update_guide() return driver if __name__ == '__main__': appium_desired(devices_list[0], 4723) appium_desired(devices_list[1], 4725) 复制代码 基于Docker+STF Appium并发测试（有兴趣的可以了解一下） Docker STF 实践案例：https://github.com/haifengrundadi/DisCartierEJ 小结 这一篇和上一篇合起来是一个微型的demo，有兴趣的童鞋和小伙伴们可以自己完善一下这个demo，最好是应用在实际工作中。 好了并发测试就分享到这里吧！ 个人公众号 微信群 （微信群已满100，可以加宏哥的微信拉你进群，请备注：进群） 您的肯定就是我进步的动力。如果你感觉还不错，就请鼓励一下吧！记得点波 推荐 哦！！！（点击右边的小球即可！(^__^) 嘻嘻……） 作者：北京-宏哥 　　　　 出处：https://www.cnblogs.com/du-hong/ 　　　　 本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。（尊重别人就是尊重自己，尊重笔者的劳动哦，转载请说明出处哦，商用请征得作者本人同意，谢谢！！！）https://www.cnblogs.com/du-hong/p/11489933.html