В этой практической работе мы реализуем проект, который позволит пользователям конвертировать видеофайлы в GIF. Такая задача хорошо подходит для Cloud Functions, потому что конвертирование отнимает немало ресурсов процессора, и чем качественнее видео, тем больше ресурсов требуется на его обработку.

Представим, что мы попытались решить эту задачу «в лоб». Пользователь заходит на страницу и вводит ссылку на видеофайл. Сервис скачивает его, конвертирует и отдает ссылку на GIF. Возникают две серьёзные проблемы:

Чтобы избежать этих проблем, в архитектуру сервиса необходимо встроить очередь.

Создайте сервисный аккаунт с именем ffmpeg-account-for-cf:

export SERVICE_ACCOUNT=$(yc iam service-account create --name ffmpeg-account-for-cf \
  --description "service account for serverless" \
  --format json | jq -r .)

Проверьте текущий список сервисных аккаунтов:

yc iam service-account list

После проверки запишите ID созданного сервисного аккаунта в переменную SERVICE_ACCOUNT_ID:

echo "export SERVICE_ACCOUNT_FFMPEG_ID=<ID>" >> ~/.bashrc && . ~/.bashrc
echo $SERVICE_ACCOUNT_FFMPEG_ID

 

Добавим вновь созданному сервисному аккаунту роли storage.viewer, storage.uploader, ymq.reader, ymq.writer, ydb.admin, serverless.functions.invoker, и lockbox.payloadViewer:

echo "export FOLDER_ID=$(yc config get folder-id)" >> ~/.bashrc && . ~/.bashrc
echo $FOLDER_ID

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role storage.viewer

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role storage.uploader

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role ymq.reader

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role ymq.writer

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role ydb.admin

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role serverless.functions.invoker

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role lockbox.payloadViewer

yc resource-manager folder add-access-binding $FOLDER_ID \
  --subject serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --role editor

Вы можете назначить несколько ролей и с помощью команды set-access-binding. Но эта команда полностью перезаписывает права доступа к ресурсу и все текущие роли на него будут удалены! Поэтому сначала убедитесь, что ресурсу не назначены роли, которые вы не хотите потерять:

yc resource-manager folder list-access-bindings $FOLDER_ID

yc resource-manager folder set-access-bindings $FOLDER_ID \
  --access-binding role=storage.viewer,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=storage.uploader,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=ymq.reader,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=ymq.writer,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=ydb.admin,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=serverless.functions.invoker,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=lockbox.payloadViewer,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID \
  --access-binding role=editor,subject=serviceAccount:$SERVICE_ACCOUNT_FFMPEG_ID

 

Этот этап нужен для получения идентификатора ключа доступа и секретного ключа, которые будут использованы для загрузки файлов в Object Storage, работы с Yandex Message Queue и т. д. Для создания ключа доступа необходимо вызвать следующую команду:

yc iam access-key create --service-account-name ffmpeg-account-for-cf

В результате вы получите примерно следующее:

    access_key:
        id: ajefraollq5puj2tir1o
        service_account_id: ajetdv28pl0a1a8r41f0
        created_at: "2021-08-23T21:13:05.677319393Z"
        key_id: BTPNvWthv0ZX2xVmlPIU
    secret: cWLQ0HrTM0k_qAac43cwMNJA8VV_rfTg_kd4xVPi 

Здесь key_id — это идентификатор ключа доступа ACCESS_KEY_ID. А secret — это секретный ключ SECRET_ACCESS_KEY. Переменные ACCESS_KEY_ID и SECRET_ACCESS_KEY могут быть использованы для задания соответствующих значений aws_access_key_id и aws_secret_access_key при использовании библиотеки boto3.

 

В сервисе Lockbox (находится на стадии Preview) создайте ваш первый секрет, состоящий из набора версий, в которых хранятся ваши данные. Версия содержит наборы ключей и значений:

