Введение
Консольный производитель продемонстрировал модель событий и ключей. Приложению нужно учитывать ещё одно различие: вызов API отправки может поместить запись в локальную очередь клиента до подтверждения брокером. Программа, завершившаяся в этот момент, может сообщить об успехе, не доставив записи.
Вы воспользуетесь установленным Python-клиентом confluent-kafka с подготовленным локальным брокером Apache Kafka. Этот пакет — клиентская библиотека; учётная запись Confluent Cloud не нужна. Вы сериализуете словари в байты JSON/UTF-8, отправите ключи заказов и изучите обратные вызовы доставки и конечное состояние очереди. Недоступный локальный адрес позволит проверить ошибку с ограниченным временем ожидания.
В этой независимой работе подготовлены топик с тремя разделами и два набора входных данных. Последующие работы курса предполагают знание функций, циклов, словарей, JSON и запуска скриптов Python; с подготовкой поможет курс Python. Установка Kafka и настройка сервисов уже выполнены. Используйте Terminal.
Преобразуйте словарь в ключ и значение для передачи
На этом шаге вы изучите преобразование до обращения к Kafka. Начните с входного файла из шести событий:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Каждая строка — полный объект JSON. json.loads преобразует этот текст в словарь Python. json.dumps преобразует словарь в текст JSON, а encode("utf-8") преобразует текст в байты, подходящие для клиентского API. Ключ кодируется отдельно из order_id; он не является event_id.
Сохраните и запустите этот небольшой пример для просмотра. Ограничитель here-document в кавычках сохраняет исходный код Python буквально:
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Найдите словарь, текст JSON и байтовые представления. Отображение bytes в Python начинается с b; это обозначение вывода не входит в передаваемое значение. Первый ключ должен кодировать order-4001. JSON сохраняет именованные поля события, а UTF-8 задаёт кодировку символов.
Пока ни одно событие не отправлено. На следующем шаге эти байты будут переданы производителю, а вы проверите подтверждение брокером.
Опубликуйте события и подтвердите каждую фактическую доставку
На этом шаге вы запустите полного производителя с предоставленной обработкой аргументов и обратным вызовом доставки. Producer настраивает клиент; bootstrap.servers задаёт начальный адрес. acks=all запрашивает подтверждение от текущих синхронизированных реплик. В этой среде с одним брокером одна реплика, поэтому такая настройка не демонстрирует отказоустойчивость нескольких брокеров. Предоставленная настройка маршрутизации murmur2_random сохраняет совместимость маршрутизации по ключам с предыдущими примерами консоли Java при неизменном числе разделов.
Обратите внимание на ключевое различие в программе: produce помещает запись в локальную очередь. poll обрабатывает обратные вызовы, а flush ждёт выполнения работы из очереди, также обрабатывая обратные вызовы.
Обратный вызов — это функция, которую вы передаёте клиентской библиотеке, чтобы та вызвала её при появлении результата доставки. Запись callback=delivery передаёт саму функцию; она не вызывает её немедленно. poll(0) обрабатывает доступные обратные вызовы без ожидания, а flush также ждёт незавершённых отправок. Обратный вызов сохраняет либо подтверждённые брокером топик, раздел и смещение, либо ошибку. После flush программа проверяет оставшиеся записи, ошибки и число подтверждений, прежде чем объявить успех. Ошибка обратного вызова сохраняется и приводит к ненулевому коду завершения процесса; она не просто выводится и игнорируется.
Предоставленный раздел argparse сопоставляет параметры командной строки с атрибутами args: например, --input становится args.input, а --delivery-timeout-ms — args.delivery_timeout_ms. Path(args.input).open(...) открывает выбранный локальный файл; цикл читает по одному событию JSON из каждой непустой строки. Чтобы пользоваться этим клиентом, не нужно проектировать разбор аргументов.
Сохраните полную программу. Структура разбора аргументов и обратного вызова предоставлена, а сериализация и вызов produce используют байты, которые вы только что изучили:
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Запустите производителя и сохраните оба потока вывода. > направляет стандартный вывод в журнал; 2>&1 направляет стандартный поток ошибок туда же, поэтому ошибки обратных вызовов и диагностика соединения тоже сохраняются.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Ожидаются шесть строк обратных вызовов DELIVERED и Delivered records: 6. Фактические раздел и смещение каждого обратного вызова могут различаться. Один лишь успешный возврат из produce не подтверждает получение этих подтверждений.
Независимо проверьте брокер уже знакомым консольным потребителем:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Найдите события от py1 до py6 с ключами, совпадающими с их значениями order_id. Независимое чтение связывает отчёт приложения о доставке с настоящими записями брокера; конкретный номер раздела или общий порядок вывода не требуются.
Измените одно поле нового события и снова используйте производителя
На этом шаге вы повторно используете то же приложение для одного нового события. В подготовленных данных next-event новый event_id, py7, поэтому это событие ещё не опубликовано. Просмотрите его поле channel:
cat /home/labex/project/data/next-event.jsonl
В отдельном рабочем файле измените только channel этого неопубликованного события с online на store. Событие по-прежнему относится к order-4001 и следует за его предыдущими событиями с sequence 3:
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Ожидаются один новый успешный обратный вызов и семь сохранённых событий при независимом чтении. У py7 должно быть channel store, а события от py1 до py6 сохраняют исходное содержимое. Связанные события order-4001 следуют своей последовательности внутри одного раздела, выбранного маршрутизацией; порядок вывода записей из разных разделов может различаться.
Реализация производителя не изменилась. Указанный путь входного файла позволил ему сериализовать и доставить новые данные вместо встраивания фиксированного учебного ответа.
Проверьте ошибку с ограниченным ожиданием по недоступному адресу
На этом шаге вы проверите, что постановка в очередь не принимается за доставку. Тот же клиент может создать локальную очередь, даже когда по адресу localhost:1 нет брокера Kafka. Предоставленный параметр тайм-аута доставки ограничивает ожидание результата сообщения до 3000 миллисекунд, а flush добавляет небольшой ограниченный запас времени. Настоящий брокер остаётся по адресу localhost:9092.
Отправьте файл с одним событием на этот заведомо неверный адрес. Следующая команда должна завершиться с ошибкой.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Немедленно сохраните код завершения, пока другая команда его не заменила. $? — код завершения предыдущей команды: ноль при успехе и ненулевое значение при ошибке. printf сохраняет его как строку текста:
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Ожидаются ненулевой код завершения, диагностика соединения, обратный вызов DELIVERY_FAILED и Delivery incomplete вместо успешного итогового числа. Программа проверила ошибки обратных вызовов, даже когда очередь опустела после неудачных отправок сообщений.
Отправка на недоступный адрес не должна была добавить второй py7 в настоящий топик. Все семь ранее успешно отправленных событий остаются доступными. Исправление адреса и повторная отправка бизнес-события — отдельное решение, а не следствие предположения, что запись в очереди уже подтверждена; намеренная ошибка в этой работе ничего не отправила настоящему брокеру. Далее вы декодируете полученный JSON, сохраните его и зафиксируете позицию чтения в потребителе на Python.
Итоги
Вы сериализовали словари и ключи заказов в байты UTF-8, опубликовали их настоящим Python-клиентом и сопоставили успешные обратные вызовы с фактическими позициями брокера. Изменённое новое событие было отправлено той же программой, а недоступный адрес привёл к ненулевому коду ошибки с ограниченным ожиданием вместо ложного успеха. Далее потребитель на Python явно покажет декодирование, вывод в файл и запись перед фиксацией смещения.
