我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用dummy_threading.Thread()。
def _reader(self, name, stream, outbuf): """ Thread runner for reading lines of from a subprocess into a buffer. :param name: The logical name of the stream (used for logging only). :param stream: The stream to read from. This will typically a pipe connected to the output stream of a subprocess. :param outbuf: The list to append the read lines to. """ while True: s = stream.readline() if not s: break s = s.decode('utf-8').rstrip() outbuf.append(s) logger.debug('%s: %s' % (name, s)) stream.close()
def test_make_msgid_collisions(self): # Test make_msgid uniqueness, even with multiple threads class MsgidsThread(Thread): def run(self): # generate msgids for 3 seconds self.msgids = [] append = self.msgids.append make_msgid = Utils.make_msgid clock = time.time tfin = clock() + 3.0 while clock() < tfin: append(make_msgid()) threads = [MsgidsThread() for i in range(5)] with start_threads(threads): pass all_ids = sum([t.msgids for t in threads], []) self.assertEqual(len(set(all_ids)), len(all_ids))
def _get_pages(self, locations, req): """Yields (page, page_url) from the given locations, skipping locations that have errors, and adding download/homepage links""" pending_queue = Queue() for location in locations: pending_queue.put(location) done = [] seen = set() threads = [] for i in range(min(10, len(locations))): t = threading.Thread(target=self._get_queued_page, args=(req, pending_queue, done, seen)) t.setDaemon(True) threads.append(t) t.start() for t in threads: t.join() return done