Версия не изменяется. Для любого изменения количества пар ключей-значений или их содержимого необходимо создать новую версию. Создадим секрет с именем ffmpeg-sa-key и парой ключей ACCESS_KEY_ID и SECRET_ACCESS_KEY:

yc lockbox secret create --name ffmpeg-sa-key \
  --folder-id $FOLDER_ID \
  --description "keys for serverless" \
  --payload '[{"key": "ACCESS_KEY_ID", "text_value": <ACCESS_KEY_ID>}, {"key": "SECRET_ACCESS_KEY", "text_value": "<SECRET_ACCESS_KEY>"}]'

Получим и запишем значение SECRET_ID, оно нам потребуется при создании функции:

yc lockbox secret list

yc lockbox secret get --name ffmpeg-sa-key

echo "export SECRET_ID=<SECRET_ID>" >> ~/.bashrc && . ~/.bashrc
echo $SECRET_ID

 

Для создания очереди Yandex Message Queue вы можете использовать три разных способа:

Воспользуемся AWS CLI. Для начала задайте конфигурацию с помощью команды aws configure. При этом от вас потребуется ввести:

По завершению конфигурации вы сможете создать очередь:

aws configure
aws sqs create-queue --queue-name ffmpeg --endpoint https://message-queue.api.cloud.yandex.net/

В результате успешного выполнения предыдущей команды в ответ вы получите URL:

    {
        "QueueUrl": "https://message-queue.api.cloud.yandex.net/b1ga4gj7agij03ln6aov/dj6000000003kv2t02b3/ffmpeg"
    } 

Запишем значения URL в переменную YMQ_QUEUE_URL. Она потребуется нам при создании функции:

echo "export YMQ_QUEUE_URL=<YMQ_QUEUE_URL>" >> ~/.bashrc && . ~/.bashrc
echo $YMQ_QUEUE_URL

Ещё вам потребует значение атрибута QueueArn, получим его:

aws sqs get-queue-attributes \
  --endpoint https://message-queue.api.cloud.yandex.net \
  --queue-url $YMQ_QUEUE_URL \
  --attribute-names QueueArn

В результате вы получите ответ вида:

    {
        "Attributes": {
            "QueueArn": "yrn:yc:ymq:ru-central1:b1gl21bkgss4msekt08i:ffmpeg"
        }
    } 

Сохраним значение QueueArn в переменную YMQ_QUEUE_ARN:

echo "export YMQ_QUEUE_ARN=<YMQ_QUEUE_ARN>" >> ~/.bashrc && . ~/.bashrc
echo $YMQ_QUEUE_ARN

Создадим базу данных YDB с именем ffmpeg и типом serverless, используя для этого флаг --serverless:

yc ydb database create ffmpeg \
  --serverless \
  --folder-id $FOLDER_ID

yc ydb database list

Сразу получим и сохраним document_api_endpoint в значение переменной DOCAPI_ENDPOINT:

yc ydb database get --name ffmpeg
echo "export DOCAPI_ENDPOINT=<DOCAPI_ENDPOINT>" >> ~/.bashrc && . ~/.bashrc
echo $DOCAPI_ENDPOINT

Как только база данных создана, воспользуемся ранее использованной утилитой AWS CLI для создания документной таблицы в этой базе данных. Всю конфигурацию возьмем из файла tasks.json:

{
  "AttributeDefinitions": [
    {
      "AttributeName": "task_id",
      "AttributeType": "S"
    }
  ],
  "KeySchema": [
    {
      "AttributeName": "task_id",
      "KeyType": "HASH"
    }
  ],
  "TableName": "tasks"
}

Находясь в одном каталоге с файлом tasks.json, вызовите следующую команду для создания таблицы:

aws dynamodb create-table \
  --cli-input-json file://tasks.json \
  --endpoint-url $DOCAPI_ENDPOINT \
  --region ru-central1

