k10. Потоки, asyncio и очередь сообщений

Чистый код

# Пример: фоновый поток публикует данные, основной поток читает снимок
import threading
import time
from dataclasses import dataclass

@dataclass
class TelemetrySample:
    t_s: float
    altitude_m: float

class TelemetryHub:
    def __init__(self) -＞ None:
        self._lock = threading.Lock()
        self._latest: TelemetrySample | None = None
        self._stop = False

    def publish(self, sample: TelemetrySample) -＞ None:
        with self._lock:
            self._latest = sample

    def snapshot(self) -＞ TelemetrySample | None:
        with self._lock:
            return self._latest

    def request_stop(self) -＞ None:
        self._stop = True

    @property
    def stop(self) -＞ bool:
        return self._stop

def producer(hub: TelemetryHub) -＞ None:
    t = 0.0
    alt = 0.0
    while not hub.stop:
        hub.publish(TelemetrySample(t_s=t, altitude_m=alt))
        t += 0.2
        alt += 1.5
        time.sleep(0.2)

hub = TelemetryHub()
worker = threading.Thread(target=producer, args=(hub,), daemon=True)
worker.start()
time.sleep(0.5)
print(hub.snapshot())
hub.request_stop()
worker.join(timeout=1.0)

• Построчный разбор

  Ниже приведён тот же фрагмент с пояснением после каждой смысловой строки. Строки, которые начинаются с последовательности символов «# →», содержат комментарий преподавателя. Эти строки не являются обязательной частью программы.

  import threading
  # → Модуль стандартной библиотеки для работы с потоками.
  
  import time
  # → Нужен для пауз sleep.
  
  from dataclasses import dataclass
  # → Учебный снимок телеметрии как dataclass.
  
  @dataclass
  # → Декоратор для структуры данных.
  
  class TelemetrySample:
  # → Один отсчёт: время и высота.
  
      t_s: float
  # → Время в секундах.
  
      altitude_m: float
  # → Высота в метрах.
  
  class TelemetryHub:
  # → Общее хранилище последнего отсчёта для двух потоков.
  
      def __init__(self) -＞ None:
  # → Создаём блокировку и пустое состояние.
  
          self._lock = threading.Lock()
  # → Lock защищает чтение и запись _latest от одновременного доступа.
  
          self._latest: TelemetrySample | None = None
  # → Пока данных нет, хранится None.
  
          self._stop = False
  # → Внутренний флаг остановки фонового цикла.
  
      def publish(self, sample: TelemetrySample) -＞ None:
  # → Публикация нового отсчёта из фонового потока.
  
          with self._lock:
  # → Захватываем lock на время записи.
  
              self._latest = sample
  # → Обновляем последний снимок.
  
      def snapshot(self) -＞ TelemetrySample | None:
  # → Чтение снимка из основного потока.
  
          with self._lock:
  # → Та же блокировка, чтобы не прочитать противоречивое состояние.
  
              return self._latest
  # → Возвращаем текущее значение или None.
  
      def request_stop(self) -＞ None:
  # → Публичный способ попросить фоновый цикл завершиться.
  
          self._stop = True
  # → Устанавливаем флаг остановки.
  
      @property
  # → Свойство stop даёт читать флаг без прямого доступа к _stop снаружи.
  
      def stop(self) -＞ bool:
  # → Возвращает True, если работа должна быть остановлена.
  
          return self._stop
  # → Текущее значение флага.
  
  def producer(hub: TelemetryHub) -＞ None:
  # → Функция, которую выполнит фоновый поток.
  
      t = 0.0
  # → Учебное время.
  
      alt = 0.0
  # → Учебная высота.
  
      while not hub.stop:
  # → Цикл, пока основной поток не запросит остановку.
  
          hub.publish(TelemetrySample(t_s=t, altitude_m=alt))
  # → Кладём новый отсчёт в общее хранилище.
  
          t += 0.2
  # → Продвигаем учебное время.
  
          alt += 1.5
  # → Продвигаем учебную высоту.
  
          time.sleep(0.2)
  # → Пауза: имитация периода приёма сообщений.
  
  hub = TelemetryHub()
  # → Создаём общее хранилище.
  
  worker = threading.Thread(target=producer, args=(hub,), daemon=True)
  # → Создаём поток. Аргумент daemon=True означает, что поток не удерживает процесс после выхода main в этом учебном сценарии.
  
  worker.start()
  # → Запускаем фоновую работу.
  
  time.sleep(0.5)
  # → Основной поток коротко ждёт появления данных.
  
  print(hub.snapshot())
  # → Читаем последний снимок телеметрии.
  
  hub.request_stop()
  # → Просим производителя остановиться.
  
  worker.join(timeout=1.0)
  # → Ждём завершения потока не дольше одной секунды.

