处理大量的线程和数据库连接（Python）我能做些什么来节省资源？

Question

我在搞一个广播流媒体项目。目前我正在创建一个python后端。数据库中有超过150000个在线电台流。我尝试添加的一个功能是通过当前播放的歌曲搜索广播电台。我使用Dirble's streamscrobbler从每个广播电台获取当前正在播放的歌曲，使用请求并查看元数据。在

显然，这个脚本将需要多线程，以便在一个可行的时间内抓取当前正在播放的歌曲。不超过2分钟。如果可能的话，最好是1到30分钟。在

我以前从来没有搞过这么大规模的项目。创建太多线程会占用资源，因此最好创建一个ThreadPoolExecutor。我还使用SQLAlchemy将这些歌曲插入数据库。显然，SQLAlchemy使用默认实现的连接池？在

我正在使用轻量级的scheduler python module by Daniel Bader调度此任务。看来效果不错。在

现在，我遇到的问题是我得到了这个错误： 错误：无法启动新线程

我猜这是因为我用了太多的资源。我能做什么？我可以减少线程的数量，但是任务似乎没有在我需要的时间内完成，所以这会增加遍历每个流url的时间。在

from streamscrobbler import streamscrobbler
from concurrent.futures import ThreadPoolExecutor
import re
from sqlalchemy import *

#get song name from station
def manageStation(station_id, station_link):
    current_song = getCurrentSong(station_link)
    current_song = current_song.replace("'", "")
    current_song = current_song.replace("\"", "")
    current_song = current_song.replace("/", "")
    current_song = current_song.replace("\\", "")
    current_song = current_song.replace("%", "")

    if current_song:
        with db.connect() as con:
            rs = con.execute("INSERT INTO station_songs VALUES('" + str(station_id) + "', '" + current_song + "', '') ON DUPLICATE KEY UPDATE song_name = '" + current_song + "';")
    return ""

def getCurrentSong(stream_url):
    streamscrobblerobj = streamscrobbler()
    stationinfo = streamscrobblerobj.getServerInfo(stream_url)
    metadata = stationinfo.get("metadata")
    regex = re.search('\'song\': \'(.*?)\'' , str(metadata))
    if regex:
        return regex.group(1)
    return ""

def update() :
    print 'update starting'
    global db
    db = create_engine('mysql://root:pass@localhost:3306/radio')
    global threadExecutor 
    threadExecutor = ThreadPoolExecutor(max_workers=20000)
    with db.connect() as con:
        rs = con.execute("SELECT id, link FROM station_table")
        for row in rs.fetchall():
            threadExecutor.submit(manageStation, row[0], row[1])

Answer 1

您不需要为每个任务提供一个真正的线程，因为大多数时候，线程将等待来自套接字（web请求）的IO。在

您可以尝试使用green threads使用类似于gevent的方法，使用如下架构：

from gevent import monkey; monkey.patch_socket()

NUM_GLETS = 20    
STATION_URLS = (
   'http://station1.com',
   ...
)

pool = gevent.Pool(NUM_GLETS)
tasks = [pool.spawn(analyze_station, url) for url in STATION_URLS]
pool.join(tasks)

其中analyze_station是获取和分析特定站点的代码。在

结果应该是一个单线程程序，但是不是阻塞每个web请求，而是在套接字等待数据时运行另一个绿色线程。这比为大多数空闲工作生成真正的线程要高效得多。在