带Python的DevOps：Python consturrent.Futures并发教程 - 现实世界的示例-DEV365 开发者社区

作者注：此博客文章不是Python或Devops的初学者指南。

假定Python，Devops，Kubernetes和Helm的基本知识。

0背景

0.1为什么要python

编程语言随着时间的推移而上升和下降。

荷兰软件质量保证公司

Tiobe一直在跟踪编程语言的普及。根据其programming community index（及其首席执行官Paul Jansen）的说法，Python现在排名第一：“这是20多年来的第一次，我们有了新的领导者：Python编程语言。 - Java和C的霸权已经结束。”

0.2为什么要使用Python的Devops

引用Real Python：

Python是练习DevOps的团队使用的主要技术之一。它的灵活性和可访问性使Python非常适合这项工作，使整个团队能够构建Web应用程序，数据可视化以及通过自定义实用程序改善其工作流程。

最重要的是，Ansible和其他流行的DevOps工具是用Python编写的，可以通过Python进行控制。

加上，我是Python易于阅读的无支架代码样式的忠实拥护者。

一个人可能想知道为什么易于阅读的内容如此必不可少。 '推杆用模棱两可的变量名称，冗长的函数，一千个（如果不是数千）行的代码或所有这些文件都在一起执行代码没有问题。它会正确运行，对

好吧，是的。但是引用Knuth：

程序旨在由人类阅读，仅偶然才能执行计算机。

发明了所有方法和想法，例如重构，干净的代码，命名惯例，代码气味等，以便我们，人类，可以更好地阅读代码，而不是计算机可以更好地运行它。

0.3为什么并发

好吧，这很容易：

因为我们可以。

开玩笑，原因当然是绩效：

并发更快（通常）。

例如，如果您在一个kubernetes群集的一个名称空间中安装了多个头盔图表，并且您想清除所有掌舵版本，当然，您可以一个一个一个一个卸载它们，等待第一个版本要卸载，请卸载，然后开始卸载第二个，等等

对于某些应用，舵机卸载部分可能会很慢。

即使在一些简单的图表中，同时卸载它们仍然可以大大节省时间。

基于本地测试，卸载了三个头盔图（NGINX，REDIS和MYSQL），一一将C.A. 0.8秒，虽然同时完成为0.48，但降低了40％。

如果问题的规模上升了，就像您有数十个图表要卸载，并且需要在多个名称空间中进行操作，则必须保存的时间。

接下来，让我们使用python来处理这个特定的示例。

1任务

您有多个团队和开发人员，他们与DevEnv。

共享相同的Kubernetes群集

为了实现资源隔离，将一个名称空间分配给每个开发人员。每个开发人员都需要进行一些掌舵安装，以使他们的应用程序和依赖关系启动和运行，以便他们可以开发和测试。

现在，由于有许多名称空间，每个名称空间中都有许多舵机版本，并且许多pods占用了许多节点，因此您可能想通过在工作时间结束时删除所有这些POD来优化成本，以便群集群集可以缩小以节省一些VM成本的美元。

您想要某种形式的自动化，以在某些名称空间中卸载所有版本。

让我们在Python中解决这个问题。出于演示目的，我们将在default名称空间中安装nginx，redis和mysql，然后编写一些自动化内容以删除'em。

让我们走。

2准备测试我们自动化的环境

不是我们正在进行测试驱动的开发，而是在编写任何代码之前，让我们创建一个本地环境作为对当前问题的模拟，以便我们有一些测试我们的自动化脚本的东西。

在这里，我们使用Docker，Minikube和Helm。如果您尚未安装它们，请查看官方网站：

通过运行：
开始您的本地kubernetes群集

minikube start

然后，在default名称空间中安装一些舵表图，我们将使用自动化来删除以后：

# make sure we select the default namespace
kubectl config set-context --current --namespace=default

# add some helm repos and update
helm repo add nginx-stable https://helm.nginx.com/stable
helm repo add bitnami https://charts.bitnami.com/bitnami
helm repo update

# install three applications that we will use automation to delete
helm install nginx nginx-stable/nginx-ingress
helm install redis bitnami/redis
helm install mysql bitnami/mysql

本地测试模拟完成。

3个非电流版本

首先，让我们编写一些单线程，非连续代码以解决此问题。

我们将使用subprocess模块运行helm list来获取所有版本，在所有版本上运行一个简单的循环，然后将其运行。这里没有什么幻想的；仅使用Python运行一些CLI命令。

谈话很便宜；向我展示代码：

https://gist.github.com/IronCore864/ca1e74a65f4a97937d93c63c094e9d32

4介绍`concurrent.futures`

4.1 `multiprocessing`和`threading`

直接跳入concurrent.futures（如本博客的标题中所宣传的那样），让我们谈谈multiprocessing和threading：

threading模块可让您使用多个线程（也称为轻量级过程或任务） - 控制共享其全局数据空间的多个线程。
multiprocessing是支持产卵过程的软件包。 multiprocessing使用子过程而不是线程解决了全局解释器锁定问题。

在两者之间进行选择时，简单地（可能不是100％精确，但这就是要点）：

如果您的任务是CPU密集型的，请使用multiprocessing（通过使用多个进程而不是线程来绕过GIL问题）。
如果您的任务是i/o密集型，则threading模块应起作用。

4.2无论如何，什么是`concurrent.futures`？

现在我们已经有了这两个模块，什么是concurrent.futures？

这是一个高级接口，可以在threading和multiprocessing的顶部启动异步任务和抽象层。当您只想同时运行一件代码并且不需要threading或multiprocessing模块API提供的额外功能时，这是首选工具。

4.3以示例学习

好的，足够的理论，让我们以一个例子来弄脏和学习，这是官方文档中的一个例子：

import concurrent.futures
import urllib.request

URLS = ['http://www.cnn.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        data = future.result()
        print('%r page is %d bytes' % (url, len(data)))

一些观察：

executor，无论它是什么都需要的。（根据文档，可以是ThreadPoolExecutor或ProcessPoolExecutor。）
Executor.submit()方法“提交”（或以普通英语为“计划”）函数调用（带有参数）并返回未来对象。
concurrent.futures.as_completed方法返回未来实例的迭代器。

5解决任务的并发代码

一旦我们理解语法并通过从示例中复制和粘贴并具有一些创意来获得对其实际工作方式的基本了解，就可以很容易地将我们的非电流版本从上一节转换为并发。将所有内容放在一起：

请参阅下面的代码：

https://gist.github.com/IronCore864/ad21130aa796d407624805c5342201db

voila！

请注意，concurrent.futures部分与官方concurrent.futures示例完全相同。

概括

concurrent.futures备忘单：

    # or, with concurrent.futures.ProcessPoolExecutor()
    with futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_objects = {
            executor.submit(some_func, param1, param2 ...): param1 for param1 in xxx
        }

        for f in futures.as_completed(future_objects):
            res = future_objects[f]
            do_something()

经验法则：使用ThreadPoolExecutor进行I/O密集型工作量，而ProcessPoolExecutor用于CPU密集型工作负载。

如果您喜欢这篇文章，请喜欢，评论，订阅。在下一块中见。

0背景