В консоли управления убедитесь, что БД ffmpeg создана, и в ней есть пустая таблица tasks.

Самый простой способ создания бакета в Object Storage — это использование консоли управления.

В консоли управления в вашем рабочем каталоге выберите сервис Object Storage. Нажмите кнопку Создать бакет. На странице создания бакета введите имя, в нашем примере это будет storage-for-ffmpeg, остальные параметры не меняйте.

Нажмите кнопку Создать бакет для завершения операции. Далее вы всегда сможете поменять класс хранилища, его размер и настройки доступа.

Сохраним название бакета для дальнейшего использования:

echo "export S3_BUCKET=<имя бакета>" >> ~/.bashrc && . ~/.bashrc
echo $S3_BUCKET

При создании функций нам потребуется ряд переменных:

Проверим заданные ранее переменные:

echo $SERVICE_ACCOUNT_FFMPEG_ID
echo $SECRET_ID
echo $YMQ_QUEUE_URL
echo $DOCAPI_ENDPOINT
echo $S3_BUCKET

Для обработки видео понадобится утилита FFmpeg. Скачайте статический релизный бинарный файл для Linux amd64 на сайте ffmpeg.org (обычно он находится в разделе FFmpeg Static Builds и называется примерно так: ffmpeg-release-amd64-static.tar.xz). Извлеките из архива файл ffmpeg. Обратите внимание, что у этого файла должны быть заданы права доступа на выполнение (установить нужный флаг можно с помощью команды chmod a+x ffmpeg).

Поскольку через консоль управления можно прикладывать файлы размером не более 3,5 МБ, загрузим код функций и файл ffmpeg в объектное хранилище (Object Storage).

Исходный код в файле index.py содержит обе необходимые нам функции:

import json
import os
import subprocess
import uuid
from urllib.parse import urlencode

import boto3
import requests
import yandexcloud
from yandex.cloud.lockbox.v1.payload_service_pb2 import GetPayloadRequest
from yandex.cloud.lockbox.v1.payload_service_pb2_grpc import PayloadServiceStub

boto_session = None
storage_client = None
docapi_table = None
ymq_queue = None


def get_boto_session():
    global boto_session
    if boto_session is not None:
        return boto_session

    # initialize lockbox and read secret value
    yc_sdk = yandexcloud.SDK()
    channel = yc_sdk._channels.channel("lockbox-payload")
    lockbox = PayloadServiceStub(channel)
    response = lockbox.Get(GetPayloadRequest(secret_id=os.environ['SECRET_ID']))

    # extract values from secret
    access_key = None
    secret_key = None
    for entry in response.entries:
        if entry.key == 'ACCESS_KEY_ID':
            access_key = entry.text_value
        elif entry.key == 'SECRET_ACCESS_KEY':
            secret_key = entry.text_value
    if access_key is None or secret_key is None:
        raise Exception("secrets required")
    print("Key id: " + access_key)

    # initialize boto session
    boto_session = boto3.session.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key
    )
    return boto_session


def get_ymq_queue():
    global ymq_queue
    if ymq_queue is not None:
        return ymq_queue

    ymq_queue = get_boto_session().resource(
        service_name='sqs',
        endpoint_url='https://message-queue.api.cloud.yandex.net',
        region_name='ru-central1'
    ).Queue(os.environ['YMQ_QUEUE_URL'])
    return ymq_queue


def get_docapi_table():
    global docapi_table
    if docapi_table is not None:
        return docapi_table

    docapi_table = get_boto_session().resource(
        'dynamodb',
        endpoint_url=os.environ['DOCAPI_ENDPOINT'],
        region_name='ru-central1'
    ).Table('tasks')
    return docapi_table


def get_storage_client():
    global storage_client
    if storage_client is not None:
        return storage_client

    storage_client = get_boto_session().client(
        service_name='s3',
        endpoint_url='https://storage.yandexcloud.net',
        region_name='ru-central1'
    )
    return storage_client