Чистый код

# Пример: две очереди — события телеметрии и команды
import queue
import threading
import time
from dataclasses import dataclass

@dataclass(frozen=True)
class TelemetryEvent:
    altitude_m: float

@dataclass(frozen=True)
class Command:
    name: str
    detail: str

def reader(events: queue.Queue) -＞ None:
    for alt in (0.5, 8.0, 15.0):
        events.put(TelemetryEvent(altitude_m=alt))
        time.sleep(0.05)
    events.put(None)

def worker(events: queue.Queue, commands: queue.Queue) -＞ None:
    while True:
        item = events.get()
        if item is None:
            commands.put(None)
            break
        if item.altitude_m ＞= 10.0:
            commands.put(Command("HOLD", f"alt={item.altitude_m}"))
        else:
            commands.put(Command("CLIMB", f"alt={item.altitude_m}"))

events: queue.Queue = queue.Queue()
commands: queue.Queue = queue.Queue()
threading.Thread(target=reader, args=(events,), daemon=True).start()
threading.Thread(target=worker, args=(events, commands), daemon=True).start()
while True:
    cmd = commands.get()
    if cmd is None:
        break
    print(cmd)

• Построчный разбор

  Ниже приведён тот же фрагмент с пояснением после каждой смысловой строки. Строки, которые начинаются с последовательности символов «# →», содержат комментарий преподавателя. Эти строки не являются обязательной частью программы.

  import queue
  # → Очередь Queue безопасна для передачи объектов между потоками.
  
  import threading
  # → Запуск потоков reader и worker.
  
  import time
  # → Небольшая пауза в учебном издателе.
  
  from dataclasses import dataclass
  # → Структуры события и команды.
  
  @dataclass(frozen=True)
  # → Неизменяемое событие телеметрии.
  
  class TelemetryEvent:
  # → Сообщение «пришла высота».
  
      altitude_m: float
  # → Высота в метрах.
  
  @dataclass(frozen=True)
  # → Неизменяемая команда-решение.
  
  class Command:
  # → Учебная команда для печати (не MAVLink).
  
      name: str
  # → Имя команды, например HOLD или CLIMB.
  
      detail: str
  # → Пояснение для журнала.
  
  def reader(events: queue.Queue) -＞ None:
  # → Поток-издатель кладёт события в очередь events.
  
      for alt in (0.5, 8.0, 15.0):
  # → Три учебных отсчёта высоты.
  
          events.put(TelemetryEvent(altitude_m=alt))
  # → put добавляет элемент в конец очереди.
  
          time.sleep(0.05)
  # → Короткая пауза между событиями.
  
      events.put(None)
  # → Специальный маркер конца потока данных.
  
  def worker(events: queue.Queue, commands: queue.Queue) -＞ None:
  # → Поток-обработчик читает события и пишет команды.
  
      while True:
  # → Цикл до маркера конца.
  
          item = events.get()
  # → get ждёт и забирает следующий элемент из очереди.
  
          if item is None:
  # → Получен маркер конца.
  
              commands.put(None)
  # → Передаём конец дальше по конвейеру.
  
              break
  # → Выходим из цикла обработчика.
  
          if item.altitude_m ＞= 10.0:
  # → Учебное правило принятия решения.
  
              commands.put(Command("HOLD", f"alt={item.altitude_m}"))
  # → Высота достаточна — команда удержания.
  
          else:
  # → Иначе — набор высоты.
  
              commands.put(Command("CLIMB", f"alt={item.altitude_m}"))
  # → Команда набора.
  
  events: queue.Queue = queue.Queue()
  # → Очередь событий телеметрии.
  
  commands: queue.Queue = queue.Queue()
  # → Очередь команд-решений.
  
  threading.Thread(target=reader, args=(events,), daemon=True).start()
  # → Запуск издателя.
  
  threading.Thread(target=worker, args=(events, commands), daemon=True).start()
  # → Запуск обработчика.
  
  while True:
  # → Основной поток читает готовые команды.
  
      cmd = commands.get()
  # → Ожидание следующей команды.
  
      if cmd is None:
  # → Конец конвейера.
  
          break
  # → Выход.
  
      print(cmd)
  # → Печать учебной команды.

Чистый код

# Пример: две async-задачи в одном цикле событий
import asyncio
from dataclasses import dataclass

@dataclass
class FakeLink:
    ticks: int = 0

async def telemetry_ticker(link: FakeLink, n: int) -＞ None:
    for i in range(n):
        await asyncio.sleep(0.1)
        link.ticks += 1
        print(f"telemetry tick {i + 1}, total={link.ticks}")

async def status_printer(link: FakeLink, n: int) -＞ None:
    for i in range(n):
        await asyncio.sleep(0.15)
        print(f"status {i + 1}: ticks_seen={link.ticks}")

