简介
接着上一篇继续看一下如何并发测试以及并发测试的过程中，可能遇到的问题，在这里宏哥把宏哥遇到的和小伙伴或者童鞋们，一起分享一下。
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模块来校验端口是否被占用。
什么是socket？
网络上的两个程序通过一个双向的通信连接实现数据的交换，这个连接的一端称为一个socket。建立网络通信连接至少要一对端口号(socket)。
socket本质是编程接口(API)，对TCP/IP的封装，TCP/IP也要提供可供程序员做网络开发所用的接口，这就是Socket编程接口；HTTP是轿车，提供了封装或者显示数据的具体形式；Socket是发动机，提供了网络通信的能力。
例如当你用浏览器打开我要博客园主页时，你的浏览器会创建一个socket并命令它去连接博客园的服务器主机，服务器也对客户端的请求创建一个socket进行监听。两端使用各自的socket来发送和接收信息。在socket通信的时候，每个socket都被绑定到一个特定的IP地址和端口。
补充资料： 网络工程师视频教程
代码实现
参考代码
check_port.py
# coding=utf-
# .先设置编码，utf-8可支持中英文，如上，一般放在第一行
# .注释：包括记录创建时间，创建人，项目名称。
'''
Created on --
@author: 北京-宏哥 QQ交流群：
Project:学习和使用appium自动化测试-并发测试
'''
# .导入模块
import socket
def check_port(host, port):
"""检测指定的端口是否被占用"""
# 创建socket对象
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect((host, port))
s.shutdown()
except OSError as msg:
print('port %s is available! ' % port)
print(msg)
return True
else:
print('port %s already be in use !' % port)
return False
if __name__ == '__main__':
host = '127.0.0.1'
port =
check_port(host, port)
方法
shutdown(self, flag)：禁止在一个Socket上进行数据的接收与发送。利用shutdown()函数使socket双向数据传输变为单向数据传输。shutdown()需要一个单独的参数， 该参数表示了如何关闭socket
参数
- 0表示禁止将来读；
- 1表示禁止将来写
- 2表示禁止将来读和写。
当端口不可以使用时，运行上边代码，控制台输出如下：此使说明服务端已经开启这个端口服务，所以不可用。
这个端口不可用是由于我用命令行启动这个端口的appium服务
将appium服务关闭后，
端口可以使用时，运行上边代码，控制台输出如下：此使说明服务端没有开启这个端口服务，所以可用。
端口释放
如果端口被占用，则需要释放该端口。那么怎么样去释放被占用的端口呢？
代码实现
参考代码
check_port.py
# coding=utf-
# .先设置编码，utf-8可支持中英文，如上，一般放在第一行
# .注释：包括记录创建时间，创建人，项目名称。
'''
Created on --
@author: 北京-宏哥 QQ交流群：
Project:学习和使用appium自动化测试-并发测试
'''
# .导入模块
import os
def release_port(port):
"""释放指定的端口"""
# 查找对应端口的pid
cmd_find = 'netstat -aon | findstr %s' % port
print(cmd_find)
# 返回命令执行后的结果
result = os.popen(cmd_find).read()
print(result)
if str(port) and 'LISTENING' in result:
# 获取端口对应的pid进程
i = result.index('LISTENING')
start = i + len(
end = result.index('\n')
pid = result[start:end]
# 关闭被占用端口的pid
cmd_kill = 'taskkill -f -pid %s' % pid
print(cmd_kill)
os.popen(cmd_kill)
else:
print('port %s is available !' % port)
if __name__ == '__main__':
host = '127.0.0.1'
port =
# check_port(host,port)
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
# coding=utf-
# .先设置编码，utf-8可支持中英文，如上，一般放在第一行
# .注释：包括记录创建时间，创建人，项目名称。
'''
Created on --
@author: 北京-宏哥 QQ交流群：
Project:学习和使用appium自动化测试-并发测试
'''
# .导入模块
appium_devices_sync.py
from appium_sync.multi_appium import appium_start
from appium_sync.multi_devices import appium_desired
from appium_sync.check_port import *
from time import sleep
import multiprocessing
devices_list = ['emulator-5554', '127.0.0.1:62001']
def start_appium_action(host, port):
'''检测端口是否被占用，如果没有被占用则启动appium服务'''
if check_port(host, port):
appium_start(host, port)
return True
else:
print('appium %s start failed!' % port)
return False
def start_devices_action(udid, port):
'''先检测appium服务是否启动成功，启动成功则再启动App,否则释放端口'''
host = '127.0.0.1'
if start_appium_action(host, port):
appium_desired(udid, port)
else:
release_port(port)
def appium_start_sync():
'''并发启动appium服务'''
print('====appium_start_sync=====')
# 构建appium进程组
appium_process = []
# 加载appium进程
for i in range(len(devices_list)):
host = '127.0.0.1'
port = + * i
appium = multiprocessing.Process(target=start_appium_action, args=(host, port))
appium_process.append(appium)
# 启动appium服务
for appium in appium_process:
appium.start()
for appium in appium_process:
appium.join()
sleep()
def devices_start_sync():
'''并发启动设备'''
print('===devices_start_sync===')
# 定义desired进程组
desired_process = []
# 加载desired进程
for i in range(len(devices_list)):
port = + * i
desired = multiprocessing.Process(target=start_devices_action, args=(devices_list[i], port))
desired_process.append(desired)
# 并发启动App
for desired in desired_process:
desired.start()
for desired in desired_process:
desired.join()
if __name__ == '__main__':
appium_start_sync()
devices_start_sync()
补充资料：谈谈TCP中的TIME_WAIT
运行代码控制台输出如下日志，这是怎么回事了？？？
这个是因为宏哥一开始用cmd命令窗口启动了appium，所以会出现下边的样子。
再次运行代码控制台输出如下日志，这又是怎么回事了？？？
这个是因为第一步启动appium服务已经将端口4723和4725两个端口占用了，第二步appium服务连接设备再次使用的还是同样的端口，所以才会出现如下错误，这个是代码里的bug。宏哥考考你们能不能自己找到修改。
修改bug后，再次运行代码再看一下，如下就正常了，说明你找到bug并已经修改好了。
并发用例执行
测试场景
再上面的场景基础之上，并发启动设备后然后执行跳过引导页面操作。
代码实现
参考代码
kyb_test.py
# coding=utf-
# .先设置编码，utf-8可支持中英文，如上，一般放在第一行
# .注释：包括记录创建时间，创建人，项目名称。
'''
Created on --
@author: 北京-宏哥 QQ交流群：
Project:学习和使用appium自动化测试-并发测试
'''
# .导入模块
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-
# .先设置编码，utf-8可支持中英文，如上，一般放在第一行
# .注释：包括记录创建时间，创建人，项目名称。
'''
Created on --
@author: 北京-宏哥 QQ交流群：
Project:学习和使用appium自动化测试-并发测试
'''
# .导入模块
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()
k = KybTest(driver)
k.skip_update_guide()
return driver
if __name__ == '__main__':
appium_desired(devices_list[], )
appium_desired(devices_list[], )
基于Docker+STF Appium并发测试（有兴趣的可以了解一下）
实践案例：https://github.com/haifengrundadi/DisCartierEJ
小结
这一篇和上一篇合起来是一个微型的demo，有兴趣的童鞋和小伙伴们可以自己完善一下这个demo，最好是应用在实际工作中。
好了并发测试就分享到这里吧！
您的肯定就是我进步的动力。如果你感觉还不错，就请鼓励一下吧！记得点波 推荐 哦！！！（点击右边的小球即可！(^__^) 嘻嘻……）