# API handler

def create_task(src_url):
    task_id = str(uuid.uuid4())
    get_docapi_table().put_item(Item={
        'task_id': task_id,
        'ready': False
    })
    get_ymq_queue().send_message(MessageBody=json.dumps({'task_id': task_id, "src": src_url}))
    return {
        'task_id': task_id
    }


def get_task_status(task_id):
    task = get_docapi_table().get_item(Key={
        "task_id": task_id
    })
    if task['Item']['ready']:
        return {
            'ready': True,
            'gif_url': task['Item']['gif_url']
        }
    return {'ready': False}


def handle_api(event, context):
    action = event['action']
    if action == 'convert':
        return create_task(event['src_url'])
    elif action == 'get_task_status':
        return get_task_status(event['task_id'])
    else:
        return {"error": "unknown action: " + action}

# Converter handler

def download_from_ya_disk(public_key, dst):
    api_call_url = 'https://cloud-api.yandex.net/v1/disk/public/resources/download?' + \
                   urlencode(dict(public_key=public_key))
    response = requests.get(api_call_url)
    download_url = response.json()['href']
    download_response = requests.get(download_url)
    with open(dst, 'wb') as video_file:
        video_file.write(download_response.content)


def upload_and_presign(file_path, object_name):
    client = get_storage_client()
    bucket = os.environ['S3_BUCKET']
    client.upload_file(file_path, bucket, object_name)
    return client.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': object_name}, ExpiresIn=3600)


def handle_process_event(event, context):
    for message in event['messages']:
        task_json = json.loads(message['details']['message']['body'])
        task_id = task_json['task_id']
        # Download video
        download_from_ya_disk(task_json['src'], '/tmp/video.mp4')
        # Convert with ffmpeg
        subprocess.run(['ffmpeg', '-i', '/tmp/video.mp4', '-r', '10', '-s', '320x240', '/tmp/result.gif'])
        result_object = task_id + ".gif"
        # Upload to Object Storage and generate presigned url
        result_download_url = upload_and_presign('/tmp/result.gif', result_object)
        # Update task status in DocAPI
        get_docapi_table().update_item(
            Key={'task_id': task_id},
            AttributeUpdates={
                'ready': {'Value': True, 'Action': 'PUT'},
                'gif_url': {'Value': result_download_url, 'Action': 'PUT'},
            }
        )
    return "OK"

Сгенерируйте файл requirements.txt:

pipreqs $PWD --force

Находясь в директории с исходными файлами, упакуем все нужные файлы в ZIP-архив.

zip src.zip index.py requirements.txt ffmpeg

В Object Storage для простоты используем тот же бакет, куда далее будем складывать видео. На вкладке Объекты, вверху справа нажмите кнопку Загрузить и выберите созданный архив.

Создадим функции ffmpeg-api и ffmpeg-converter, при этом сразу зададим все необходимые переменные и сервисный аккаунт:

yc serverless function create \
  --name ffmpeg-api \
  --description "function for ffmpeg-api"

yc serverless function create \
  --name ffmpeg-converter \
  --description "function for ffmpeg-converter"

yc serverless function version create \
  --function-name ffmpeg-api \
  --memory=256m \
  --execution-timeout=5s \
  --runtime=python37 \
  --entrypoint=index.handle_api \
  --service-account-id $SERVICE_ACCOUNT_FFMPEG_ID \
  --environment SECRET_ID=$SECRET_ID \
  --environment YMQ_QUEUE_URL=$YMQ_QUEUE_URL \
  --environment DOCAPI_ENDPOINT=$DOCAPI_ENDPOINT \
  --package-bucket-name $S3_BUCKET \
  --package-object-name src.zip

