appium2.X之python启动appium服务

python实现多进程并发启动appium服务，对端口检测和释放

废柴码农进阶传

1087人浏览 · 2024-07-18 16:22:50

废柴码农进阶传 · 2024-07-18 16:22:50 发布

一、使用python启动多个appium服务

1）前提条件：配置正确环境变量

将appium安装路径和node安装路径添加到系统环境变量path下

备注：appium安装路径使用命令where appium查询

2）启动appium服务端口4723和4725

python需要执行命令：

cmd = f"start /b appium -a {host} -p {port}"

备注：/b 表示在后台运行程序，不阻塞当前命令窗口，也就是隐藏了cmd命令窗口

-a 指定服务器IP地址，-p 指定服务器端口

-bp 引导端口BootstrapPort用于appium和移动设备之间通信

 bootstrap_port = str(port+1)
 cmd = 'start /b appium -a '+host+' -p '+str(port)+' -bp ' + str(bootstrap_port)

注：此处加入-bp会报错[error] unrecognized arguments: -bp 4724所以直接把-bp去掉就成功启动appium服务

2）实现代码multi_appium.py

import subprocess
from time import ctime


def appium_start(host, port):
    """启动appium server"""
    cmd = f"start /b appium -a {host} -p {port}"
    print("%s at %s" % (cmd, ctime()))
    appium_log = "../log/"+str(port)+".log"
    with open(appium_log, "a", encoding='utf-8') as log_file:
        subprocess.Popen(cmd, shell=True, stdout=log_file, stderr=subprocess.STDOUT, encoding='utf-8')


if __name__ == "__main__":
    host = "127.0.0.1"
    for i in range(2):
       port = 4723 + 2 * i
       appium_start(host, port)

3）验证启动appium服务成功

3.1）根据生成的log文件，查看log日志内容

3.2）使用命令查询：netstat -ano |findstr 端口号

4）关闭appium服务

4.1）通过netstat找到的appium进程pid在系统任务管理器去关闭

4.2）使用命令关闭：taskkill -f -pid appium进程号

二、多进程并发启动appium服务

1）实现代码multi_appium_sync.py

# -*- coding: utf-8 -*-
import subprocess
import multiprocessing
from time import ctime


def appium_start(host, port):
    """启动appium server"""
    cmd = f"start /b appium -a {host} -p {port}"
    print("%s at %s" % (cmd, ctime()))
    appium_log = "../log/"+str(port)+".log"
    with open(appium_log, "a", encoding='utf-8') as log_file:
        subprocess.Popen(cmd, shell=True, stdout=log_file, stderr=subprocess.STDOUT, encoding='utf-8')

# 构建appium进程组
appium_process = []

# 加载appium进程
for i in range(2):
    host = "127.0.0.1"
    port = 4723 + 2 * i
    appium = multiprocessing.Process(target=appium_start, args=(host, port))
    appium_process.append(appium)


if __name__ == "__main__":
    # 并发启动appium服务
    for appium in appium_process:
        appium.start()
    for appium in appium_process:
        appium.join()

2）验证启动appium服务成功

三、appium端口检测

1）appium端口检测原因：启动appium之前必须保证对应的端口没有被占用，否则会出现如下报错

2）使用socket模块来校验端口是否被占用

2.1）自动检测端口受否被占用，如果端口被占用则自动关闭对应端口的进程

方法：shutdown(self,flag) 禁止在一个socket上进行数据的接收与发送。利用shutdown() 函数使socket双向数据传输变为单项数据传输。shutdown() 函数需要一个单独的参数，该参数表示了如何关闭socket

参数：

0 表示禁止将来读

1 表示禁止将来写

2 表示禁止将来读和写

3）实现代码check_port.py

# -*- coding: utf-8 -*-
import socket


def check_port(host, port):
    """检测指定的端口是否被占用"""

    # 创建socket对象
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        s.connect((host, port))  # 尝试连接指定的（host ,port）
        s.shutdown(2)  # 关闭连接
    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 = 4723
    check_port(host, port)

3.1）当端口可以使用时，控制台会输出如下：词输出说明服务端没有开启端口服务，所以可以使用。

3.2）如下，说明端口已被占用

四、端口释放

1）查询对应端口号，获取端口号对应pid进程，taskkill关闭占用的端口，实现代码release.py

# -*- coding: utf-8 -*-
import os


def release(port):
    """释放指定的端口"""

    # 查找对应端口的pid
    cmd_find_pid= "netstat -ano | findstr %s" % port
    print(cmd_find_pid)

    # 返回命令执行后的结果
    result = os.popen(cmd_find_pid).read()
    print(result)

    # 在返回结果中判断是否存在端口号和字符串 LISTENING
    if str(port) and "LISTENING" in result:
        # 获取端口号对应pid进程
        i = result.index("LISTENING")  # 序号从字符串 LISTENING第一个字母开始
        start = i + len("LISTENING") + 7
        end = result.index("\n")
        pid = result[start:end]

        # 关闭被占用的端口号
        cmd_kill_pid = "taskkill -f -pid %s" % pid
        print(cmd_kill_pid)
        os.popen(cmd_kill_pid)

    else:
        print("port %s is available!" % port)



if __name__ == "__main__":
    host = "127.0.0.1"
    for i in range(2):
       port = 4723 + 2 * i
       release(port)

1.1）端口释放成功如下