python threading模块

# http://www.runoob.com/python3/python3-multithreading.html
# https://docs.python.org/3/library/threading.html
# http://yoyzhou.github.io/blog/2013/02/28/python-threads-synchronization-locks/
# http://www.jb51.net/article/119414.htm
# http://blog.csdn.net/tomato__/article/details/46606045
# https://www.cnblogs.com/Ajen-lq/p/5325827.html


import threading  # module
import socket
import random
import time
import queue
import requests


# ['get_ident',           1、返回线程号，一个数字
#  'active_count',        2、存活的线程数
#  'Condition',           3、条件锁
#  'current_thread',      4、返回当前线程的对象
#  'enumerate',           5、存活线程的列表
#  'main_thread',         6、返回主线程对象，python解释器启动的线程
#  'TIMEOUT_MAX',
#  'Event',               8、事件锁
#  'Lock',                9、互斥锁
#  'RLock',               10、重入锁，递归锁
#  'Semaphore',           11、信号量锁
#  'BoundedSemaphore',    12、有限信号量类
#  'Thread',              13、线程
#  'Barrier',             14、栅栏
#  'BrokenBarrierError',
#  'Timer',               16、定时任务
#  'ThreadError',
#  'setprofile',
#  'settrace',
#  'local',               20、
#  'stack_size']
# Python线程同步机制: Locks, RLocks, Semaphores, Conditions, Events和Queues


# 1、get_ident
def foo():
    print('threading identify no: %s' % threading.get_ident())


t1 = threading.Thread(target=foo)
t2 = threading.Thread(target=foo)
t3 = threading.Thread(target=foo)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()


# 2、active_count
def foo():
    time.sleep(1)


t1 = threading.Thread(target=foo)
t2 = threading.Thread(target=foo)
t3 = threading.Thread(target=foo)
t1.start()
print(threading.active_count())    # 2
t2.start()
print(threading.active_count())    # 3
t3.start()
print(threading.active_count())    # 4
t1.join()
t2.join()
t3.join()


# 3、Condition 一个线程等待特定条件，而另一个线程发出特定条件满足的信号
# class Condition(builtins.object) Class that implements a condition variable.
# A condition variable allows one or more threads to wait until they are notified by another thread.
# notify(self, n=1) Wake up one or more threads waiting on this condition, if any.
# notify_all(self) Wake up all threads waiting on this condition.
# wait(self, timeout=None) Wait until notified or until a timeout occurs.
# wait_for(self, predicate, timeout=None) Wait until a condition evaluates to True.
# wait()方法释放锁，然后阻塞直到另一个线程调用notify()或者notify_all()唤醒它。唤醒后，wait()重新获取锁并返回。它也可以指定一个超时时间。
# notify()方法唤醒等待线程中的一个；notify_all()方法唤醒所有等待线程。
# 注意：notify()和notify_all()方法不释放锁；这意味着唤醒的线程或者线程组将不会从wait()调用中立即返回。
# 一个条件变量允许一个或多个线程等待，直到他们被另一个线程通知。
# wait()一旦唤醒或者超时，它重新请求锁并返回。


class Producer(threading.Thread):

    def __init__(self, integers, condition):
        threading.Thread.__init__(self)
        self.integers = integers
        self.condition = condition

    def run(self):
        while True:
            integer = random.randint(0, 256)
            with self.condition:  # 获取条件锁
                print('condition acquired by %s' % self.name)
                self.integers.append(integer)
                print('%d appended to list by %s' % (integer, self.name))
                print('condition notified by %s' % self.name)
                self.condition.notify()    # 唤醒消费者线程
                print('condition released by %s' % self.name)
            time.sleep(1)  # 暂停1秒钟


class Consumer(threading.Thread):
    def __init__(self, integers, condition):
        threading.Thread.__init__(self)
        self.integers = integers
        self.condition = condition

    def run(self):
        while True:
            with self.condition:  # 获取条件锁
                print('condition acquired by %s' % self.name)
                while True:
                    if self.integers:  # 判断是否有整数
                        integer = self.integers.pop()
                        print('%d popped from list by %s' % (integer, self.name))
                        break
                    print('condition wait by %s' % self.name)
                    self.condition.wait()  # 等待商品，并且释放资源让生产者执行
                print('condition released by %s' % self.name)


def main():
    integers = []
    condition = threading.Condition()
    t1 = Producer(integers, condition)
    t2 = Consumer(integers, condition)
    t1.start()
    t2.start()
    t1.join()
    t2.join()


main()


# 4、current_thread
def foo():
    print(threading.current_thread())


t1 = threading.Thread(target=foo)
t1.start()
t1.join()


# 5、enumerate
def foo():
    time.sleep(1)


t1 = threading.Thread(target=foo)
t2 = threading.Thread(target=foo)
t3 = threading.Thread(target=foo)
t1.start()
t2.start()
t3.start()
print(threading.enumerate())
t1.join()
t2.join()
t3.join()


# 6、main_thread
def foo():
    time.sleep(1)


