Python从门到精通(六):线程-01-线程（线程 Python）

Python从门到精通(六):线程-01-线程（线程 Python）

Thread基本是老生常谈了，说实话在业务开发中真的很少能用了，但又是非常关键的内容。吐槽一点，如果同学们在将来有机会当面试官的话，一定避免面试做火箭实际拧螺丝。一定要从实际出发。与其聊八股文，不如挖下候选人的思维和反应能力。python的线程会在单独的系统级线程中执行，一旦启动，将独立执行直到返回目标函数。python的线程原理和java差不太多，不清楚的可以了解笔者的 ​​jvm专题(4) - 【1/3】多线程-基础知识​​ 中的描述。

一、基础实现

1.1、线程启动

每隔5秒打印一行语句，打印3次。

import timedef countdown(n): while n > 0: print('T-minus', n) n -= 1 time.sleep(5)from threading import Threadif __name__ == '__main__': t = Thread(target=countdown, args=(3,)) #t = Thread(target=countdown, args=(3,)， daemon=Ture)首护线程 t.start() #启动 if t.is_alive(): print('Still running') else: print('Completed')

import timefrom threading import Threadclass CountdownThread(Thread): def __init__(self, n): super().__init__() self.n = n def run(self): while self.n > 0: print(f'T-minus: {self.n}') self.n -= 1 time.sleep(5)c = CountdownThread(5)c.start()#下面明确了线程在单独的进程中执行代码c = CountdownTask()p = multiprocessing.Process(target=c.run)p.start()

1.2、线程停止

import timefrom threading import Threadclass CountdownTask: def __init__(self): self._running = True def terminate(self): self._running = False def run(self, n): while self._running and n > 0: print(f'T-minus: {n}') n -= 1 time.sleep(1)c = CountdownTask()t = Thread(target=c.run, args=(10,))t.start()c.terminate()t.join()

1.3、唤醒单个线程

import threadingdef worker(n, sema): # Wait to be signaled sema.acquire() # Do some work print('Working', n)# Create some threadssema = threading.Semaphore(0)nworkers = 10for n in range(nworkers): t = threading.Thread(target=worker, args=(n, sema,)) t.start()

1.4、守护线程

下面的例子可通过 python daemon_exp.py start 运行

import osimport sysimport atexitimport signaldef daemonize(pidfile, *, stdin='/dev/null', stdout='/dev/null', stderr='/dev/null'): if os.path.exists(pidfile): raise RuntimeError('Already running') # First fork (detaches from parent) try: if os.fork() > 0: raise SystemExit(0) except OSError as e: raise RuntimeError('fork #1 failed.') os.chdir('/') os.umask(0) os.setsid() try: if os.fork() > 0: raise SystemExit(0) except OSError as e: raise RuntimeError('fork #2 failed.') # Flush I/O buffers sys.stdout.flush() sys.stderr.flush() # Replace file descriptors for stdin, stdout, and stderr with open(stdin, 'rb', 0) as f: os.dup2(f.fileno(), sys.stdin.fileno()) with open(stdout, 'ab', 0) as f: os.dup2(f.fileno(), sys.stdout.fileno()) with open(stderr, 'ab', 0) as f: os.dup2(f.fileno(), sys.stderr.fileno()) with open(pidfile,'w') as f: print(os.getpid(),file=f) # Arrange to have the PID file removed on exit/signal atexit.register(lambda: os.remove(pidfile)) # Signal handler for termination (required) def sigterm_handler(signo, frame): raise SystemExit(1) signal.signal(signal.SIGTERM, sigterm_handler)def main(): import time sys.stdout.write(f'Daemon started with pid {os.getpid()}\n') while True: sys.stdout.write(f'Daemon Alive! {time.ctime()}\n') time.sleep(10)if __name__ == '__main__': PIDFILE = '/tmp/daemon.pid' if len(sys.argv) != 2: print(f'Usage: {sys.argv[0]} [start|stop]', file=sys.stderr) raise SystemExit(1) if sys.argv[1] == 'start': try: daemonize(PIDFILE, stdout='/tmp/daemon.log', stderr='/tmp/dameon.log') except RuntimeError as e: print(e, file=sys.stderr) raise SystemExit(1) main() elif sys.argv[1] == 'stop': if os.path.exists(PIDFILE): with open(PIDFILE) as f: os.kill(int(f.read()), signal.SIGTERM) else: print('Not running', file=sys.stderr) raise SystemExit(1) else: print(f'Unknown command {sys.argv[1]!r}', file=sys.stderr) raise SystemExit(1)

