这个 Python 代码是使用多线程的安全方法吗

Question

我用于图形的应用程序具有嵌入式 Python 解释器 - 除了有一些特殊对象之外，它的工作方式与任何其他 Python 解释器完全相同。

基本上我正在尝试使用 Python 下载一堆图像并进行其他网络和磁盘 I/O。如果我在没有多线程的情况下执行此操作，我的应用程序将冻结（即视频停止播放），直到下载完成。

为了解决这个问题，我尝试使用多线程。但是，我无法触及任何主要进程。

我已经写了这段代码。唯一对该程序独特的部分进行了注释。 me.store / me.fetch 基本上是获取全局变量的一种方式。 op('files') 指的是一个全局表。

这是两件事，“在主进程中”，只能以线程安全的方式进行访问。我不确定我的代码是否这样做。

我会感谢任何关于为什么或（为什么不是）此代码是线程安全的以及我如何以线程安全的方式访问全局变量的输入。

我担心的一件事是counter 是如何被许多线程多次获取的。由于它仅在写入文件后才更新，这是否会导致不同线程访问具有相同值的计数器的竞争条件（然后不正确存储递增的值）。或者，如果磁盘写入失败，计数器会发生什么。

from urllib import request
import threading, queue, os

url = 'http://users.dialogfeed.com/en/snippet/dialogfeed-social-wall-twitter-instagram.json?api_key=ac77f8f99310758c70ee9f7a89529023'

imgs = [
    'http://search.it.online.fr/jpgs/placeholder-hollywood.jpg.jpg',
    'http://www.lpkfusa.com/Images/placeholder.jpg',
    'http://bi1x.caltech.edu/2015/_images/embryogenesis_placeholder.jpg'
]

def get_pic(url):
    # Fetch image data
    data = request.urlopen(url).read()
    # This is the part I am concerned about, what if multiple threads fetch the counter before it is updated below
    # What happens if the file write fails?
    counter = me.fetch('count', 0)

    # Download the file
    with open(str(counter) + '.jpg', 'wb') as outfile:
        outfile.write(data)
        file_name = 'file_' + str(counter)
        path = os.getcwd() + '\\' + str(counter) + '.jpg'
        me.store('count', counter + 1)
        return file_name, path


def get_url(q, results):
    url = q.get_nowait()
    file_name, path = get_pic(url)
    results.append([file_name, path])
    q.task_done()

def fetch():
    # Clear the table
    op('files').clear()
    results = []
    url_q = queue.Queue()
    # Simulate getting a JSON feed
    print(request.urlopen(url).read().decode('utf-8'))

    for img in imgs:
        # Add url to queue and start a thread
        url_q.put(img)
        t = threading.Thread(target=get_url, args=(url_q, results,))
        t.start()

    # Wait for threads to finish before updating table
    url_q.join()
    for cell in results:
        op('files').appendRow(cell)
    return

# Start a thread so that the first http get doesn't block
thread = threading.Thread(target=fetch) 
thread.start()

Answer 1

您的代码似乎根本不安全。要点：

• 追加到results 是不安全的——两个线程可能会尝试同时追加到列表中。
• 访问和设置counter 是不安全的——我在另一个线程设置新的counter 值之前获取counter 的线程。
• 传递一个 url 队列是多余的 - 只需将一个新 url 传递给每个作业。

另一种方式 (concurrent.futures)

既然您使用的是 python 3，为什么不使用 concurrent.futures 模块，它使您的任务更易于管理。下面我以不需要显式同步的方式编写了您的代码——所有工作都由期货模块处理。

from urllib import request
import os
import threading

from concurrent.futures import ThreadPoolExecutor
from itertools import count

url = 'http://users.dialogfeed.com/en/snippet/dialogfeed-social-wall-twitter-instagram.json?api_key=ac77f8f99310758c70ee9f7a89529023'

imgs = [
    'http://search.it.online.fr/jpgs/placeholder-hollywood.jpg.jpg',
    'http://www.lpkfusa.com/Images/placeholder.jpg',
    'http://bi1x.caltech.edu/2015/_images/embryogenesis_placeholder.jpg'
]

def get_pic(url, counter):
    # Fetch image data
    data = request.urlopen(url).read()

    # Download the file
    with open(str(counter) + '.jpg', 'wb') as outfile:
        outfile.write(data)
        file_name = 'file_' + str(counter)
        path = os.getcwd() + '\\' + str(counter) + '.jpg'
        return file_name, path

def fetch():
    # Clear the table
    op('files').clear()

    with ThreadPoolExecutor(max_workers=2) as executor:
        count_start = me.fetch('count', 0)
        # reserve these numbers for our tasks
        me.store('count', count_start + len(imgs))
        # separate fetching and storing is usually not thread safe
        # however, if only one thread modifies count (the one running fetch) then 
        # this will be safe (same goes for the files variable)

        for cell in executor.map(get_pic, imgs, count(count_start)):
            op('files').appendRow(cell)


# Start a thread so that the first http get doesn't block
thread = threading.Thread(target=fetch) 
thread.start()

如果多个线程修改计数，那么在修改计数时应该使用锁。

例如。

lock = threading.Lock()

def fetch():
    ...
    with lock:
        # Do not release the lock between accessing and modifying count.
        # Other threads wanting to modify count, must use the same lock object (not 
        # another instance of Lock).
        count_start = me.fetch('count', 0)
        me.store('count', count_start + len(imgs))    
   # use count_start here

如果一项作业由于某种原因失败，则唯一的问题是您将得到一个丢失的文件编号。任何引发的异常也会中断执行器的映射，方法是在那里重新引发异常——这样你就可以在需要时做一些事情。

您可以避免使用计数器，方法是使用 tempfile 模块在将文件移动到永久位置之前找到临时存储文件的位置。

Answer 2

如果您是 python 多线程的新手，请记得查看 multiprocessing 和 threading。

您的代码看起来不错，虽然代码风格不是很容易阅读。您需要运行它以查看它是否按您的预期工作。

with 将确保您的锁被释放。进入block时调用acquire()方法，退出block时调用release()方法。

如果您添加更多线程，请确保它们没有使用队列中的相同地址并且没有竞争条件（似乎是由Queue.get() 完成的，但您需要运行它来验证）。请记住，每个线程共享相同的进程，因此几乎所有内容都是共享的。你不希望两个线程处理相同的address

Answer 3

Lock 根本不做任何事情。你只有一个线程调用download_job - 那是你分配给my_thread 的线程。另一个，主线程，调用offToOn，并在它到达该函数的末尾时立即完成。所以没有第二个线程试图获取锁，因此没有第二个线程被阻塞。您提到的表格显然位于您明确打开和关闭的文件中。如果操作系统保护这个文件不被不同程序同时访问，你就可以侥幸逃脱；否则肯定是不安全的，因为你还没有完成任何线程同步。

线程之间的正确同步要求不同的线程可以访问同一个锁；即，一个锁被多个线程访问。另请注意，“线程”不是“进程”的同义词。 Python 两者都支持。如果您真的应该避免访问主进程，则必须使用多处理模块来启动和管理第二个进程。

而且这段代码永远不会退出，因为总是有一个线程在无限循环中运行（threader）。

以线程安全的方式访问资源需要这样的东西：

a_lock = Lock()
def use_resource():
    with a_lock:
        # do something

锁只在使用它的函数之外创建一次。从任何线程对整个应用程序中资源的每次访问都必须获取相同的锁，方法是调用 use_resource 或类似的方法。