t1 = threading.Thread(target=foo)
t1.start()
print(threading.main_thread())
t1.join()


# 8、Event 事件锁 一个线程发送/传递事件，另外的线程（多线程）等待事件的触发。
# class Event(builtins.object) Class implementing event objects.
# Events manage a flag that can be set to true with the set() method and reset to false with the clear() method.
# The wait() method blocks until the flag is true.  The flag is initially false.
# set(self) Set the internal flag to true. All threads waiting for it to become true are awakened.
#           Threads that call wait() once the flag is true will not block at all.
# clear(self) Reset the internal flag to false.
# is_set(self) Return true if and only if the internal flag is true.
# wait(self, timeout=None) Block until the internal flag is true.
# isSet = is_set(self)


class Producer(threading.Thread):
    def __init__(self, integers, event):
        threading.Thread.__init__(self)
        self.integers = integers
        self.event = event

    def run(self):
        while True:
            integer = random.randint(0, 256)
            self.integers.append(integer)
            print('%d appended to list by %s' % (integer, self.name))
            print('event set by %s' % self.name)
            self.event.set()            # 设置事件
            self.event.clear()          # 发送事件
            print('event cleared by %s' % self.name)
            time.sleep(1)


class Consumer(threading.Thread):
    def __init__(self, integers, event):
        threading.Thread.__init__(self)
        self.integers = integers
        self.event = event

    def run(self):
        while True:
            self.event.wait()      # 等待事件被触发
            # self.event.wait(timeout=2)  # 等待事件被触发，超时之后，不再阻塞，继续执行
            try:
                integer = self.integers.pop()
                print('%d popped from list by %s' % (integer, self.name))
            except IndexError:
                # catch pop on empty list
                time.sleep(1)


def main():
    integers = []
    event = threading.Event()
    t1 = Producer(integers, event)
    t2 = Consumer(integers, event)
    t1.start()
    t2.start()
    t1.join()
    t2.join()


main()


# 9、Lock互斥锁
# class lock(builtins.object) A lock object is a synchronization primitive. To create a lock, call threading.Lock().
# acquire() -- lock the lock, possibly blocking until it can be obtained
# release() -- unlock of the lock
# locked() -- test whether the lock is currently locked
# acquire(...) acquire_lock(...) acquire(blocking=True, timeout=-1)
# locked(...) locked_lock(...) locked() -> bool
# release(...) release_lock(...) release()


class FetchUrls(threading.Thread):
    def __init__(self, urls, output, lock):
        threading.Thread.__init__(self)
        self.urls = urls
        self.output = output
        self.lock = lock

    def run(self):
        while self.urls:
            url = self.urls.pop()
            req = requests.get(url)
            with self.lock:
                self.output.write(req.content)
            print('write done by %s' % self.name)
            print('URL %s fetched by %s' % (url, self.name))


def main():
    lock = threading.Lock()
    urls1 = ['http://www.baidu.com', 'http://www.sina.com']
    urls2 = ['http://www.sohu.com', 'http://www.youku.com']
    f = open('output.txt', 'wb+')
    t1 = FetchUrls(urls1, f, lock)
    t2 = FetchUrls(urls2, f, lock)
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    f.close()


main()


# 10、RLock可重入锁
# RLock是可重入锁（reentrant lock），acquire()能够不被阻塞的被同一个线程调用多次。要注意的是release()需要调用与acquire()相同的次数才能释放锁。


# 11、Semaphore信号量锁
# 信号量同步基于内部计数器，每调用一次acquire()，计数器减1；每调用一次release()，计数器加1.当计数器为0时，acquire()调用被阻塞。
# class Semaphore(builtins.object) This class implements semaphore objects.
# Semaphores manage a counter representing the number of release() calls minus the number of acquire() calls, plus an initial value.
# The acquire() method blocks if necessary until it can return without making the counter negative.
# If not given, value defaults to 1.
# acquire(self, blocking=True, timeout=None) Acquire a semaphore, decrementing the internal counter by one. 超时或者不阻塞返回false
# release(self) Release a semaphore, incrementing the internal counter by one.


class MyThread(threading.Thread):
    def __init__(self, semaphore, list_):
        threading.Thread.__init__(self)
        self.semaphore = semaphore
        self.list_ = list_

    def run(self):
        with self.semaphore:
            self.list_.append(2)


def main():
    list_ = [1, 2, 3]
    semaphore = threading.Semaphore(2)    # 允许同时2个线程访问共享资源，默认是1
    t1 = MyThread(semaphore, list_)
    t2 = MyThread(semaphore, list_)
    t3 = MyThread(semaphore, list_)
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print(list_)


main()


# 12、BoundedSemaphore有限信号量类
# “有限”(bounded)信号量类，可以确保release()方法的调用次数不能超过给定的初始信号量数值(value参数)
# 重写release方法
# If the number of releases exceeds the number of acquires, raise a ValueError.