yc serverless function version create \
  --function-name ffmpeg-converter \
  --memory=2048m \
  --execution-timeout=600s \
  --runtime=python37 \
  --entrypoint=index.handle_process_event \
  --service-account-id $SERVICE_ACCOUNT_FFMPEG_ID \
  --environment SECRET_ID=$SECRET_ID \
  --environment YMQ_QUEUE_URL=$YMQ_QUEUE_URL \
  --environment DOCAPI_ENDPOINT=$DOCAPI_ENDPOINT \
  --environment S3_BUCKET=$S3_BUCKET \
  --package-bucket-name $S3_BUCKET \
  --package-object-name src.zip

В консоли управления перейдите из рабочего каталога в раздел Cloud Functions и выберите ранее созданную функцию ffmpeg-api. Перейдите на вкладку Тестирование в боковом меню, выберите шаблон данных Без шаблона и добавьте во вводные данные JSON:

{"action":"convert", "src_url":"https://disk.yandex.ru/i/38RbVC0spb_jQQ"}

Нажмите кнопку Запустить тест. Этим самым мы загрузим файл в хранилище и создадим задачу в БД. Если всё сделано правильно, то вы увидите такой результат:

    {
        "task_id": "133e05c2-1b98-41cc-9aab-b816d71af343"
    } 

Воспользуемся полученным идентификатором задачи task_id для получения статуса из базы данных. Для этого внесите в вводные данные JSON следующие изменения:

{"action":"get_task_status", "task_id":"<идентификатор задачи>"}

Нажмите кнопку Запустить тест. Так как мы ещё не обрабатывали задачи в очереди, результат очевиден:

    {
        "ready": false
    } 

Теперь создайте триггер, который будет вызывать функцию обработки сообщений из очереди. После создания триггер начинает работать через пять минут. Он будет брать по одному сообщению и раз в 10 секунд отправлять в функцию:

yc serverless trigger create message-queue \
  --name ffmpeg \
  --queue $YMQ_QUEUE_ARN \
  --queue-service-account-id $SERVICE_ACCOUNT_FFMPEG_ID \
  --invoke-function-name ffmpeg-converter  \
  --invoke-function-service-account-id $SERVICE_ACCOUNT_FFMPEG_ID \
  --batch-size 1 \
  --batch-cutoff 10s

С этого момента очередь начнёт обрабатываться. Можно проверить, готова ли задача, и, если это так, запросить по URL результат обработки из Object Storage.

Теперь у нас есть функция, которая выполняет функцию API, через которую мы можем ставить задачи в очередь на исполнение. Триггер раз в 10 секунд берет по одному сообщению в очереди и передает функции обработчику. Функция-обработчик формирует результат и обновляет данные в базе данных. При этом мы получаем сконвертированные GIF-файлы из видео.

Протестируйте систему, используя полученный ранее идентификатор задачи task_id для получения статуса из базы данных. Для этого внесите изменения в вводные данные JSON:

{"action":"get_task_status", "task_id":"133e05c2-1b98-41cc-9aab-b816d71af343"}

Нажмите кнопку Запустить тест. Если задача уже успела обработаться, то вы получите URL.

По завершении работы не забудьте удалить созданный триггер ffmpeg, иначе он будет продолжать работать:

yc serverless trigger delete ffmpeg

Не забудьте также удалить или остановить все созданные вами ресурсы.

 

На прошлом уроке вы узнали, что такое очереди и зачем они нужны. На этом уроке вы разберётесь со спецификой реализации очередей в Yandex Message Queue и их параметрами.

Основным понятием для очередей в Yandex Message Queue (YMQ) является сообщение. Сообщение состоит из тела — ваших данных — и метаданных, его дополнительных атрибутов.

Тело сообщения обрабатывается вашим приложением, а метаданные удобно использовать для других самых разнообразных целей: например, для подтверждения обработки сообщения, оценки времени его обработки или чтобы удостовериться, что оно было передано без изменений.

Представим, что мы считали это простое сообщение из очереди, давайте рассмотрим его содержимое:

