Создание уникальных потоков для уникального идентификатора в потоке python flask в сокете

Question

У меня есть приложение python flask, которое может получать данные из запроса json и затем обрабатывать.

Пример моего кода приведен ниже: -

# Start with a basic flask app webpage.
from flask_socketio import SocketIO, emit
from flask import Flask, render_template, request, url_for, copy_current_request_context
from time import sleep
from threading import Thread, Event

__author__ = 'shark'

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
app.config['DEBUG'] = True

# turn the flask app into a socketio app
socketio = SocketIO(app, async_mode=None, logger=True, engineio_logger=True)

thread = Thread()
thread_stop_event = Event()


@app.route('/platform-data', methods=['POST'])
def platformData():
    """
    Generate a random number every 1 second and emit to a socketio instance (broadcast)
    Ideally to be run in a separate thread?
    """
    # infinite loop of magical random numbers
    print("Receiving platform data")
    while not thread_stop_event.isSet():
        req_data = request.get_json()

        id = req_data['id']
        latitude = req_data['coordinates'][1]
        longitude = req_data['coordinates'][0]
        speed = req_data['speed']
        angle = req_data['angle']
        length = req_data['dimensions'][0]
        width = req_data['dimensions'][1]
        laneW = req_data['lane_width']
        spdLmt = req_data['speed_limit']

        # return testProcess(speed)

        #        print(id, latitude, longitude, speed, angle, length, width, laneW, spdLmt)

        def testProcess(id,speed):
            if speed > 30:
                print(id, " ", "slow down")
            else:
                print(id," ", "ok")

        testProcess(id,speed)

        # return {"speed": speed}

        # socketio.emit('speed', {'speed': speed}, namespace='/test')
        socketio.sleep(1)


@app.route('/')
def index():
    # only by sending this page first will the client be connected to the socketio instance
    return render_template('index.html')


@socketio.on('connect', namespace='/test')
def test_connect():
    # need visibility of the global thread object
    global thread
    print('Client connected')

    # Start the random number generator thread only if the thread has not been started before.
    if not thread.isAlive():
        print("Starting Thread")
        thread = socketio.start_background_task(platformData)


@socketio.on('disconnect', namespace='/test')
def test_disconnect():
    print('Client disconnected')


if __name__ == '__main__':
    socketio.run(app)

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

Мой json запрос выглядит следующим образом: -

{
    "id" : "1"
    "speed" : 20
}

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

Есть идеи, как это сделать?

Answer 1

Надеюсь, теперь я правильно понял ваш вопрос. Вы хотите, чтобы процесс с уникальным идентификатором и настройкой (скоростью) можно было изменить? Пожалуйста, исправьте меня, если я ошибаюсь.

Тогда я бы использовал инструмент многопроцессорной обработки Python.

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

Маленький Пример:

from multiprocessing import Process, Value
import time

speed_var = Value('i', 0)


def testProcess(speed_var):
    while True:
        speed = speed_var.value
        if speed >= 30:
            print("It's running fast! Perfect")
        else:
            print("It's so slow, even my grandma would be faster")
        print(speed)
        time.sleep(0.05)

id_1 = Process(target=testProcess, args=(speed_var, ))
id_1.start()

speed = 0

while speed >= 40:
   speed += 3
   speed_var.value = speed
   time.sleep(0.2)
id_1.terminate()

Другой подход заключается в работе с очередью, а не со значениями. Затем вы можете просто подождать элемента в очереди и начать его обработку. Для получения дополнительной информации просто спросите или прочитайте документы: https://docs.python.org/2/library/multiprocessing.html#shared -ctypes-objects

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