标签: python-multithreading

我有一个调用API的脚本.为了加快脚本速度,我试图实现线程化.

当我在IDLE时,下面的脚本工作,但是当我尝试从命令行使用sys argv运行它时,我收到了下面列出的两种类型的错误.

错误1

Fatal Python error: PyImport_GetModuleDict: no module dictionary!

This application has requests the Runtime to terminate it in an unusual way.  Please         contact the application's support team for more information.

错误2

Exception in thread Thread-1 (most likely raised during iterpreter shutdown): 
Exception in thread Thread-2 (most likely raised during iterpreter shutdown):
Exception in thread Thread-3 (most likely raised during iterpreter shutdown):
Exception in thread Thread-5 (most likely raised during iterpreter shutdown):

我找不到任何关于这些错误的信息.所以,任何帮助表示赞赏.下面是处理线程的脚本部分.

import threading
import diffbot

urls …

我试图创建86个task.py实例同时运行.

import sys
import subprocess

for file in range(86):
    subprocess.call([sys.executable,'task.py',str(file)+'in.csv',str(filen)+'out.csv'])

我正在开发一个需要位自动化和网络抓取的项目，我正在使用Selenium和BeautifulSoup (python2.7)。

我只想打开Web 浏览器的一个实例并登录到网站，保持该会话，我试图打开由线程独立控制的新选项卡，每个线程控制一个选项卡并执行自己的任务。我该怎么做？一个示例代码会很好。好吧，这是我的代码：

def threadFunc(driver, tabId):
    if tabId == 1:
        #open a new tab and do something in it
    elif tabId == 2:
        #open another new tab with some different link and perform some task
    .... #other cases


class tabThreads(threading.Thread):

    def __init__(self, driver, tabId):
        threading.Thread.__init__(self)
        self.tabID = tabId
        self.driver = driver

    def run(self):
        print "Executing tab ", self.tabID
        threadFunc(self.driver, self.tabID)

def func():
    # Created a main window
    
    driver = …

我想使用 python ThreadPoolExecutor （附加代码）同时调用两个 api。如果这两个 api 调用中的任何一个有响应，我想停止调用另一个。因为对于我的用例，两个 api 之一将需要很长时间才能返回响应，我想避免调用。

def get_rest_api_response(url):
    return requets.get(url)
    
import requests, os
import concurrent.futures
from concurrent.futures import ThreadPoolExecutor, as_completed
with ThreadPoolExecutor(max_workers=4) as executor:
    f1 = executor.submit(get_rest_api_response, url="REST_API_URL_1")
    f2 = executor.submit(get_rest_api_response, url="REST_API_URL_2")
    
    no_future_is_done = True
    while(no_future_is_done):
        if f1.done():
            no_future_is_done = False
            print("f1 is done")
            output = f1.result()
            print(f2.cancel())  ######------> Failing!
        if f2.done():
            no_future_is_done = False
            print("f2 is done")
            output = f2.result()
            print(f1.cancel()) ######-------> Failing!
    print(output)

我正在使用 future.cancel() 但它失败并返回 False。 https://pd.codechef.com/docs/py/3.4.2/library/concurrent.futures.html#concurrent.futures.Future.cancel

我还有其他方法可以实现这一目标吗？

我使用Python 2 subprocess与threading螺纹采取标准输入,处理它与二进制文件A,B以及C和写入修改的数据标准输出.

这个脚本(让我们称之为:) A_to_C.py非常慢,我想学习如何解决它.

一般流程如下:

A_process = subprocess.Popen(['A', '-'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
produce_A_thread = threading.Thread(target=produceA, args=(sys.stdin, A_process.stdin))

B_process = subprocess.Popen(['B', '-'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
convert_A_to_B_thread = threading.Thread(target=produceB, args=(A_process.stdout, B_process.stdin))

C_process = subprocess.Popen(['C', '-'], stdin=subprocess.PIPE)
convert_B_to_C_thread = threading.Thread(target=produceC, args=(B_process.stdout, C_process.stdin)) 

produce_A_thread.start()
convert_A_to_B_thread.start()
convert_B_to_C_thread.start()

produce_A_thread.join()
convert_A_to_B_thread.join()
convert_B_to_C_thread.join()

A_process.wait()
B_process.wait()
C_process.wait()

这个想法是标准输入进入A_to_C.py:

1. 在A二进制处理标准输入的块,并创建A与该功能-output produceA.
2. 该B二进制过程的块A的标准输出并产生B通过功能-output produceB.
3. 该C二进制过程的块B …

我正在尝试为Python 3.4中的项目制作线程飞行软件,其中我需要线程重新启动,以防在传感器读取期间发生I/O错误或其他类似的侥幸崩溃.因此,我正在制作一个看门狗来检查线程是否已经死亡并重新启动它们.

起初我试图检查线程是否不再存在并重新启动它,这样做:

>>> if not a_thread.isAlive():
...     a_thread.start()
Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
  File "c:\Python34\lib\threading.py", line 847, in start
    raise RuntimeError("threads can only be started once")
RuntimeError: threads can only be started once

从threadingPython本身的角度来看,这种行为是有道理的,但这使我的工作更加困难.所以我使用字典实现了一个解决方案来存储初始线程并将其复制到一个新对象并在必要时启动它.不幸的是,这也不起作用.这是一个基本的例子:

import threading
import logging
import queue
import time
from copy import copy, deepcopy

def a():
    print("I'm thread a")
def b():
    print("I'm thread b")

# Create thread objects
thread_dict = {
'a': threading.Thread(target=a, name='a'),
'b': threading.Thread(target=b, name='b')
} …

我相信之前已经提出过许多类似的问题,但在阅读了很多类似的问题后,我仍然不太确定应该做些什么.所以,我有一个Python脚本来控制一些外部仪器(摄像头和功率计).我通过使用ctypes调用.dll文件中的C函数为两种乐器编写了类.现在它看起来像这样:

for i in range(10):
    power_reading = newport.get_reading(N=100,interval=1) # take power meter reading
    img = camera.capture(N=10)
    value = image_processing(img) # analyze the img (ndarray) to get some values
    results.append([power_reading,value]) # add both results to a list

我想同时开始执行前两行.双方newport.get_reading并camera.capture需要100ms-1运行(他们会在同一时间运行,如果我选择了正确的参数).我不需要它们在同一时间完全启动,但理想情况下延迟应该小于总运行时间的大约10-20%(因此当每次运行时运行时间小于0.2秒).根据我的阅读,我可以使用该multiprocessing模块.所以我根据这篇文章尝试这样的事情:

def p_get_reading(newport,N,interval,return_dict):
    reading = newport.get_reading(N,interval,return_dict)
    return_dict['power_meter'] = reading

def p_capture(camera,N,return_dict):
    img = camera.capture(N)
    return_dict['image'] = img

for i in range(10):
    manager = multiprocessing.Manager()
    return_dict = manager.dict()
    p = multiprocessing.Process(target=p_capture, args=(camera,10))
    p.start()
    p2 = multiprocessing.Process(target=p_get_reading, args=(newport,100,1)) …

当我在 a和viathreading._wait_for_tstate_lock之间传输大量数据时，会引发异常。ProcessThreadmultiprocessing.Queue

我的最小工作示例首先看起来有点复杂 - 抱歉。我会解释。原始应用程序将大量（不那么重要）文件加载到 RAM 中。这是在单独的进程中完成的，以节省资源。主 GUI 线程不应冻结。

1. GUI 启动一个单独的进程Thread以防止 gui 事件循环冻结。

2. Thread然后，这个单独的任务将启动一个Process应该完成工作的任务。

a) 这Thread实例化了 a multiprocess.Queue（注意这是 amultiprocessing而不是threading！）

b) 这是为了从回Process共享数据。ProcessThread

3. 做Process了一些工作（3 个步骤）并将.put()结果放入multiprocessing.Queue.

4. 当Process结束时Thread，再次接管并收集来自的数据Queue，将其存储到自己的属性中MyThread.result。

5. 告诉ThreadGUI 主循环/线程在有时间的情况下调用回调函数。

6. 回调函数 ( MyWindow::callback_thread_finished()) 从 中获取结果MyWindow.thread.result。

问题是，如果放入的数据Queue发生了很大的事情，我不明白 -MyThread永远不会结束。我必须通过 Strg+C 取消该应用程序。

我从文档中得到了一些提示。但我的问题是我没有完全理解文档。但我有一种感觉，我的问题的关键就在那里。 …

在我的程序中，我有一个用于在 shell 中执行命令的实用函数，这是它的一个简化版本：

def run_command(cmd):
    s = time.time()
    print('starting subprocess')
    proc = subprocess.Popen(cmd.split(),
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            universal_newlines=True)
    print('subprocess started after ({0}) seconds'.format(time.time() - s))

我的程序使用来自不同线程的这个函数来执行命令。有时，“Popen”行需要大约 70 秒才能完成。我的意思是在一天内在不同程序运行的数千次调用中，这种情况发生了大约 4-5 次。据我所知，Popen 是非阻塞的。对我来说奇怪的是，当它确实发生时，它需要相同的大约 70 秒才能启动。重要的是要注意，在发生这种情况时，我还有 3-4 个其他线程在循环中等待：

while some_counter > 0:
    time.sleep(0.5) 

他们这样做最多 60 秒。在他们放弃并完成他们的流程后，我看到还有大约 14 秒，直到“Popen”调用结束。从某些线程并行运行“Popen”与“等待循环”中的其他线程是否存在问题？

更新1： 我现在看到这个问题是在我从Fedora27+Python3.6切换到Fedora31+python3.7之后开始的。

我在从 python 中的分页 API 响应下载大型数据集时遇到内存问题。当我尝试使用 ThreadPoolExecutor 并行下载多个页面时，我注意到已完成和已解决的 future 不会释放其内存占用。

我尝试通过以下两个示例来简化它。第一个使用max_workers设置为 1 的 ThreadPoolExecutor 下载所有页面（据我所知，这应该具有与简单循环相同的内存占用量）：

from random import random
from concurrent.futures  import ThreadPoolExecutor, as_completed
import gc

TOTAL_PAGES = 60 

def download_data(page: int = 1) -> list[float]:
    # Send a request to some resource to get data
    print(f"Downloading page {page}.")
    return [random() for _ in range(1000000)] # mock some larga data sets

def threadpool_memory_test(): 
    processed_pages = 0
    with ThreadPoolExecutor(max_workers=1) as executor:
        future_to_page = {
            executor.submit(download_data, page): page for page in …