首页 > IT职业 > IT职场 > 自动化测试应该如何实现多线程?
// 进程 //
进程就是一个程序在一个数据集上的一次动态执行过程,我们编写的程序用来描述进程要完成哪些功能以及如何完成。
// 线程 //
线程页脚轻量级进程,他是一个基本的 CPU执行单元,是进程中的实现,线程的出现是为了降低上下文切换的小号,提高系统的并发性。
// 线程与进程 //
一个线程只能属于一个进程,而一个进程可以有多个线程。
但是少有一个线程资源分配给进程,同一进程的所有线程共享该进程的所有资源 。
CPU分给线程,即真正在CPU上运行的线程。
// 多线程原理应用 //
1 并行与并发
并行:每个线程分配给独立的核心,线程同时运行。
并发:多个线程在单个核心运行,统一时间一个线程运行,系统不停切换线程,看起来像是同时运行,实际上是线程不停切换。
2 Python的多线程
GIL 全局计时器锁:摸个线程想要执行必须先拿到GIL。
Python的多线程:其实同一时间只能运行一个线程,但是能实现并发。
3 Python多线程应用
不同代码运行效率不一样,我们可以通过多线程,形成并发,实现提高效率。
案例: Web自动化,其实 CPU执行完一次命令,大部分时间是在等待,那么这段时间,CPU会限制或者做其他进程的任务,因此我们可以使用多线程,实现多浏览器自动化同时运行,从而实现高效率。
from
appium
import
webdriver
from
appium.webdriver.extensions.android.nativekey
import
AndroidKey
from
appium.webdriver.common.touch_action
import
TouchAction
from
selenium.webdriver.support.wait
import
WebDriverWait
import
time
from
ddt
import
ddt,data,unpack
import
threading
data_sum = [ {
'sum_a'
:
2
,
'sum_b'
:
3
}, {
'sum_a'
:
2
,
'sum_b'
:
3
}, {
'sum_a'
:
5
,
'sum_b'
:
10
}]
def
run_first
(): desired_caps = {
'platformName'
:
'Android'
,
# 被测手机是安卓
'platformVersion'
:
'7.1.2'
,
# 手机安卓版本
'deviceName'
:
'xiaomi_mix'
,
# 设备名,安卓手机可以随意填写
'appPackage'
:
'com.chinatower.fghd.customer'
,
# 启动APP Package名称
'appActivity'
:
'com.ckd.fgbattery.activity.User_Login_Activity'
,
# 启动Activity名称
'unicodeKeyboard'
:
True
,
# 使用自带输入法,输入中文时填True
'resetKeyboard'
:
True
,
# 执行完程序恢复原来输入法
'noReset'
:
True
,
# 不要重置App
'newCommandTimeout'
:
5000
,
'UDID'
:
'127.0.0.1:62001'
,
'automationName'
:
'uiAutomator2'
# 'app': r'd:\apk\bili.apk',
} driver = webdriver.Remote(
'http://localhost:4723/wd/hub'
, desired_caps)
#login_pwd(1,1)
pwd = driver.find_element_by_xpath(
'//*[@text = "密码登录"]'
) print(pwd.get_attribute(
'text'
)) TouchAction(driver).tap(pwd).perform() pwd.click() driver.find_element_by_id(
"user_phone_et"
).send_keys(
'1'
) driver.find_element_by_id(
"sms_code_et"
).send_keys(
'1'
) driver.find_element_by_id(
"btn_res"
).click()
# 登录错误时,无法跳转进入页面
try
: WebDriverWait(driver,
3
,
1
).until(
lambda
x: x.find_element_by_xpath(
"//*[@text='扫码换电']"
)) scan = driver.find_elements_by_xpath(
"//*[@text='扫码换电']"
)
if
scan: print(
"登录成功"
)
return
True
except
: print(
"登录失败"
)
return
False
def
run_two
(): desired_caps_and5 = {
'platformName'
:
'Android'
,
# 被测手机是安卓
'platformVersion'
:
'5.1.1'
,
# 手机安卓版本
'deviceName'
:
'HuaWei P30'
,
# 设备名,安卓手机可以随意填写
'appPackage'
:
'com.chinatower.fghd.customer'
,
# 启动APP Package名称
'appActivity'
:
'com.ckd.fgbattery.activity.User_Login_Activity'
,
# 启动Activity名称
'unicodeKeyboard'
:
True
,
# 使用自带输入法,输入中文时填True
'resetKeyboard'
:
True
,
# 执行完程序恢复原来输入法
'noReset'
:
True
,
# 不要重置App
'newCommandTimeout'
:
5000
,
'UDID'
:
'127.0.0.1:62026'
,
'automationName'
:
'uiAutomator2'
# 'app': r'd:\apk\bili.apk',
} driver2 = webdriver.Remote(
'http://localhost:4725/wd/hub'
, desired_caps_and5)
#login_pwd(2,2)
time.sleep(
10
) print(
"第二个"
) pwd = driver2.find_element_by_xpath(
'//*[@text = "密码登录"]'
) print(pwd.get_attribute(
'text'
)) TouchAction(driver2).tap(pwd).perform() pwd.click() driver2.find_element_by_id(
"user_phone_et"
).send_keys(
'1'
) driver2.find_element_by_id(
"sms_code_et"
).send_keys(
'2'
) driver2.find_element_by_id(
"btn_res"
).click()
# 登录错误时,无法跳转进入页面
try
: WebDriverWait(driver2,
3
,
1
).until(
lambda
x: x.find_element_by_xpath(
"//*[@text='扫码换电']"
)) scan = driver2.find_elements_by_xpath(
"//*[@text='扫码换电']"
)
if
scan: print(
"登录成功"
)
return
True
except
: print(
"登录失败"
)
return
False
# @data(*data_sum)
# @unpack# def login_pwd(sum_a,sum_b):# pwd = driver.find_element_by_xpath('//*[@text = "密码登录"]')# print(pwd.get_attribute('text'))# TouchAction(self.driver).tap(pwd).perform()# pwd.click()# driver.find_element_by_id("user_phone_et").send_keys(sum_a)# driver.find_element_by_id("sms_code_et").send_keys(sum_b)# driver.find_element_by_id("btn_res").click()# # 登录错误时,无法跳转进入页面# try:# WebDriverWait(driver, 3, 1).until(lambda x: x.find_element_by_xpath("//*[@text='扫码换电']"))# scan = driver.find_elements_by_xpath("//*[@text='扫码换电']")# if scan:# print("登录成功")# return True# except:# print("登录失败")# return False
## # 连接Appium Server,初始化自动化环境# driver = webdriver.Remote('http://localhost:4723/wd/hub', desired_caps)# driver_p = webdriver.Remote('http://localhost:4725/wd/hub', desired_caps_and5)## login_pwd(driver,1,1)# login_pwd(driver_p,2,2)
if
__name__ ==
'__main__'
: t1 = threading.Thread(target=run_first,name =
'thread1'
) t2 = threading.Thread(target=run_two,name=
'thread2'
) start_time = time.time() t1.start() t1.join() t2.start() print(
"当前活跃的线程有:{}"
.format(threading.current_thread()))
print(
"正在运行的所有线程:{}"
.format(threading.enumerate()))
print(
"正在运行的线程数量:{}"
.format(threading.active_count()))
t2.join() end_time = time.time() print(
"程序运行耗时:{}"
.format(end_time - start_time))
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/69940641/viewspace-2895149/ ,如需转载,请注明出处,否则将追究法律责任。