二、线程判断

2.1、Event对象

Event对象包含一个可由线程设置的信号标志，允许线程等待某些事件的发生。在初始情况下，Event对象中的信号标志被市场为假。如果为真则会唤醒所有对象。

from threading import Thread, Eventimport time# Code to execute in an independent threaddef countdown(n, started_evt): print('countdown starting') started_evt.set() while n > 0: print(f'T-minus: {n}') n -= 1 time.sleep(5)# Create the event object that will be used to signal startupstarted_evt = Event()# Launch the thread and pass the startup eventprint('Launching countdown')t = Thread(target=countdown, args=(10,started_evt))t.start()started_evt.wait()print('countdown is running')

2.2、Condition对象

它的特点是可以单独唤醒某一个被阻塞的线程。下面是一个定时器的代码。

import threadingimport timeclass PeriodicTimer: def __init__(self, interval): self._interval = interval self._flag = 0 self._cv = threading.Condition() def start(self): t = threading.Thread(target=self.run) t.daemon = True t.start() def run(self): ''' Run the timer and notify waiting threads after each interval ''' while True: time.sleep(self._interval) with self._cv: self._flag ^= 1 self._cv.notify_all() def wait_for_tick(self): ''' Wait for the next tick of the timer ''' with self._cv: last_flag = self._flag while last_flag == self._flag: self._cv.wait()# Example use of the timerptimer = PeriodicTimer(5)ptimer.start()# Two threads that synchronize on the timerdef countdown(nticks): while nticks > 0: ptimer.wait_for_tick() print(f'T-minus: {nticks}') nticks -= 1def countup(last): n = 0 while n < last: ptimer.wait_for_tick() print(f'Counting: {n}') n += 1threading.Thread(target=countdown, args=(10,)).start()threading.Thread(target=countup, args=(5,)).start()

三、线程状态

下面是一个线程状态信息保存的一个例子

from socket import socket, AF_INET, SOCK_STREAMimport threadingclass LazyConnection: def __init__(self, address, family=AF_INET, type=SOCK_STREAM): self.address = address self.family = AF_INET self.type = SOCK_STREAM self.local = threading.local() def __enter__(self): if hasattr(self.local, 'sock'): raise RuntimeError('Already connected') self.local.sock = socket(self.family, self.type) self.local.sock.connect(self.address) return self.local.sock def __exit__(self, exc_ty, exc_val, tb): self.local.sock.close() del self.local.sock

使用方法

import threadingfrom functools import partialfrom chapter12.thread_status import LazyConnectiondef test(conn): with conn as s: s.send(b'GET /index.html HTTP/1.0\r\n') s.send(b'Host: python.org\r\n') s.send(b'\r\n') resp = b''.join(iter(partial(s.recv, 8192), b'')) print(f'Got {len(resp)} bytes')if __name__ == '__main__': conn = LazyConnection(('python.org', 80)) t1 = threading.Thread(target=test, args=(conn,)) t2 = threading.Thread(target=test, args=(conn,)) t1.start() t2.start() t1.join() t2.join()

四、计时器

import timeclass Timer: def __init__(self, func=time.perf_counter): self.elapsed = 0.0 self._func = func self._start = None def start(self): if self._start is not None: raise RuntimeError('Already started') self._start = self._func() def stop(self): if self._start is None: raise RuntimeError('Not started') end = self._func() self.elapsed += end - self._start self._start = None def reset(self): self.elapsed = 0.0 @property def running(self): return self._start is not None def __enter__(self): self.start() return self def __exit__(self, *args): self.stop()

使用方法

from app.chapter13.timer_exp import Timerdef countdown(n): while n > 0: n -= 1t = Timer()t.start()countdown(1000000)t.stop()print(f'time used: {t.elapsed}')with t: countdown(1000000)print(f'use with time use: {t.elapsed}')with Timer() as t2: countdown(1000000)print(f't2 time used: {t2.elapsed}')import timet = Timer(time.process_time)with t: countdown(1000000)print(f'process time use: {t.elapsed}')/Users/liudong/PycharmProjects/pythonProject/venv/bin/python /Users/liudong/PycharmProjects/pythonProject/app/chapter13/timer_use.pytime used: 0.052854759use with time use: 0.10032239t2 time used: 0.045731208999999995process time use: 0.04602300000000001进程已结束，退出代码为 0

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。