Как обеспечить, чтобы за один раз запускался только один поток, самые последние данные?

Question

У меня есть приложение Flask с определенным URL-маршрутом, которое при вызове вызывает функцию внутри процесса через библиотеку потоков.Эта функция содержит цикл True, который внутри имеет запрос к БД.

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

Мне нужен один поток.

Вот мой нынешний подход:

@mod.route('/api/start-zmq-listener')
def startZMQListener():
    try:
        thread = threading.Thread(target=enableZMQ, name="enableZMQ", kwargs={
            'username': 'test',
            'stopFlag': stopFlag
        })
        thread.start()
        return success_response('ok')

    except Exception as e:
        return error_response(e)


def enableZMQ(username, stopFlag):
    if not stopFlag:
        context = zmq.Context()

        listen_on = 'tcp://' + ENGINE_IP 

        sock = context.socket(zmq.SUB)
        sock.setsockopt(zmq.SUBSCRIBE, b"")
        sock.connect(listen_on)
        print ('listening on ' + listen_on)

        while True:
            print MONGODB QUERY

Допустим, «ЗАПРОС MONGODB» каждый раз одинаков, но данные, которые он запрашивает,разные.

Как мне убедиться, что всегда запущен только один процесс, который использует самый последний запрос MongoDB?

Answer 1

Одно из возможных решений - всегда иметь готовые данные в оперативной памяти.

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

Примерно так:

from threading import Thread, Lock
# Add flask stuff

global value_store = [] # Some data
global mutex = Lock()
INTERVAL = 1  # second
global should_query = False

@mod.route('/api/getSomeData')
def get_some_data():
    global mutex
    with mutex
        global value_store
        return jsonify(value_store)

def callback(data):
    global mutex 
    with mutex:
        global value_store
        value_store = data

def query_repeat():
    global should_query
    while should_query:
        # query db here
        raw data = query_database()

        # filter/process data
        data = filter_data(raw_data)

        # call callback
        callback(data)

        # sleep and repeat
        time.sleep(INTERVAL)


t = Thread(target = query_repeat, args = (,))
should_query = True
t.start()

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