Как непрерывно получать сообщения из очереди SQS внутри приложения фляги? - PullRequest
Новая
       95

Как непрерывно получать сообщения из очереди SQS внутри приложения фляги?

0 голосов
/ 28 сентября 2019

Я работаю над приложением Flask, которое выполняет некоторые операции с базой данных над первичными ключами в сообщениях, полученных из очереди AWS SQS .

В моем случае я имитировал SQS с использованием https://github.com/vsouza/docker-SQS-local.

Мне нужно получать эти сообщения из очереди SQS , как только будет завершена обработка последнего сообщения.

Я написал функцию, которая при запуске проверяет флаг workload_health_status и, если он имеет значение "UP", получает определенное количество сообщений от SQS , обрабатываетих и пытается извлечь сообщения, как только обработка будет завершена и т. д.

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

Но мне нужно, чтобы этот скрипт всегда выполнялся как фоновая задача или что-то подобное, и мне нужно, чтобы эта функциональность выполнялась как несколько потоков (что означает, что первичные ключи могут обрабатываться одним потоком каждый).

И мое приложение фляги также обслуживает некоторые API, которые можно вызывать, даже если фоновые задачи запущены или нет.Фоновые задачи не должны быть препятствием для вызовов API на сервере фляг.

Может кто-нибудь предложить предпочтительный способ сделать это.Я не могу найти лучшего решения, которое решает эту проблему.

Вот код для получения сообщений: 

import boto3
from app.config import Config
from flask import current_app as app

config = Config()
sqs = boto3.client('sqs', aws_access_key_id=None, aws_secret_access_key=None, endpoint_url=config.QUEUE_ENDPOINT_URL)
feeder_queue = config.FEEDER_QUEUE
def queue_receiver():
    while True:
        if app.config['workload_health_status'] == "UP":
            response = sqs.receive_message(QueueUrl=feeder_queue, MaxNumberOfMessages=10)
            print(response)
            for i in response['Messages']:
                print(i['Body'])

QUEUE_ENDPOINT_URL и FEEDER_QUEUE являются локальными URL-адресами.

Как сделать так, чтобы эта функция продолжала работать в фоновом режиме, когда флеш-сервер запущен?

...