# 13、Thread
# class Thread(builtins.object) A class that represents a thread of control.
# This class can be safely subclassed in a limited fashion. There are two ways to specify the activity:
# by passing a callable object to the constructor, or by overriding the run() method in a subclass.
# __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)
# group 为将来线程组扩展留的
# daemon 默认非守护线程
# getName(self)  方法
# isDaemon(self) 方法
# is_alive(self) 是否存活，run开始之后到run结束之前是True
# join(self, timeout=None) Wait until the thread terminates.
# run(self) Method representing the thread's activity.
# setDaemon(self, daemonic)
# setName(self, name)
# start(self)
# daemon 属性
# ident  属性
# name   属性

# 无join，无daemon
def foo():
    time.sleep(3)


t1 = threading.Thread(target=foo)
t1.start()
print('main thread go on')  # 最后再等待线程，释放线程的资源，这里阻塞


# 有join
def foo():
    time.sleep(3)


t1 = threading.Thread(target=foo)
t1.start()
t1.join()   # 等待线程结束，抓线程再继承执行，这里阻塞
print('main thread not go on')


# 有daemon
def foo():
    while True:
        print('abc')


t1 = threading.Thread(target=foo)
t1.setDaemon(True)
t1.start()
print('main thread go on')   # 主线程结束，要求子线程同时也结束，这里不阻塞了


# 14、Barrier
# class Barrier(builtins.object) Implements a Barrier. Useful for synchronizing a fixed number
# of threads at known synchronization points.
# Threads block on 'wait()' and are simultaneously once they have all made that call.
# __init__(self, parties, action=None, timeout=None)
# abort(self) Place the barrier into a 'broken' state.     方法，终止栅栏，wait会抛出异常
# reset(self) Reset the barrier to the initial state.      方法，恢复栅栏
# wait(self, timeout=None) Wait for the barrier.           方法，等待数量达到parties，一起通过栅栏
# broken Return True if the barrier is in a broken state.                    属性，检查栅栏的状态，True是坏
# n_waiting Return the number of threads currently waiting at the barrier.   属性，正在wait的数量
# parties Return the number of threads required to trip the barrier.         属性，parties的数量


def server(b):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(("192.168.99.156", 5003))
    s.listen(5)

    while True:
        # print(b.broken)   # False
        # b.abort()
        # print(b.broken)   # True
        # b.reset()

        b.wait()
        print(threading.get_ident())
        new_socket, client_addr = s.accept()
        new_socket.sendall(bytes("Hello world", encoding="utf-8"))
        new_socket.close()
        time.sleep(3)


def client(b):
    while True:
        b.wait()       # 超时 raise BrokenBarrierError
        print(threading.get_ident())
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.connect(("192.168.99.156", 5003))
        ret = str(s.recv(1024), encoding="utf-8")
        print(ret)
        time.sleep(0.1)


def callback():
    print('Barrier')


def main():
    b = threading.Barrier(2, action=callback, timeout=5)        # 超时 raise BrokenBarrierError, 通过一次栅栏调用一次action
    t1 = threading.Thread(target=server, args=(b, ))
    t2 = threading.Thread(target=client, args=(b, ))
    t1.start()
    t2.start()
    t1.join()
    t2.join()


main()


# 16、Timer定时任务
# class Timer(Thread) Call a function after a specified number of seconds。
# __init__(self, interval, function, args=None, kwargs=None) 继承Thread
# 新增方法：cancel(self)
# 重写方法：run(self)
# 其他方法同Thread


def func():
    print('abc')


def main():
    t = threading.Timer(2, func)
    t.start()


main()


# 20、local
localVal = threading.local()
localVal.val = "Main-Thread"   # 每个线程单独创建内存空间来存储
a = 10


def print_student():
    global a
    print('%s (in %s)' % (localVal.val, threading.current_thread().name))
    a += 100                   # 操作的是主线程的变量


def print_thread(name):
    localVal.val = name        # 每个线程单独创建内存空间来存储
    print_student()


def main():
    t1 = threading.Thread(target=print_thread, args=('One',), name='Thread-A')
    t2 = threading.Thread(target=print_thread, args=('Two',), name='Thread-B')
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print(localVal.val)
    print(a)


main()


# 21、使用队列同步线程间共享资源
# 队列 Queue
class Producer(threading.Thread):
    def __init__(self, q):
        threading.Thread.__init__(self)
        self.queue_ = q

    def run(self):
        while True:
            integer = random.randint(0, 256)
            self.queue_.put(integer)  # 将生成的整数添加到队列
            print('%d put to queue by %s' % (integer, self.name))
            time.sleep(1)


class Consumer(threading.Thread):
    def __init__(self, q):
        threading.Thread.__init__(self)
        self.queue_ = q

    def run(self):
        while True:
            integer = self.queue_.get()
            print('%d popped from list by %s' % (integer, self.name))
            self.queue_.task_done()


def main():
    q = queue.Queue(20)
    t1 = Producer(q)
    t2 = Consumer(q)
    t1.start()
    t2.start()
    t1.join()
    t2.join()


main()