async def main() -＞ None:
    link = FakeLink()
    await asyncio.gather(
        telemetry_ticker(link, 5),
        status_printer(link, 4),
    )
    print("asyncio done, ticks=", link.ticks)

asyncio.run(main())

• Построчный разбор

  Ниже приведён тот же фрагмент с пояснением после каждой смысловой строки. Строки, которые начинаются с последовательности символов «# →», содержат комментарий преподавателя. Эти строки не являются обязательной частью программы.

  import asyncio
  # → Стандартный модуль асинхронного ввода-вывода и кооперативных задач.
  
  from dataclasses import dataclass
  # → Простое состояние канала.
  
  @dataclass
  # → Учебная структура.
  
  class FakeLink:
  # → Имитация состояния: сколько тиков телеметрии уже «принято».
  
      ticks: int = 0
  # → Счётчик тиков.
  
  async def telemetry_ticker(link: FakeLink, n: int) -＞ None:
  # → Корутина: async def означает, что внутри можно await.
  
      for i in range(n):
  # → n учебных тиков.
  
          await asyncio.sleep(0.1)
  # → await отдаёт управление циклу событий на время паузы, не блокируя весь процесс так, как это делает обычный time.sleep в однозадачном коде с несколькими задачами.
  
          link.ticks += 1
  # → Увеличиваем счётчик «принятых» тиков.
  
          print(f"telemetry tick {i + 1}, total={link.ticks}")
  # → Печать хода приёма.
  
  async def status_printer(link: FakeLink, n: int) -＞ None:
  # → Вторая корутина печатает сводку.
  
      for i in range(n):
  # → n сводок.
  
          await asyncio.sleep(0.15)
  # → Другая длительность паузы — задачи чередуются.
  
          print(f"status {i + 1}: ticks_seen={link.ticks}")
  # → Видим, сколько тиков успело накопиться.
  
  async def main() -＞ None:
  # → Точка сборки асинхронных задач.
  
      link = FakeLink()
  # → Общее состояние для обеих задач.
  
      await asyncio.gather(
  # → gather запускает несколько корутин и ждёт завершения всех.
  
          telemetry_ticker(link, 5),
  # → Первая задача.
  
          status_printer(link, 4),
  # → Вторая задача.
  
      )
  # → Конец gather.
  
      print("asyncio done, ticks=", link.ticks)
  # → Итоговый счётчик.
  
  asyncio.run(main())
  # → Создаёт цикл событий, выполняет main и корректно завершает цикл.

Чистый код

# Пример: queue.get с timeout
import queue
import threading
import time

def slow_publisher(q: queue.Queue) -＞ None:
    time.sleep(0.4)
    q.put({"type": "HEARTBEAT", "ok": True})

q: queue.Queue = queue.Queue()
threading.Thread(target=slow_publisher, args=(q,), daemon=True).start()

try:
    msg = q.get(timeout=0.2)
    print("early:", msg)
except queue.Empty:
    print("timeout 0.2s: сообщения ещё нет")

msg = q.get(timeout=0.5)
print("received:", msg)

• Построчный разбор

  Ниже приведён тот же фрагмент с пояснением после каждой смысловой строки. Строки, которые начинаются с последовательности символов «# →», содержат комментарий преподавателя. Эти строки не являются обязательной частью программы.

  import queue
  # → Очередь и исключение queue.Empty.
  
  import threading
  # → Фоновая публикация сообщения.
  
  import time
  # → Задержка издателя.
  
  def slow_publisher(q: queue.Queue) -＞ None:
  # → Издатель кладёт сообщение не сразу.
  
      time.sleep(0.4)
  # → Искусственная задержка 0.4 с.
  
      q.put({"type": "HEARTBEAT", "ok": True})
  # → Учебное сообщение, похожее по смыслу на heartbeat.
  
  q: queue.Queue = queue.Queue()
  # → Пустая очередь.
  
  threading.Thread(target=slow_publisher, args=(q,), daemon=True).start()
  # → Запуск медленного издателя.
  
  try:
  # → Первая попытка чтения с коротким таймаутом.
  
      msg = q.get(timeout=0.2)
  # → Ждать элемент не дольше 0.2 с.
  
      print("early:", msg)
  # → Если успели — напечатать (в этом сценарии обычно не успеваем).
  
  except queue.Empty:
  # → Если за 0.2 с ничего не появилось, get поднимает queue.Empty.
  
      print("timeout 0.2s: сообщения ещё нет")
  # → Ожидаемая ветка для короткого таймаута.
  
  msg = q.get(timeout=0.5)
  # → Вторая попытка с большим запасом времени.
  
  print("received:", msg)
  # → К этому моменту сообщение уже должно быть в очереди.