python - 在Python中向线程发送数据

Question

我正在尝试对以下代码进行线程化并向其发送数据(以随机间隔)，但我不知道如何进行。我将所有数据保存到 txt 文件并从那里读取信息，但效果不佳。是否可以创建一个将数据发送到特定线程的函数(例如:SendDataToThread(data, ThreadNumber))？我将如何读取发送的数据？我见过一些使用队列的解决方案，但我无法理解它们。这是我暂时用来绘制我发现的图表的脚本 here 。抱歉，如果问题看起来很简单，但我以前从未遇到过线程或 matplotlib 的麻烦。

import matplotlib.pyplot as plt
from threading import Thread
plt.ion()
class DynamicUpdate():
    #Suppose we know the x range
    min_x = 0
    max_x = 10

    def on_launch(self):
        #Set up plot
        self.figure, self.ax = plt.subplots()
        self.lines, = self.ax.plot([],[], 'o')
        #Autoscale on unknown axis and known lims on the other
        self.ax.set_autoscaley_on(True)
        self.ax.set_xlim(self.min_x, self.max_x)
        #Other stuff
        self.ax.grid()
        ...

    def on_running(self, xdata, ydata):
        #Update data (with the new _and_ the old points)
        self.lines.set_xdata(xdata)
        self.lines.set_ydata(ydata)
        #Need both of these in order to rescale
        self.ax.relim()
        self.ax.autoscale_view()
        #We need to draw *and* flush
        self.figure.canvas.draw()
        self.figure.canvas.flush_events()
    #Example 
    def __call__(self):
        # read/plot data

Answer 1

这里有一些示例代码，展示了如何完成所询问的一些事情。这使用多线程而不是多处理，并显示了一些使用队列、启动/停止工作线程以及使用附加数据更新 matplotlib 图的示例。

(部分代码来自其他问题的答案，包括 this one 和 this one 。)

该代码显示了异步工作器的可能实现，可以将数据发送到该工作器以进行后续处理。工作线程使用内部队列来缓冲数据，并使用内部线程(循环)从队列中读取数据，进行一些处理并发送结果以进行显示。

还显示了异步绘图仪实现。结果可以从多个工作人员发送到该绘图仪。 (这也使用内部队列进行缓冲；这样做是为了允许主程序线程本身调用更新绘图的函数，这似乎是 matplotlib 的要求。)

注意这是为 OSX 上的 Python 2.7 编写的。希望其中一些可能有用。

import time
import threading
import Queue
import math
import matplotlib.pyplot as plt

class AsynchronousPlotter:
    """
    Updates a matplotlib data plot asynchronously. 
    Uses an internal queue to buffer results passed for plotting in x, y pairs.
    NB the output_queued_results() function is intended be called periodically
    from the main program thread, to update the plot with any waiting results.
    """

    def output_queued_results(self):
        """
        Plots any waiting results. Should be called from main program thread.
        Items for display are x, y pairs
        """
        while not self.queue.empty():
            item = self.queue.get()
            x, y = item
            self.add_point(x, y)
            self.queue.task_done()

    def queue_result_for_output(self, x, y):
        """
        Queues an x, y pair for display. Called from worker threads, so intended 
        to be thread safe.
        """
        self.lock.acquire(True)
        self.queue.put([x, y]) 
        self.lock.release()

    def redraw(self):
        self.ax.relim()
        self.ax.autoscale_view()
        self.fig.canvas.draw()
        plt.pause(0.001)

    def add_point(self, x, y):
        self.xdata.append(x)
        self.ydata.append(y)
        self.lines.set_xdata(self.xdata)
        self.lines.set_ydata(self.ydata)
        self.redraw()

    def __init__(self):
        self.xdata=[]
        self.ydata=[]
        self.fig = plt.figure()
        self.ax = self.fig.add_subplot(111)
        self.lines, = self.ax.plot(self.xdata, self.ydata, 'o')
        self.ax.set_autoscalex_on(True)   
        self.ax.set_autoscaley_on(True)   
        plt.ion()
        plt.show()
        self.lock = threading.Lock()
        self.queue = Queue.Queue()

class AsynchronousWorker:
    """
    Processes data asynchronously. 
    Uses an internal queue and internal thread to handle data passed in.
    Does some processing on the data in the internal thread, and then
    sends result to an asynchronous plotter for display
    """

    def queue_data_for_processing(self, raw_data):
        """
        Queues data for processing by the internal thread.
        """
        self.queue.put(raw_data)

    def _worker_loop(self):
        """
        The internal thread loop. Runs until the exit signal is set.
        Processes the supplied raw data into something ready
        for display. 
        """
        while True:
            try:
                # check for any data waiting in the queue
                raw_data = self.queue.get(True, 1)
                # process the raw data, and send for display
                # in this trivial example, change circle radius -> area
                x, y = raw_data
                y = y**2 * math.pi   
                self.ap.queue_result_for_output(x, y)
                self.queue.task_done()
            except Queue.Empty:
                pass
            finally:
                if self.esig.is_set():
                    return

    def hang_up(self):
        self.esig.set()    # set the exit signal...
        self.loop.join()   # ... and wait for thread to exit

    def __init__(self, ident, ap):
        self.ident = ident
        self.ap = ap
        self.esig = threading.Event()
        self.queue = Queue.Queue()
        self.loop = threading.Thread(target=self._worker_loop)
        self.loop.start()

if __name__ == "__main__":     
    ap = AsynchronousPlotter()
    num_workers = 5   # use this many workers

    # create some workers. Give each worker some ID and tell it 
    # where it can find the output plotter
    workers = []
    for worker_number in range (num_workers):
        workers.append(AsynchronousWorker(worker_number, ap)) 

    # supply some data to the workers
    for worker_number in range (num_workers):
        circle_number = worker_number
        circle_radius = worker_number * 4
        workers[worker_number].queue_data_for_processing([circle_number, circle_radius])

    # wait for workers to finish then tell the plotter to plot the results
    # in a longer-running example we would update the plot every few seconds
    time.sleep(2)
    ap.output_queued_results();

    # Wait for user to hit return, and clean up workers
    raw_input("Hit Return...")
    for worker in workers:
        worker.hang_up()