专栏首页随心DevOps[实战篇] Python 运维中使用并发

[实战篇] Python 运维中使用并发

今天从大哥手里接了一个需求:

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

简单粗暴的方法就是拿到老仓库中的镜像列表(Image List),在新仓库模拟用户重新拉取(pull)一遍来验证,我们开始


subprocess

如果我们用 Shell 来写,执行 Docker 命令很容易,直接写就是了,但是对结果的判断就不那么友好了(Shell 大神忽略),那么 Python 呢,如何优雅的执行 Linux 命令呢?这里我们用到了一个 Python 标准库(standard module) :

import subprocess

我们都知道,命令执行过程中会有标准输出(stdout)和标准错误(stderror):

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

查看了一下行数(镜像数量),4254 个,还行,不算太多


思路:

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

如何标记呢,因为我们将会使用多进程,多个进程间通信还是蛮麻烦的,这里偷个懒:直接使用 append 模式直接将结果写入文件

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

我们先写出如何验证一个镜像的逻辑:

def pull_worker(image):
    # ?

公众号代码支持太差了,可以去文末的点击阅读原文查看

后面就仅仅是并发的问题了


sys

首先我们想控制并发数量,最简单是使用 sys 模块

if len(sys.argv) == 4:
    pass
else:
    print “Need three params
    return
    
# 这里同样使用了序列解包,第一个参数是脚本名字,忽略掉
_, file, coreNum, poolNum = sys.argv

这样的程序执行起来像这样:

python check_images.py imagelist 8 5

gevent

然后是实现,我们使用的这个模块需要安装,它是大名鼎鼎的 gevent,为什么使用它,因为我们的任务是 I/O密集 型的,gevent 擅长处理这类任务(有兴趣可以去了解下猴子补丁)

pip install gevent

我们看导入模块的代码:

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

最后一行也是使用了 Python 的标准库,多进程模块:multiprocessing

不要和我说什么Python 有全局解释器锁(GIL),多进程没有 GIL,多进程没有 GIL,多进程没有 GIL

如何并发呢:

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

所以我们总的并发数就是 40,这样就完成了可控制并发的脚本

代码如下:

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()

这里需要注意的一点是,4254 个镜像,是按照核心数量分组(slice_list),然后交给不同的进程处理的。

本文分享自微信公众号 - 随心DevOps(heart-devops),作者:临书

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-01-10

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 你可能不知道的 Django Rest Framework 的两个新特性

    前端、甚至大前端发展如此火爆的 2017 年(我相信 2018 年一样)刚过去,像 MVC 架构那样直接使用后端模板渲染前端页面的方式已经不被推崇了(当然有些场...

    临书
  • 使用 React 和 Django REST Framework 构建你的网站

    在我们最近的工作中,构建网站使用的架构是带有 Django REST Framework(DRF)后端的 React 前端。它们是通过在前端使用 axios(前...

    临书
  • 如何量化并行效率?阿姆达尔来帮你

    今天在《代码的未来》这本书中看到一个定律,叫:阿姆达尔定律,原来并行效率可以量化的这么漂亮,它叫 加速比,简单的讲就是提升效率的倍数,很简单就一个公式: ? 看...

    临书
  • 在main方法之前,到底执行了什么?

    本人在做接口测试的时候,需要用一个公共类来把所有的执行的代码,然后这个公共类有hsot和hosttype等属性来区分各个测试环境,然后在去不同的地方取用例和请求...

    FunTester
  • 网络协议&建立TCP连接 原

    OSI/RM模型(Open System Interconnection/Reference Model)将计算机网络体系结构的通信协议划分为七层,自下而上依次...

    阿dai学长
  • WCF Web API 说再见,继承者ASP.NET Web API

    从 .NET 3.5 开始 WCF 已经支持用 WebHttpBinding 构建 RESTful Web 服务,基于 WCF 框架的 RESTful Web ...

    张善友
  • JAVA实现Excel导入导出

    IT故事会
  • Web项目或WCF发布IIS后,如何通过VS2010调试

    在做项目的时候,例如WCF服务一般都会将WCF服务承载于控制台应用程序,或者WinForm窗体应用程序,因为这样可以直接在服务代码上打断点,然后就可以调试了。但...

    aehyok
  • UWP IRandomAccessStream 与 Stream 互转

    本文告诉大家如何将 IRandomAccessStream 和 Stream 互转

    林德熙
  • Spark常见错误问题汇总

    王知无

扫码关注云+社区

领取腾讯云代金券