【临实战】使用 Python 运维之并发与并行

验证一下新的 Docker 镜像仓库（Docker Registry）是否迁移成功了

subprocess

import subprocess

def run_cmd(cmd):
    return subprocess.Popen(cmd, 
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE).communicate()

管道其实就是文件描述符对，子进程会继承父进程中的所有文件描述符

stdout, stderr = run_cmd('uname -a')

cat imagelist | wc -l

思路：

• 拉取列表中的镜像，拉取成功后将其删除并标记为成功
• 拉取失败就标记为失败和并记录错误
• 如果拉取超时，就标记超时

with open('timeout_image.txt','a') as timeout_file:
    timeout_file.write(image)

def pull_worker(image):
    try:
        # 这里我们设置了 10 分钟的超时
        with Timeout(600):
            cmd = 'docker pull {0}'.format(image)
            shell_cmd(cmd)
            out, err = shell_cmd('docker images {0}'.format(image))
            # 如果验证 pull 成功
            if not err:
                out, err = shell_cmd('docker rmi {0}'.format(image))
                # 如果删除镜像失败
                if err:
                    print "[ERROR docker rmi] {0} {1}".format(image, err)
                    return
                # 完美
                with open('good_image.txt','a') as goodImageFile:
                    goodImageFile.write(image)
            # 如果 pull 失败，将镜像名和失败信息写入文件
            else:
                with open('error_image.txt','a') as timeout_file:
                    timeout_file.write(image + " {0}".format(err))
    # 如果超时
    except Timeout:
        with open('timeout_image.txt','a') as timeout_file:
            timeout_file.write(image)

sys

if len(sys.argv) == 4:
    pass
else:
    print "Need three params:\n# 1 File\n# 2 Process numbers\n# 3 Parallels numbers for each process"
    return


# 这里同样使用了序列解包，第一个参数是脚本名字，忽略掉
_, file, coreNum, poolNum = sys.argv

python check_images.py imagelist 8 5

gevent

pip install gevent

import gevent.pool
import gevent.monkey
from gevent import Timeout
gevent.monkey.patch_all() # 猴子补丁
from multiprocessing import Process

不要和我说什么Python 有全局解释器锁（GIL），多进程没有 GIL，多进程没有 GIL，多进程没有 GIL

1. 启动和核数相等的进程（跑满机器，尽快完成任务为目的）
2. 每个进程里面 docker pull 的并发为 5（gevent 协程池）

def each_process(task_object_list):
    pool = gevent.pool.Pool(int(poolNum))
    pool.map(pull_worker, task_object_list)
    stop = time.time()
    elapsed = stop - start
    print "End precess with {0} s".format(elapsed)

with open(file) as f:
    for line in f:
        line = line.strip()
        all_task_list.append(line)

print "All task: {0}".format(len(all_task_list))
for sliced_task_list in slice_list(all_task_list, int(coreNum)):
    print "Start process with tasks: {0}".format(len(sliced_task_list))
    p = Process(target=each_process, args=(sliced_task_list,))
    p.start()