三 python并发编程之多线程-重点

#_*_coding:utf-8_*_

#!/usr/bin/env python

import multiprocessing

import threading

import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)

s.bind(('127.0.0.1',8080))

s.listen(5)

def action(conn):

    while True:

        data=conn.recv(1024)

        print(data)

        conn.send(data.upper())

if __name__ == '__main__':

    while True:

        conn,addr=s.accept()

        p=threading.Thread(target=action,args=(conn,))

        p.start()

#_*_coding:utf-8_*_

#!/usr/bin/env python

import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)

s.connect(('127.0.0.1',8080))

while True:

    msg=input('>>: ').strip()

    if not msg:continue

    s.send(msg.encode('utf-8'))

    data=s.recv(1024)

    print(data)

from threading import Thread

msg_l=[]

format_l=[]

def talk():

    while True:

        msg=input('>>: ').strip()

        if not msg:continue

        msg_l.append(msg)

def format_msg():

    while True:

        if msg_l:

            res=msg_l.pop()

            format_l.append(res.upper())

def save():

    while True:

        if format_l:

            with open('db.txt','a',encoding='utf-8') as f:

                res=format_l.pop()

                f.write('%s\n' %res)

if __name__ == '__main__':

    t1=Thread(target=talk)

    t2=Thread(target=format_msg)

    t3=Thread(target=save)

    t1.start()

    t2.start()

    t3.start()

from threading import Thread

import time

def foo():

    print(123)

    time.sleep(1)

    print("end123")

def bar():

    print(456)

    time.sleep(3)

    print("end456")

t1=Thread(target=foo)

t2=Thread(target=bar)

t1.daemon=True

t1.start()

t2.start()

print("main-------")

from threading import Thread,Lock

import os,time

def work():

    global n

    lock.acquire()

    temp=n

    time.sleep(0.1)

    n=temp-1

    lock.release()

if __name__ == '__main__':

    lock=Lock()

    n=100

    l=[]

    for i in range(100):

        p=Thread(target=work)

        l.append(p)

        p.start()

    for p in l:

        p.join()

    print(n) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全

#不加锁:并发执行,速度快,数据不安全

from threading import current_thread,Thread,Lock

import os,time

def task():

    global n

    print('%s is running' %current_thread().getName())

    temp=n

    time.sleep(0.5)

    n=temp-1

if __name__ == '__main__':

    n=100

    lock=Lock()

    threads=[]

    start_time=time.time()

    for i in range(100):

        t=Thread(target=task)

        threads.append(t)

        t.start()

    for t in threads:

        t.join()

    stop_time=time.time()

    print('主:%s n:%s' %(stop_time-start_time,n))

'''

Thread-1 is running

Thread-2 is running

......

Thread-100 is running

主:0.5216062068939209 n:99

'''

#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全

from threading import current_thread,Thread,Lock

import os,time

def task():

    #未加锁的代码并发运行

    time.sleep(3)

    print('%s start to run' %current_thread().getName())

    global n

    #加锁的代码串行运行

    lock.acquire()

    temp=n

    time.sleep(0.5)

    n=temp-1

    lock.release()

if __name__ == '__main__':

    n=100

    lock=Lock()

    threads=[]

    start_time=time.time()

    for i in range(100):

        t=Thread(target=task)

        threads.append(t)

        t.start()

    for t in threads:

        t.join()

    stop_time=time.time()

    print('主:%s n:%s' %(stop_time-start_time,n))

'''

Thread-1 is running

Thread-2 is running

......

Thread-100 is running

主:53.294203758239746 n:0

'''

#有的人可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊

#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是

#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的

#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.

from threading import current_thread,Thread,Lock

import os,time

def task():

    time.sleep(3)

    print('%s start to run' %current_thread().getName())

    global n

    temp=n

    time.sleep(0.5)

    n=temp-1

if __name__ == '__main__':

    n=100

    lock=Lock()

    start_time=time.time()

    for i in range(100):

        t=Thread(target=task)

        t.start()

        t.join()

    stop_time=time.time()

    print('主:%s n:%s' %(stop_time-start_time,n))

'''

Thread-1 start to run

Thread-2 start to run

......

Thread-100 start to run

主:350.6937336921692 n:0 #耗时是多么的恐怖

'''

from threading import Thread,Lock

import time

mutexA=Lock()

