在上一章中，我们学习了如何使用 PIR 传感器检测运动，以及如何使用超声波传感器和霍尔效应传感器测量距离和检测运动。
在本章中，我们将讨论在使用电子传感器（输入设备）和执行器（输出设备）时，构建Python 程序的替代方法。我们将介绍经典的事件循环编程方法，然后介绍更高级的方法，包括在 Python 中使用线程、发布者/订阅者模型，最后是使用 Python 进行异步 I/O 编程。
我向你保证，互联网上有很多关于这些主题的博客文章和教程；然而，我们将在本章中介绍的内容将特别侧重于实际的电子接口。本章中我们的方法将涉及创建一个简单的电路，其中包含一个按钮、一个电位计和两个 LED，我们将使其以不同的速率闪烁，并介绍四种不同的编码方法以使电路工作。
以下是我们将在本章中介绍的内容：
- 构建和测试我们的电路
- 探索事件循环方法
- 探索线程方法
- 探索发布者-订阅者替代方案
- 探索异步 IO 方法
要执行本章中的练习，您需要以下内容：
- 树莓皮 4 B 型
- Raspbian OS Buster（带桌面和推荐软件）
- 最低 Python 版本 3.5
这些需求是本书中代码示例的基础。只要您的 Python 版本是 3.5 或更高版本，就可以合理地期望代码示例在 Raspberry Pi 3 Model B 或不同版本的 Raspbian OS 上无需修改即可工作。
您可以在 GitHub 存储库的chapter12文件夹中找到本章的源代码，该文件夹位于https://github.com/PacktPublishing/Practical-Python-Programming-for-IoT 。
您需要在终端中执行以下命令，以设置虚拟环境并安装本章代码所需的 Python 库：
$ cd chapter12 # Change into this chapter's folder
$ python3 -m venv venv # Create Python Virtual Environment
$ source venv/bin/activate # Activate Python Virtual Environment
(venv) $ pip install pip --upgrade # Upgrade pip
(venv) $ pip install -r requirements.txt # Install dependent packages以下依赖项是从requirements.txt安装的：
- PiGPIO：PiGPIO GPIO 库（https://pypi.org/project/pigpio
- ADS1X15：ADS1X15 ADC 库（https://pypi.org/project/adafruit-circuitpython-ads1x15
- PyPubSub：进程内消息和事件（https://pypi.org/project/PyPubSub
本章练习所需的电子元件如下：
- 2 个红色发光二极管
- 2 x 200Ω电阻器
- 1 个按钮开关
- 1 个 ADS1115 模块
- 1 x 10kΩ电位计
为了最大限度地提高您在本章中的学习，我们对已有的知识和经验进行了一些假设：
- 从电子接口的角度来看，我假设您已经阅读了本书前面的 11 章，并且对本书中介绍的 PiGPIO 和 ADS1115 Python 库感到满意。
- 从编程的角度来看，我假设已有的面向对象编程（OOP技术）知识以及它们是如何在 Python 中实现的。
- 熟悉事件循环、线程、**发布者订阅者、和同步与异步范例的概念也将是有利的。
如果您不熟悉上述任何主题，您会发现许多在线教程非常详细地介绍了这些主题。建议请参见本章末尾的进一步阅读部分。
我将以实践练习的形式介绍本章的电路和程序。让我们暂时假设我们被要求设计并构建一个具有以下要求的gizmo：
- 它有两个发光二极管闪烁。
- 电位计用于调整 LED 闪烁的速率。
- 当程序启动时，两个 LED 将以电位计位置确定的相同速率闪烁。
- 闪烁率为 0 秒表示 LED 熄灭，而最大闪烁率为 5 秒表示 LED 点亮 5 秒，然后熄灭 5 秒，然后再重复循环。
- 一个按钮用于选择调整电位计时哪个 LED 改变其闪烁率。
- 当按下按钮并保持 0.5 秒时，所有 LED 同步到相同的速率，由电位计的位置确定。
- 理想情况下，程序代码应易于扩展，以最少的编码工作量支持更多的 LED。
下面是一个演示 gizmo 使用的场景：
- 通电（程序启动）后，所有 LED 开始以 2.5 秒的速率闪烁，因为电位计的刻度盘位于其旋转的中点（50%）。
- 用户调整电位计，使第一个LED 以 4 秒的速率闪烁。
- 接下来，用户短暂按下并释放按钮，电位计将改变秒LED 的闪烁率。
- 现在，用户调整电位计，使秒LED 以 0.5 秒的速率闪烁。
- 最后，用户按下并按住按钮 0.5 秒，使第一个和第二个LED 以 0.5 秒的速率一致闪烁（电位计在步骤 4中设置的速率）。
现在，对于我提到的挑战，在我们进入本章的电路和代码之前，我要求您停止阅读，尝试创建一个电路，并编写一个实现上述要求的程序。
You will find a short video demonstrating these requirements at https://youtu.be/seKkF61OE8U.
我预计您将遇到挑战，并对最佳方法提出疑问。没有最好的方法；然而，通过拥有自己的实现——不管它是否有效——您将能够与我将在本章中介绍的四种解决方案进行比较和对比。我相信，如果你先尝试一下，你会获得更深的理解和更多的洞察力。嘿，也许你会创造一个更好的解决方案！
如果您需要一些建议来帮助您开始，请访问以下网站：
当您准备好后，我们将查看满足上述要求的电路。
在图 12.1中，电路符合我们刚才列出的要求。它有一个按钮，一个分压器形式的电位计连接到 ADS1115 模数转换器，两个 LED 通过限流电阻器连接。添加额外的 LED 将非常简单，只需在 GND 和一个空闲 GPIO 引脚之间连接更多的 LED 和电阻器对：
Figure 12.1 – Reference circuit schematic
如果您还没有自己创建类似的电路，我们现在将在您的试验板上创建此电路。我们将分三个部分构建此电路。让我们开始：
Figure 12.2 – Reference circuit (part 1 of 3)
下面是创建我们的试验板构建的第一部分所要遵循的步骤，我们将在其中放置组件。步骤编号与图 12.2中黑色圆圈中的编号匹配：
- 将 ADS1115 模块放入试验板。
- 将电位计放入你的实验板。
- 将 LED 放入您的实验板，注意如图所示定位 LED 的腿部。
- 将第二个 LED 放入您的实验板中，注意如图所示确定 LED 支脚的方向。
- 将一个 200Ω电阻器（R1）放入试验板。该电阻器的一端与步骤 3中放置的 LED 的阳极支腿共用一行。
- 将另一个 200Ω电阻器（R2）放入试验板。该电阻器的一端与您在步骤 5中放置的第二个 LED 的阳极支腿共用一行。
- 把按钮放在你的实验板上。
现在，我们已经将组件放置到试验板中，让我们开始对它们进行布线：
Figure 12.3 – Reference circuit (part 2 of 3)
下面是继续我们的实验板构建的第二部分需要遵循的步骤。步骤编号与图 12.3中黑色圆圈中的编号匹配：
- 将 Raspberry Pi 的一个 3.3 伏针脚连接到左侧电源导轨的正极导轨上。
- 将 ADS1115 的 Vdd 端子连接至左侧电源导轨的正极导轨。
- 将 ADS1115 的 GND 端子连接至左侧电源轨的负极导轨。
- 将 ADS1115 的 SCL 端子连接到 Raspberry Pi 上的 SCL 引脚。
- 将 ADS1115 的 SDA 端子连接到 Raspberry Pi 上的 SDA 引脚。
- 将 Raspberry Pi 上的 GND 引脚连接到左侧电源导轨的负极导轨上。
- 将电位计的外端子连接到左侧电源导轨的正极导轨上。
- 将电位计的另一个外部端子连接到左侧电源导轨的负极导轨上。
- 将电位计的中心端子连接到 ADS1115 的端口 A0。
您是否记得，此配置中的电位计正在创建一个可变分压器？如果没有，您可能想为软件工程师重温第 6 章、电子 101。此外，如果您想了解有关 ADS1115 模块的详细信息，请参阅第 5 章、将您的树莓 Pi 连接到物理世界。
让我们继续构建：
Figure 12.4 – Reference circuit (part 3 of 3)
下面是继续我们的试验板构建的最后一部分所要遵循的步骤。步骤编号与图 12.4中黑色圆圈中的编号匹配：
- 将 GPIO 26 从 Raspberry Pi 连接到 200Ω电阻器（R1）。
- 将 Raspberry Pi 的 GPIO 19 连接到第二个 200Ω电阻器（R2）。
- 将覆盆子 Pi 的 GPIO 21 连接到按钮的一条腿上。
- 将 LED 的两个阴极支脚连接在一起。
- 将 LED 的阴极支脚连接到左侧电源导轨的负极导轨上。
- 将按钮的第二个支脚连接到左侧电源导轨的负极导轨上。
现在我们已经完成了电路构建，我们准备运行示例代码以使电路工作。
本章提供了四种不同版本的代码，可用于前面图 12.1 所示的电路。您将在chapter12文件夹中找到按版本组织的代码：
chapter12/version1_eventloop是一个基于事件循环的示例。chapter12/version2_thread是一个基于线程和回调的示例。chapter12/version3_pubsub是一个基于发布者订户的示例。chapter12/version4_asyncio是一个基于异步 IO异步 IO的示例。**
***所有版本在功能上是等效的；然而，它们在代码结构和设计上有所不同。测试电路后，我们将更详细地讨论每个版本。
以下是运行每个版本（从版本 1 开始）和测试电路所需的步骤：
- 更改到
version1_eventloop文件夹。 - 简要查看
main.py源文件和文件夹中的任何其他 Python 文件，了解它们包含的内容以及程序的结构。 - 在终端中运行
main.py（记得先切换到章节的虚拟环境）。
At this point, if you receive errors regarding I2C or ADS11x5, remember that there is the i2cdetect tool, which can be used to confirm that an I2C device such as the ADS1115 is correctly connected and visible to your Raspberry Pi. Refer to Chapter 5 , Connecting Your Raspberry Pi to the Physical World , for more information.
- 转动电位计刻度盘，观察第一个LED 的闪烁率变化。
- 短按按钮。
- 转动电位计刻度盘，观察秒LED 闪烁率变化。
- 按住按钮 0.5 秒，观察两个 LED 现在以相同的速率同时闪烁。
以下是您将收到的终端输出示例：
(venv) $ cd version1_eventloop
(venv) $ python main.py
INFO:Main:Version 1 - Event Loop Example. Press Control + C To Exit.
INFO:Main:Setting rate for all LEDs to 2.5
INFO:Main:Turning the Potentiometer dial will change the rate for LED #0
INFO:Main:Changing LED #0 rate to 2.6
INFO:Main:Changing LED #0 rate to 2.7
INFO:Main:Turning the Potentiometer dial will change the rate for LED #1
INFO:Main:Changing LED #1 rate to 2.6
INFO:Main:Changing LED #1 rate to 2.5
# Truncated
INFO:Main:Changing LED #1 rate to 0.5
INFO:Main:Changing rate for all LEDs to 0.5- 在终端中按Ctrl+C退出程序。
- 对
version2_threads、version3_pubsub和version4_asyncio重复步骤 1到8。
您刚刚测试并浏览了四个不同程序的源代码（如果您挑战自己创建自己的程序，可能有五个），它们都以不同的方式实现了完全相同的最终结果。
现在是了解这些程序是如何构建的时候了。让我们从程序的事件循环版本开始。
我们将通过讨论一种基于事件循环的方法来构建我们在上一节中刚刚测试过的示例 gizmo，从而开始我们的代码探索。
基于事件循环的方法的代码可以在chapter12/version1_eventloop文件夹中找到。您将找到一个名为main.py的文件。请花点时间停下来阅读main.py中包含的代码，以基本了解程序的结构和工作原理。或者，您可以添加断点或在代码中插入print()语句，然后再次运行它以了解其工作原理。
进展如何，你注意到了什么？如果你觉得恶心或者迷失在循环、if语句和状态变量的网络中，那么做得好！这意味着您投入了时间来考虑这种方法以及代码是如何构造的。
我所说的事件循环方法在代码中通过第 1 行中缩写的while True:循环进行了演示：
# chapter12/version1_eventloop
#
# Setup and initialization code goes before while loop.
#
if __name__ == "__main__":
# Start of "Event Loop"
while True: # (1)
#
# ... Main body of logic and code is within the while loop...
#
sleep(SLEEP_DELAY)诚然，我本可以使用函数甚至外部类来减少while循环中代码的数量（并可能提高可读性），但是，总体设计范式保持不变——程序控件的主体处于永久循环中。
If you are familiar with Arduino programming, you will be intimately familiar with this approach to programming. That's why I titled this section event-loop due to the similarity of approach and the popularity of the term. Notwithstanding, note that the term event-loop has a wider context within Python, as we will see when we look at the AsyncIO (version 4) of our program.
您可能已经意识到，本书中的许多示例都使用了这种事件循环编程方法。三个例子如下：
- 当我们想要一个定时事件，比如 LED 闪烁（第 2 章、Python 和 IoT 入门）
- 轮询 DHT 11 或 DHT 22 温度/湿度传感器（第 9 章、测量温度、湿度和亮度）
- 轮询连接到光相关电阻器（LDR）的 ADS1115 模数转换器（也可用于第 9 章，测量温度、湿度和光照等级）
在这种情况下，对于一个单一的重点示例，事件循环是有意义的。当你尝试新的想法，学习新的执行器或传感器时，它们甚至纯粹为了方便而变得有意义。然而，正如我们的version1_eventloop/main.py程序所演示的，只要您添加多个组件（如电位计、两个 LED 和一个按钮），并希望它们为特定目的一起工作，代码就会很快变得复杂。
例如，考虑下面的 3 行代码，它负责闪烁所有的 LED，并且记住这个代码块在每个循环迭代中被评估一次，并且负责闪烁每个 LED：
#
# Blink the LEDs.
#
now = time() # (3)
for i in range(len(LED_GPIOS)):
if led_rates[i] <= 0:
pi.write(LED_GPIOS[i], pigpio.LOW) # LED Off.
elif now >= led_toggle_at_time[i]:
pi.write(LED_GPIOS[i], not pi.read(LED_GPIOS[i])) # Toggle LED
led_toggle_at_time[i] = now + led_rates[i]将其与普通的替代方案（类似于我们将在其他方法中看到的内容）进行比较，一眼就能明白：
while True:
pi.write(led_gpio, not pi.read(led_gpio)) # Toggle LED GPIO High/Low
sleep(delay)如果你还考虑下面的代码块，从第 2 行开始，它负责检测按钮按压，那么你会发现近 40 行代码（在实际的 To.t0 文件中）只是检测按钮正在做什么：
while True:
button_pressed = pi.read(BUTTON_GPIO) == pigpio.LOW # (2)
if button_pressed and not button_held:
# Button has been pressed.
# ... Truncated ...
elif not button_pressed:
if was_pressed and not button_held:
# Button has been released
# ... Truncated ...
if button_hold_timer >= BUTTON_HOLD_SECS and not button_held:
# Button has been held down
# ... Truncated ...
# ... Truncated ...您将计算起作用的多个变量–button_pressed、button_held、was_pressed和button_hold_timer–这些变量都在每个while循环迭代中进行评估，主要用于检测按钮保持事件。我相信您会理解，像这样编写和调试代码可能会很乏味，而且容易出错。
We could have used a PiGPIO callback to handle button presses outside of the while loop, or even a GPIO Zero Button class. Both approaches would help reduce the complexity of the button-handling logic. Likewise, maybe we could have mixed in a GPIO Zero LED class to handle the LED blinking. However, if we did, our example would not be a purely event-loop-based example.
现在，我并不是说事件循环是一种糟糕或错误的方法。它们有它们的用途，它们是需要的，本质上，我们每次使用while循环或另一个循环构造时都会创建一个循环—因此基本理想无处不在，但它不是构建复杂程序的理想方法，因为这种方法使它们更难理解、维护和调试。
每当你发现你的程序正朝着这个事件循环路径前进，停下来思考时，因为也许是时候考虑重构你的代码，采用一种不同的和更可维护的方法，比如线程/回调方法，我们将在下一步研究。
现在我们已经探索了基于事件循环的方法来创建程序，让我们考虑使用线程、回调和 OOP 构建的另一种方法，并看看这种方法如何提高代码可读性和可维护性，并促进代码重用。
基于线程的方法的代码可以在chapter12/version2_threads文件夹中找到。您将找到四个文件–主程序main.py和三个类定义：LED.py、BUTTON.py和POT.py。
请花点时间停下来阅读main.py中包含的代码，以基本了解程序的结构和工作方式。然后，继续审查LED.py、BUTTON.py和POT.py。
进展如何，你注意到了什么？我猜您会发现这个版本的程序（在阅读main.py时）更快、更容易理解，并且注意到没有繁琐复杂的while循环，而是pause()调用，这是阻止我们的程序退出所必需的，如第 3 行所总结：
# chapter12/version2_threads/main.py
if __name__ == "__main__": # (3)
# Initialize all LEDs
# ... Truncated ...
# No While loop!
# It's our BUTTON, LED and POT classes and the
# registered callbacks doing all the work.
pause()在本程序示例中，我们采用了面向对象技术，并使用三个类将程序组件化：
- 一个按钮类（
BUTTON.py，负责所有按钮逻辑 - 电位计类（
POT.py，负责所有电位计和模数转换逻辑 - LED 类（
LED.py，负责制作单个LED 闪光灯
通过使用 OOP 方法，我们的main.py代码大大简化。它现在的角色是创建和初始化类实例，并容纳使程序工作的回调处理程序和逻辑。
考虑下面的 OOP 方法用于我们的按钮：
# chapter12/version2_threads/main.py
# Callback Handler when button is pressed, released or held down.
def button_handler(the_button, state):
global led_index
if state == BUTTON.PRESSED: # (1)
#... Truncated ...
elif state == BUTTON.HOLD: # (2)
#... Truncated
# Creating button Instance
button = BUTTON(gpio=BUTTON_GPIO,
pi=pi,
callback=button_handler)与事件循环示例中的按钮处理代码相比，这大大简化了，可读性也提高了很多–非常明确地说明了此代码在哪里以及如何响应第 1 行中按下的按钮和第 2 行中保持的按钮。
让我们考虑一下在 Type T1 文件中定义的 AUT0T0 类。此类是 PiGPIO 回调函数的增强包装，该函数将按钮 GPIO 引脚的HIGH/LOW状态转换为PRESSED、RELEASED和HOLD事件，如BUTTON.py第 1 行的以下代码所述：
# chapter12/version2_threads/BUTTON.py
def _callback_handler(self, gpio, level, tick): # PiGPIO Callback # (1)
if level == pigpio.LOW: # level is LOW -> Button is pressed
if self.callback: self.callback(self, BUTTON.PRESSED)
# While button is pressed start a timer to detect
# if it remains pressed for self.hold_secs
timer = 0 # (2)
while (timer < self.hold_secs) and not self.pi.read(self.gpio):
sleep(0.01)
timer += 0.01
# Button is still pressed after self.hold_secs
if not self.pi.read(self.gpio):
if self.callback: self.callback(self, BUTTON.HOLD)
else: # level is HIGH -> Button released
if self.callback: self.callback(self, BUTTON.RELEASED)与事件循环示例的按钮处理代码相比，我们没有引入和询问多个状态变量来检测按钮保持事件，而是在第 2 行将此逻辑简化为简单的线性方法。
接下来，当我们考虑 AUT0T0 类（在 Tyl T1 中定义）和 AuthT2R 类（在 ToalT3^中定义）时，我们将看到线程进入我们的程序。
Did you know that even in a multi-threaded Python program, only one thread is active at a time? While it seems counter-intuitive, it was a design decision known as the Global Interpreter Lock ( GIL ) made back when the Python language was first created. If you want to learn more about the GIL and the many other forms of achieving concurrency with Python, you will find resources in the Further reading section of this chapter.
以下是POT类的线程运行方法，可在POT.py源文件中找到，并从第 1 行开始说明了中间轮询 ADS1115 ADC 以确定电位计位置的方法。在本书中，我们已经多次看到这个轮询示例，从第 5 章开始，将树莓 Pi 连接到物理世界，在那里我们首先讨论了模数转换、ADS1115 模块和电位计：
# chapter12/version2_threads/POT.py
def run(self):
while self.is_polling: # (1)
current_value = self.get_value()
if self.last_value != current_value: # (2)
if self.callback:
self.callback(self, current_value) # (3)
self.last_value = current_value
timer = 0
while timer < self.poll_secs: # Sleep for a while
sleep(0.01)
timer += 0.01
# self.is_polling has become False and the Thread ends.
self.__thread = None这里代码的不同之处在于，我们正在监测 ADC 第 2 行上的电压变化（例如，当用户转动电位计时），并将其转换为第 3 行上的回调，当您查看该文件中的源代码时，您将在main.py中看到这一点。
现在让我们讨论一下我们是如何实现version2LED 相关代码的。正如您所知，以定义的速率打开和关闭 LED 的基本代码模式包括一个while循环和一个sleep语句。这是 LED 类中采用的方法，如LED.py中第 3 行的run()方法所示：
# chapter12/version2_threads/LED.py
def run(self): # (3)
""" Do the blinking (this is the run() method for our Thread) """
while self.is_blinking:
# Toggle LED On/Off
self.pi.write(self.gpio, not self.pi.read(self.gpio))
# Works, but LED responsiveness to rate chances can be sluggish.
# sleep(self.blink_rate_secs)
# Better approach - LED responds to changes in near real-time.
timer = 0
while timer < self.blink_rate_secs:
sleep(0.01)
timer += 0.01
# self.is_blinking has become False and the Thread ends.
self._thread = None我相信您会同意，这比我们在上一节讨论的事件循环方法更容易理解。然而，重要的是要记住，事件循环方法是在一个单个代码块中，在一个单个线程（程序的主线程）内处理并改变所有LED 的闪烁率。
Notice the two sleep approaches shown in the preceding code. While the first approach using sleep(self.blink_rate_secs) is common and tempting, the caveat is that it blocks the thread for the full duration of the sleep. As a result, the LED will not respond to rate changes immediately and will feel sluggish to a user when they turn the potentiometer. The second approach, commended #Better approach, alleviates this issue and allows the LED to respond to rate changes in (near) real time.
我们的version2程序示例使用具有自己内部线程的 LED 类，这意味着我们有多个线程——每个 LED 一个——所有线程都使 LED 相互独立闪烁。
你能想到这可能带来的任何潜在问题吗？好的，如果你已经阅读了version2源文件，这可能是显而易见的——当按下按钮 0.5 秒时，所有 LED 都以相同的速率同步闪烁！
通过引入多个线程，我们引入了多个计时器（即，sleep()语句），因此每个线程都根据自己的独立时间表闪烁，而不是从一个共同的参考点开始闪烁。
这意味着，如果我们在多个 LED 上简单地调用led.set_rate(n)，虽然它们都会以n的速率闪烁，但它们不一定会同时闪烁。
解决这个问题的一个简单方法是，在我们开始以相同的速率闪烁之前，同步关闭所有 LED。也就是说，我们从一个公共状态（即关闭）开始它们闪烁，然后开始它们一起闪烁。
以下代码段显示了这种方法，代码段从LED.py中的第 1 行开始。同步的核心是通过第 2 行的led._thread.join()语句实现的：
# chapter12/version2_threads/LED.py
@classmethod # (1)
def set_rate_all(cls, rate):
for led in cls.instances: # Turn off all LEDs.
led.set_rate(0)
for led in cls.instances:
if led._thread:
led._thread.join() # (2)
# We do not get to this point in code until all
# LED Threads are complete (and LEDS are all off)
for led in cls.instances: # Start LED's blinking
led.set_rate(rate)这是一个很好的同步第一步，出于实际目的，它非常适合我们的情况。如前所述，我们所做的就是确保我们的 LED 同时从关闭状态开始闪烁（好的，非常非常非常接近同一时间，取决于 Python 在for循环中迭代所需的时间）。
Try commenting out led._thread.join() and the embodying for loop on line 2 in the preceding code and run the program. Make the LEDs blink at different rates, then try to synchronize them by holding down the button. Does it always work?
但是，必须注意的是，我们仍在处理多线程和独立计时器，以使 LED 闪烁，因此存在发生时间漂移的可能性。如果这是一个实际问题，那么我们需要探索其他技术来同步每个线程中的时间，或者我们可以创建并使用一个类来一起管理多个 LED（基本上使用事件循环示例中的方法，仅将其重构为一个类和一个线程）。
这里关于线程的要点是，当您将线程引入应用程序时，您可以引入可能被设计或同步的计时问题。
If your first pass at a prototype or new program involves an event-loop-based approach (as I often do), then as you refactor that code out into classes and threads, always think about any timing and synchronizing issues that may arise. Discovering synchronization-related bugs by accident during testing (or worse, when in production) is frustrating as they can be hard to reliably replicate, and could result in the need for extensive rework.
我们刚刚看到了如何使用 OOP 技术、线程和回调创建示例 gizmo 程序。我们已经看到了这种方法如何使代码更易于阅读和维护，我们还发现了同步线程化代码所需的额外需求和工作。接下来，我们将看一下我们程序的第三个变体，它基于发布者-订阅者模型。
现在我们已经看到了使用线程、回调和 OOP 技术来创建程序的方法，让我们考虑使用一个胡氏 T0.出版商订阅方 ORT T1 模型的第三种方法。
发布者-订阅者方法的代码可在chapter12/version3_pubsub文件夹中找到。您将找到四个文件–主程序main.py和三个类定义：LED.py、BUTTON.py和POT.py。
请花点时间停下来阅读main.py中包含的代码，以基本了解程序的结构和工作方式。然后，继续审查LED.py、BUTTON.py和POT.py。
您会注意到，整个程序结构（尤其是类文件）与我们在前面的标题中介绍的version2线程/回调示例非常相似。
您可能也已经意识到，这种方法在概念上与 MQTT 采用的发布者/订阅方法非常相似，我们在第 4 章、与 MQTT、Python 和 Mosquitto MQTT 代理的联网中详细讨论了这一方法。主要区别在于，在我们当前的version3示例中，我们的发布者订阅上下文仅限于我们的程序运行时环境，而不是网络分布式程序集，这是我们 MQTT 示例的场景。
我已经使用PyPubSubPython 库在version3中实现了发布订阅层，该库可从pypi.org获得，并使用pip安装。我们不会详细讨论此库，因为您应该已经熟悉此类库的总体概念和使用，如果不熟悉，我相信您在查看version3源代码文件后（如果您还没有这样做的话）会立即了解发生了什么。
There are alternative PubSub libraries available for Python through PyPi.org. The choice to use PyPubSub for this example was due to the quality of its documentation and the examples provided there. You will find a link to this library in the Technical requirements section at the start of this chapter.
由于version2（线程方法）和version3（发布者-订阅者方法）示例的相似性，我们将不详细讨论每个代码文件，只指出核心区别：
- 在
version2（线程化）中，我们的led、button和pot类实例是如何相互通信的：- 我们在
button和pot类实例的main.py中注册了回调处理程序。 button和pot通过此回调机制发送事件（例如，按钮按下或电位计调整）。- 我们使用
set_rate()实例方法和set_rate_all()类方法直接与 LED 类实例交互。
- 我们在
- 在
version3（发布者-订阅者）中，以下是类内通信结构和设计：- 每个类实例都是松散耦合的。
- 没有回调。
- 在
PyPubSub创建并注册后，我们不会直接与任何类实例进行交互。 - 类和线程之间的所有通信都使用
PyPubSub提供的消息层进行。
现在，老实说，我们的 gizmo 程序并没有从发布者-订阅者方法中受益。我个人的偏好是对像这样的小程序采用回调版本。不过，我已经提供了 PUBLISHER 订户替代实现作为一个参考点，这样您就有了考虑您自己需求的替代方案。
publisher-subscriber 方法在更复杂的程序中大放异彩，在这些程序中，有许多组件（这里我指的是软件组件，不一定是电子组件）需要共享数据，并且可以以异步方式共享数据。
We're presenting the coding and design approaches in this chapter in four very discrete and focused examples. In practice, however, it's common to combine these approaches (and other design patterns) in a hybrid and mixed fashion when creating your programs. Remember, the approach or combination of approaches to use is whatever makes the most sense for what you are trying to achieve.
正如我们刚才所讨论的，您将在查看version3代码时看到，gizmo 程序的发布者-订阅者方法是线程和回调方法的一个简单变体，在这种方法中，我们不使用回调并直接与类实例交互，而是将所有代码通信标准化到消息层。接下来，我们将介绍编写 gizmo 程序的最后一种方法，这次采用 AsyncIO 方法。
到目前为止，在本章中，我们已经看到了实现同一最终目标的三种不同的编程方法。我们的第四种也是最后一种方法将使用 Python3 提供的 AsyncIO 库构建。正如我们将看到的，这种方法与我们以前的方法有相同点和不同点，并且还为我们的代码及其运行方式增加了额外的维度。
根据我自己的经验，这种方法在您第一次使用 Python 进行异步编程时可能会感到复杂、麻烦和混乱。是的，同步编程有一个陡峭的学习曲线（在本节中我们只能勉强触及表面）。然而，当你学会掌握概念并获得实践经验时，你可能会开始发现这是一种优雅的程序创建方式！
If you are new to asynchronous programming in Python, you will find curated tutorial links in the Further reading section to deepen your learning. It is my intention in this section to give you a simple working AsyncIO program that focuses on electronic interfacing, which you can use as a reference as you learn more about this style of programming.
基于异步的方法的代码可以在chapter12/version4_asyncio文件夹中找到。您将找到四个文件–主程序main.py和三个类定义：LED.py、BUTTON.py和POT.py。
请花点时间停下来阅读main.py中包含的代码，以基本了解程序的结构和工作方式。然后继续审查LED.py、BUTTON.py和POT.py。
If you are also a JavaScript developer – particularly Node.js – you will already know that JavaScript is an asynchronous programming language; however, it looks and feels very different from what you are seeing in Python! I can assure you that the principles are the same. Here is a key reason why they feel very different – JavaScript is asynchronous by default. As any experienced Node.js developer knows, we often have to go to (often extreme) lengths in code to make parts of our code behave synchronously. The opposite is true for Python – it's synchronous by default, and we need to extend extra programming effort to make parts of our code behave asynchronously.
在您阅读源代码文件时，我希望您考虑一下我们的version4异步 IO 程序，它包含基于version1事件循环的程序和version2线程化/回调程序的元素。以下是主要差异和相似性的总结：
- 整个程序结构与
version2线程/回调示例非常相似。 - 在
main.py的末尾，我们有几行新的代码，我们以前在这本书中没有见过——例如loop = asyncio.get_event_loop()。 - 像
version2程序一样，我们使用 OOP 技术将组件分解成类，类也有run()方法——但请注意，这些类中没有线程实例，也没有与启动线程相关的代码。 - 在类定义文件
LED.py、BUTTON.py和POT.py中，async和await关键字散布在run()函数中，在while循环中延迟 0 秒，即asyncio.sleep(0)，因为我们根本没有睡觉！ - 在
BUTTON.py中，我们不再使用 PiGPIO 回调来监视被按下的按钮，而是在while循环中轮询按钮 GPIO。
The Python 3 AsyncIO library has evolved significantly over time (and still is evolving), with new API conventions, the addition of higher-level functionality. and deprecated functions. Due to this evolution, code can get out of date with the latest API conventions quickly, and two code examples illustrating the same underlying concepts can be using seemingly different APIs. I highly recommend you glance through the latest Python AsyncIO library API documentation as it will give you hints and examples of newer versus older API practices, which may help you better interpret code examples.
我将以一种简化的方式引导您通过高级程序流程来解释这个程序是如何工作的。当您能够掌握正在发生的事情的一般概念时，您就可以很好地理解 Python 中的异步编程了。
You will also find a file named chapter12/version4_asyncio/main_py37.py. This is a Python 3.7+ version of our program. It uses an API available since Python 3.7. If you look through this file, the differences are clearly commented.
在main.py文件的末尾，我们看到以下代码：
if __name__ == "__main__":
# .... truncated ....
# Get (create) an event loop.
loop = asyncio.get_event_loop() # (1)
# Register the LEDs.
for led in LEDS:
loop.create_task(led.run()) # (2)
# Register Button and Pot
loop.create_task(pot.run()) # (3)
loop.create_task(button.run()) # (4)
# Start the event loop.
loop.run_forever() # (5)Python 中的异步程序围绕事件循环发展。我们看到这是在第 1 行创建的，从第 5 行开始。我们将立即回到第 2、3 和 4 行之间发生的注册。
这个异步事件循环的总体原理类似于我们的 version1 事件循环示例；然而，语义是不同的。这两个版本都是单线程的，并且两组代码都在一个循环中循环。在version1中，这是非常明确的，因为我们的代码主体包含在一个外部while循环中。在我们的异步version4中，它更隐式，并且有一个核心区别——如果编程正确，它是非阻塞*，我们很快就会看到，这就是类run()方法中await asyncio.sleep()调用的目的。*
如前所述，我们已经在第 2、3 和 4 行的循环中注册了 c 类run()方法。在第 5 行上启动事件循环后，以下是简化后的情况：
- 调用第一个LED 的
run()功能（如下代码所示）：
# version4_asyncio/LED.py
async def run(self):
""" Do the blinking """
while True: # (1)
if self.toggle_at > 0 and
(time() >= self.toggle_at): # (2)
self.pi.write(self.gpio, not self.pi.read(self.gpio))
self.toggle_at += self.blink_rate_secs
await asyncio.sleep(0) # (3)- 它进入第 1 行的
while环路，并根据闪烁率从第 2 行切换 LED 的开或关。 - 接下来，它到达第 3 行
await asyncio.sleep(0)，并且产生控制。此时，run()方法有效暂停，另一个while循环迭代不会开始。 - 控制通过秒LED 的
run()功能，并在while回路中运行一次，直到到达await asyncio.sleep(0)。然后产生控制权。 - 现在，pot 实例的
run()方法（如下代码所示）开始运行：
async def run(self):
""" Poll ADC for Voltage Changes """
while True:
# Check if the Potentiometer has been adjusted.
current_value = self.get_value()
if self.last_value != current_value:
if self.callback:
self.callback(self, current_value)
self.last_value = current_value
await asyncio.sleep(0)-
run()方法对while循环进行一次迭代，直到达到await asyncio.sleep(0)。然后产生控制权。 -
控制传递给
button实例的run()方法（以下代码部分显示），该方法有多条await asyncio.sleep(0)语句：
async def run(self):
while True:
level = self.pi.read(self.gpio) # LOW(0) or HIGH(1)
# Waiting for a GPIO level change.
while level == self.__last_level:
await asyncio.sleep(0)
# ... truncated ...
while (time() < hold_timeout_at) and \
not self.pi.read(self.gpio):
await asyncio.sleep(0)
# ... truncated ...
await asyncio.sleep(0)- 一旦按钮的
run()方法到达await asyncio.sleep(0)的任何实例，它就会产生控制。 - 现在，我们所有注册的
run()方法都有机会运行，所以第一个LED 的run()方法将再次控制并执行一次while循环迭代，直到到达await asyncio.sleep(0)。同样，在这一点上，它产生控制，秒LED 的run()方法获得另一轮运行……并且该过程不断重复，每个run()方法以循环方式获得一轮运行。
让我们把一些你可能会有疑问的零散的事情联系起来：
- 按钮的
run()功能及其许多await asyncio.sleep(0)语句如何？
当在任何await asyncio.sleep(0)语句中产生控件时，函数在此点产生。下次run()按钮获得控制时，代码将从生成的await asyncio.sleep(0)语句下面的下一个语句继续。
- 为什么睡眠延迟为 0 秒？
等待零延迟睡眠是产生控制的最简单方式（请注意，它是来自asyncio库的sleep()函数，而不是来自time库的sleep()函数）。但是，您可以await任何异步方法，但这超出了我们简单示例的范围。
为了简单地解释程序的工作原理，我在本例中使用了零秒延迟，但是您可以使用非零延迟。所有这一切都意味着产生的run()函数将在这段时间内休眠—在这段时间到期之前，事件循环不会让它运行。
- 那么
async和await关键字呢？我如何知道在哪里使用它们？
这当然需要实践；但是，以下是基本设计规则：
编写和学习异步程序需要实践和实验。您将面临的初始设计挑战之一是知道将await语句放在何处（以及有多少），以及您应该在多长时间内放弃控制。我鼓励您使用version4代码库，添加您自己的调试print()或日志语句，然后进行实验和修补，直到您感觉到这一切是如何结合在一起的。在某种程度上，您将获得aha时刻，此时，您刚刚打开了进一步探索 Python 异步 IO 库提供的许多高级功能的大门。
现在我们已经了解了异步程序的结构和运行时的行为，我想给你们一些东西来进行实验和思考。
让我们做个实验。也许你想知道version4（AsyncIO）是如何有点像我们的version1（事件循环）代码的，只是它被重构成类，就像version2（线程化）代码一样。那么，难道我们不能将version1 while循环中的代码重构为类，在while循环中创建并调用一个函数（例如，run()），而不去理会所有异步的东西及其额外的库和语法吗？
让我们试试看。您将在chapter12/version5_eventloop2文件夹中找到类似的版本。试着运行这个版本，看看会发生什么。您会发现第一个 LED 闪烁，第二个 LED 始终亮起，按钮和电位计不工作。
你能找出原因吗？
这里有一个简单的答案：在main.py中，一旦第一个 LED 的run()功能被调用，我们将永远陷入其while循环中！
对sleep()（来自time库）的调用不产生控制；它只是在下一个while循环迭代发生之前暂停 LED 的run()方法。
因此，这就是为什么我们说同步程序是阻塞的（不产生控制），以及为什么同步程序是非阻塞的（它们产生控制并给其他代码一个运行的机会）的一个例子。
我希望您喜欢我们探索的四种构建电子接口程序的替代方法——其中一种是我们不应该采用的方法。让我们总结一下我们在本章学到的知识。
在本章中，我们研究了构造与电子设备接口的 Python 程序的四种不同方式。我们学习了一种事件循环编程方法，这是基于线程的方法的两种变体——回调和发布者-订阅者模型，最后我们研究了异步 IO 编程方法的工作原理。
我们讨论的四个例子中的每一个都是非常离散和具体的方法。虽然我们简要讨论了每种方法的相对优点和缺点，但值得记住的是，在实践中，您的项目可能会使用这些（以及潜在的其他）方法的混合，这取决于您试图实现的编程和接口目标。
在下一章中，我们将把注意力转向物联网平台，并讨论可用于构建物联网项目的各种选项和备选方案。
最后，以下是一系列问题，供您测试您对本章内容的了解。您可以在本书的评估部分找到答案：
- 发布者-订阅者模型何时是一种好的设计方法？
- Python GIL 是什么，它对经典线程有什么意义？
- 为什么对于复杂的应用程序来说，纯事件循环通常是一个糟糕的选择？
- 事件循环方法是个坏主意吗？为什么？
thread.join()函数调用的目的是什么？- 您已经使用线程通过模数转换器轮询新的模拟组件。但是，您会发现代码对组件中的更改反应迟缓。有什么问题吗？
- 用 Python 设计 IoT 或电子接口应用程序的最佳方法是使用事件循环、线程/回调、发布者-订阅者模型还是基于异步 IO 的方法？
realpython.com网站提供了一系列优秀的教程，涵盖 Python 中的所有并发性，包括以下内容：
- 什么是 Python GIL？https://realpython.com/python-gil
- 使用并发性加速 Python 程序：https://realpython.com/python-concurrency
- Python 线程简介：https://realpython.com/intro-to-python-threading
- Python 中的异步 IO：完整演练：https://realpython.com/async-io-python
以下是官方 Python（3.7）API 文档中的相关链接：