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

     Thread基本是老生常谈了，说实话在业务开发中真的很少能用了，但又是非常关键的内容。吐槽一点，如果同学们在将来有机会当面试官的话，一定避免面试做火箭实际拧螺丝。一定要从实际出发。与其聊八股文，不如挖下候选人的思维和反应能力。

python的线程会在单独的系统级线程中执行，一旦启动，将独立执行直到返回目标函数。python的线程原理和java差不太多，不清楚的可以了解笔者的 ​​jvm专题(4) - 【1/3】多线程-基础知识​​ 中的描述。

一、基础实现

1.1、线程启动

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

import time

def countdown(n):
while n > 0:
print('T-minus', n)
n -= 1
time.sleep(5)

from threading import Thread

if __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 time
from threading import Thread

class 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 time
from threading import Thread

class 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 threading

def worker(n, sema):
# Wait to be signaled
sema.acquire()

# Do some work
print('Working', n)

# Create some threads
sema = threading.Semaphore(0)
nworkers = 10
for n in range(nworkers):
t = threading.Thread(target=worker, args=(n, sema,))
t.start()

1.4、守护线程

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

import os
import sys

import atexit
import signal

def 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, Event
import time

# Code to execute in an independent thread
def 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 startup
started_evt = Event()

# Launch the thread and pass the startup event
print('Launching countdown')
t = Thread(target=countdown, args=(10,started_evt))
t.start()

started_evt.wait()
print('countdown is running')

2.2、Condition对象

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

import threading
import time

class 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 timer
ptimer = PeriodicTimer(5)
ptimer.start()

# Two threads that synchronize on the timer
def countdown(nticks):
while nticks > 0:
ptimer.wait_for_tick()
print(f'T-minus: {nticks}')
nticks -= 1

def countup(last):
n = 0
while n < last:
ptimer.wait_for_tick()
print(f'Counting: {n}')
n += 1

threading.Thread(target=countdown, args=(10,)).start()
threading.Thread(target=countup, args=(5,)).start()

三、线程状态

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

from socket import socket, AF_INET, SOCK_STREAM
import threading

class 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 threading
from functools import partial
from chapter12.thread_status import LazyConnection

def test(conn):
with conn as s:
s.send(b'GET /index.html HTTP/1.0\r\n')
s.send(b'Host: www.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(('www.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 time

class 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 Timer

def countdown(n):
while n > 0:
n -= 1

t = 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 time
t = 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.py
time used: 0.052854759
use with time use: 0.10032239
t2 time used: 0.045731208999999995
process time use: 0.04602300000000001

进程已结束，退出代码为 0