Задача YMQ — организовать очередь между приложениями-отправителями и приложениями получателями сообщений. Чтобы отправить и получить сообщение, отправители и получатели должны обращаться к сервису сами.

Каждое сообщение за время жизни проходит следующие этапы:

Сервис YMQ поддерживает два типа очередей — стандартные и FIFO. На предыдущем уроке мы уже разобрали их основные отличия. Давайте остановимся теперь на них подробнее.

Стандартные очереди позволяют сохранять сообщения, которые затем читаются приложениями в произвольном порядке. Такой подход упрощает систему обработки сообщений. Приложения должны быть рассчитаны на ситуации приёма данных не в хронологическом порядке их поступления.

Стандартные очереди обеспечивают гарантию, что каждое сообщение будет доставлено до получателя хотя бы один раз. В исключительных случаях данные могут быть доставлены до считывающего приложения несколько раз. Обрабатывающие системы должны быть готовы к подобным ситуациям. Такие очереди лучше подходят для обработки не связанных между собой сообщений и обеспечивают более высокую пропускную способность, то есть работают быстрее.

Очереди FIFO позволяют обеспечить строгую очерёдность выдачи сообщений запрашивающей/обрабатывающей стороне и обеспечивают семантику строгой однократной гарантированной доставки сообщений. Такие очереди подходят для передачи связанных сообщений и работают медленнее из-за того, что сообщения должны быть обработаны по очереди.

FIFO очереди часто используются для обработки финансовых данных. Представьте, что в одну FIFO очередь отправляются действия с банковскими счетами разных пользователей. Данные из очереди обрабатывают несколько получателей, и было бы удобно, чтобы все действия одного пользователя попадали одному получателю. Для такой цели можно использовать группировку сообщений.

При помощи специальных идентификаторов группы можно обеспечить отправку сразу нескольких потоков упорядоченных сообщений для разных получателей сообщений в рамках одной очереди FIFO. Вместо одной очереди FIFO получается несколько очередей по количеству групп. В каждой группе запись сообщений и их считывание происходит по схеме FIFO.

Давайте разберём, какие параметры бывают у очередей в Yandex Message Queue.

В Базовых параметрах помимо имени и типа очереди вы можете указать:

В блоке Настройки очередей недоставленных сообщений вы можете настроить работу с так называемой Dead Letter Queue (DLQ, дословно — очередь невостребованных писем). Это специальная очередь, куда могут перенаправляться сообщения, которые получатели не смогли обработать в обычных очередях. Собирая такие сообщения в отдельной очереди, вы можете исследовать ошибки, возникающие при их обработке.

Чтобы воспользоваться этой функцией, вам придется сначала завести отдельную очередь того же типа, что и очередь, откуда перенаправляются сообщения, которые дошли до адресата, но не были обработаны. Включите функцию Перенаправлять недоставленные сообщения, выберите заранее созданную DLQ, а затем укажите количество попыток, после которых необработанное сообщение направляется в эту очередь.

Помещение данных в очередь выполняется при помощи программных вызовов через специальный API, либо с использованием AWS CLI. Пример использования AWS CLI с YMQ вы можете найти в документации.

YMQ поддерживает API и другие подходы, которые используют в сервисе Amazon SQS, поэтому для работы с ними вы можете использовать уже существующие инструменты, например библиотеки boto3 для Python .

В рамках сервиса Message Queue тарифицируется количество запросов к стандартным очередям и очередям FIFO, а также исходящий трафик. Для целей тарификации каждые 64 КБ данных запроса считаются отдельным запросом.

Первые 100 000 запросов в месяц к очередям любого типа не оплачиваются. А далее их стоимость зависит от типа очереди: для стандартных очередей это 48,7600 ₽ за 1 миллион запросов, для FIFO — 61,1500 ₽.

Исходящий трафик до 10 ГБ не тарифицируется, а затем оплата за него  составляет 1,5300 ₽ за 1 ГБ.