mutexB=Lock()

class MyThread(Thread):

    def run(self):

        self.func1()

        self.func2()

    def func1(self):

        mutexA.acquire()

        print('\033[41m%s 拿到A锁\033[0m' %self.name)

        mutexB.acquire()

        print('\033[42m%s 拿到B锁\033[0m' %self.name)

        mutexB.release()

        mutexA.release()

    def func2(self):

        mutexB.acquire()

        print('\033[43m%s 拿到B锁\033[0m' %self.name)

        time.sleep(2)

        mutexA.acquire()

        print('\033[44m%s 拿到A锁\033[0m' %self.name)

        mutexA.release()

        mutexB.release()

if __name__ == '__main__':

    for i in range(10):

        t=MyThread()

        t.start()

'''

Thread-1 拿到A锁

Thread-1 拿到B锁

Thread-1 拿到B锁

Thread-2 拿到A锁

然后就卡住，死锁了

'''

from threading import Thread,Semaphore

import threading

import time

# def func():

#     if sm.acquire():

#         print (threading.currentThread().getName() + ' get semaphore')

#         time.sleep(2)

#         sm.release()

def func():

    sm.acquire()

    print('%s get sm' %threading.current_thread().getName())

    time.sleep(3)

    sm.release()

if __name__ == '__main__':

    sm=Semaphore(5)

    for i in range(23):

        t=Thread(target=func)

        t.start()

from threading import Thread,Event

import threading

import time,random

def conn_mysql():

    count=1

    while not event.is_set():

        if count > 3:

            raise TimeoutError('链接超时')

        print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))

        event.wait(0.5)

        count+=1

    print('<%s>链接成功' %threading.current_thread().getName())

def check_mysql():

    print('\033[45m[%s]正在检查mysql\033[0m' % threading.current_thread().getName())

    time.sleep(random.randint(2,4))

    event.set()

if __name__ == '__main__':

    event=Event()

    conn1=Thread(target=conn_mysql)

    conn2=Thread(target=conn_mysql)

    check=Thread(target=check_mysql)

    conn1.start()

    conn2.start()

    check.start()

def condition_func():

    ret = False

    inp = input('>>>')

    if inp == '':

        ret = True

    return ret

def run(n):

    con.acquire()

    con.wait_for(condition_func)

    print("run the thread: %s" %n)

    con.release()

if __name__ == '__main__':

    con = threading.Condition()

    for i in range(10):

        t = threading.Thread(target=run, args=(i,))

        t.start()

from threading import Timer

import random,time

class Code:

    def __init__(self):

        self.make_cache()

    def make_cache(self,interval=5):

        self.cache=self.make_code()

        print(self.cache)

        self.t=Timer(interval,self.make_cache)

        self.t.start()

    def make_code(self,n=4):

        res=''

        for i in range(n):

            s1=str(random.randint(0,9))

            s2=chr(random.randint(65,90))

            res+=random.choice([s1,s2])

        return res

    def check(self):

        while True:

            inp=input('>>: ').strip()

            if inp.upper() ==  self.cache:

                print('验证成功',end='\n')

                self.t.cancel()

                break

if __name__ == '__main__':

    obj=Code()

    obj.check()

import queue

q=queue.Queue()

q.put('first')

q.put('second')

q.put('third')

print(q.get())

print(q.get())

print(q.get())

'''

结果(先进先出):

first

second

third

'''

import queue

q=queue.LifoQueue()

q.put('first')

q.put('second')

q.put('third')

print(q.get())

print(q.get())

print(q.get())

'''

结果(后进先出):

third

second

first

'''

import queue

q=queue.LifoQueue()

q.put('first')

q.put('second')

q.put('third')

print(q.get())

print(q.get())

print(q.get())

'''

结果(后进先出):

third

second

first

'''

Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.

The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).

exception queue.Empty

Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.

exception queue.Full

Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.

Queue.qsize()

Queue.empty() #return True if empty

Queue.full() # return True if full

Queue.put(item, block=True, timeout=None)

Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).

Queue.put_nowait(item)

Equivalent to put(item, False).

Queue.get(block=True, timeout=None)

Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).

Queue.get_nowait()

Equivalent to get(False).

Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.

Queue.task_done()

Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).

Raises a ValueError if called more times than there were items placed in the queue.

Queue.join() block直到queue被消费完毕