如何在 Flask 应用程序中启动后台线程？

如何解决如何在 Flask 应用程序中启动后台线程？

我编写了一个运行良好的 Flask 应用程序，我希望在它运行时，一个单独的后台线程应该与它并行执行一些操作。问题是，这样做根本不会产生线程，但我知道我的代码是正确的，因为在简单的 Python 脚本上使用完全相同的线程代码部分可以按预期工作。

app.py

weatherCollectorThread = WeatherDataCollectorThread()
...
if __name__ == '__main__':
    try:
        print("Starting Weather Collector Thread...")       
        weatherCollectorThread.start()
        print("Starting the WebApp...")
        app.run(debug=True)
    except KeyboardInterrupt:
        try:
            weatherCollectorThread.stop()
        except:
            pass

WeatherDataCollectorThread 类

class WeatherDataCollectorThread:
    def __init__(self):
        self.weatherStations = DBHelper.get_weather_stations()
        self.weatherApiKey = "REDACTED"
        self.baseURL = "SOME URL"
        self.isThreadRunning = False
        self.result_log = open('results.log','a+')

    def storeWeatherData(self,weather):
        conn = DBHelper.get_connection()
        cur = conn.cursor()
        cur.execute("INSERT INTO weather_data(city,country,Now_unixtime,last_updated_unixtime,temperature,isDay,condition_text,condition_icon,windspeed,winddir,pressure,precipitation,cloud,humidity) VALUES (?,?,?)",[weather['city'],weather['country'],weather['Now_unixtime'],weather['last_updated_unixtime'],weather['temperature'],weather['isDay'],weather['condition_text'],weather['condition_icon'],weather['windspeed'],weather['winddir'],weather['pressure'],weather['precipitation'],weather['cloud'],weather['humidity']])
        conn.commit()
        conn.close()

    def collectWeatherData(self):
        self.isThreadRunning = True 
        while self.isThreadRunning: 
            for each_station in self.weatherStations:
                if each_station['isWorking'] != 1:
                    continue
                print("Sending request")
                params = {'q':each_station['location'],'key':self.weatherApiKey}
                resp = requests.get(url=self.baseURL,params=params)
                print("Request received")
                weatherData = json.loads(resp.text)
                location = weatherData['location']
                current = weatherData['current']
                weather = {}
                weather['city'] = location['name']
                weather['country'] = location['country']
                weather['Now_unixtime'] = location['localtime_epoch']
                weather['last_updated_unixtime'] = current['last_updated_epoch']
                weather['temperature'] = current['temp_c']
                weather['isDay'] = current['is_day']
                weather['condition_text'] = current['condition']['text']
                weather['condition_icon'] = current['condition']['icon']
                weather['windspeed'] = current['wind_kph']
                weather['winddir'] = current['wind_dir']
                weather['pressure'] = current['pressure_mb']
                weather['precipitation'] = current['precip_mm']
                weather['cloud'] = current['cloud']
                weather['humidity'] = current['humidity']
                self.storeWeatherData(weather)
                print("Data stored\n" + '-'*24)
                self.result_log.write(resp.text + '\n')
                
            sleep(60)

    def start(self):
        self.thread = Thread(target=self.collectWeatherData)
        self.thread.start()

    def join_instrument(self,session):
        conn = DBHelper.get_connection()
        cur = conn.cursor()
        cur.execute("UPDATE weather_stations SET isWorking=1 WHERE weatherStationID=?",[session['weatherStationID']])
        conn.commit()
        conn.close()

    def detach_instrument(self,session):
        conn = DBHelper.get_connection()
        cur = conn.cursor()
        cur.execute("UPDATE weather_stations SET isWorking=0 WHERE weatherStationID=?",[session['weatherStationID']])
        conn.commit()
        conn.close()

    def stop(self):
        self.result_log.close()
        self.isThreadRunning = False

解决方法

所以我想出了解决方案。

您会看到，当您使用 flask run 运行您的网络应用程序时，它会忽略脚本中的每一个函数调用并解析装饰器并自行启动应用程序。因此，如果您执行以下操作：

if __name__ == '__main__':
   app.start()
   someOtherFunction()

app.start() 和 someOtherFunction() 都不会启动。

那么解决方案呢？

只需使用 python3 app.py 即可运行脚本。

...是的，就是这么简单:|