Допустим, за месяц было сделано 2,75 млн запросов объемом 48 КБ к стандартной очереди. Это значит, что нам нужно вычесть нетарифицируемое количество запросов из общего и привести его к тарифу за миллион запросов, а затем приплюсовать исходящий трафик исходя из того, что каждое сообщение умещается в 64 КБ, приведя размер каждого сообщения к гигабайтам. Считаем:

$48,7600\times ((2750000-100000)/1000000)+1,53\times ((2750000\times 48/1024/1024)-10)=129,214+177,30=306,514$ ₽

Более подробную информацию вы найдёте в документации.

На следующем уроке вы расширите созданное ранее приложение для проверки доступности yandex.ru, добавив в него возможность ставить задачи по проверке доступности других веб-ресурсов при помощи очередей.

 

Предположим, вы создаете поисковую систему. У вас есть роботы (поставщики), которые собирают ссылки и передают их обработчикам (потребителям) для разбора страниц и записи результата в базу данных. Самый простой способ передавать ссылки от роботов к обработчикам — вызывать обработчики напрямую из роботов. Однако, такая реализация имеет ряд недочётов, например:

Практически все эти проблемы решаются при помощи очередей, т. е. последовательности некоторой информации. Главная задача очередей — организовать независимую работу поставщиков и потребителей информации в системах реального времени. Вот что это означает в нашем примере.

Робот пишет сообщение в очередь: «У меня есть новая ссылка, вот она». Обработчик ссылок читает записанное сообщение из очереди и забирает себе ссылку на разбор. Чтобы другой обработчик не начал обрабатывать ту же самую ссылку, очередь скрывает это сообщение на некоторый промежуток времени (таймаут видимости), достаточный для его обработки потребителем.

Если сообщение обработано успешно, оно удаляется из очереди, а обработчик забирает себе следующее. Если обработчик вышел из строя или произошёл обрыв соединения, по истечении таймаута видимости сообщение снова становится доступным в очереди, и его может взять другой обработчик.

Таким образом, очередь служит буфером между поставщиком и потребителем, позволяя развязать два параллельных процесса. Если потребитель начинает медленнее вычитывать элементы, то очередь будет их накапливать. А шанс догнать процесс и обработать данные побыстрее даст потом.

Очереди используются внутри программ при взаимодействии потоков и в организации больших систем, где взаимодействуют несколько программ или сервисов. В таких системах очереди помогают решать проблемы интеграции приложений, их отказоустойчивости и масштабирования, например:

Стоит учитывать, что у очередей есть ряд ограничений. Главным из них является неопределённое время на обработку принимающей стороной. Сложно предсказать, когда потребитель получит и обработает элемент.

Существует два распространённых типа очередей: FIFO и стандартная.

FIFO означает «first in, first out», т. е. порядок размещения данных в очереди совпадает с порядком, в котором данные из нее считываются. Очереди FIFO используются в тех случаях, когда нужно обеспечить строгий порядок доставки и однократную обработку сообщений. Например, соблюдение исходного порядка важно для финансовых транзакций. Если представить, что у клиента на счету было 2 000 рублей, и он сначала положил туда 10 000, а затем потратил 5 000, то очевидно, что порядок выполнения транзакций имеет значение. Стоит отметить, что у очередей FIFO сравнительно небольшое количество вызовов API в секунду.

В свою очередь из стандартных очередей сообщения по возможности считываются последовательно, но соблюдение исходного порядка при доставке сообщений не гарантируется. Преимуществом этого подхода является более высокая в сравнении с FIFO пропускная способность: стандартные очереди поддерживают сравнительно большое количество вызовов API в секунду (отправка, принятие или удаление сообщения).

В Yandex Cloud сервисом очередей является Yandex Message Queue. Он относится к PaaS-слою и объединён с другими serverless-сервисами в группу Бессерверные вычисления. В следующем уроке мы поговорим о его специфике.