Concurrent Programming of python Foundation of I
- 2021-12-11 18:20:23
- OfStack
101. Summary of producer and consumer models
1. Process (Process)
It is a running activity of a program with 1 independent function about a data set
2. Threads (Thread)
Is the smallest unit that the operating system can schedule operations. It is contained in the process and is the actual operation unit in the process.
3. Concurrent programming solutions:
1. There are three ways to realize multitasking:
Multi-process mode; Multithreading mode; Multi-process + multi-thread mode
4. Multithreaded implementation (two types)
1. The first function method
# Method packaging - Start multithreading
from threading import Thread
from time import sleep, time
def func1(name):
print("Threading:{} start".format(name))
sleep(3)
print("Threading:{} end".format(name))
if __name__ == '__main__':
# Start time
start = time()
# Create a Thread List
t_list = []
# Loop to create threads
for i in range(10):
t = Thread(target=func1, args=('t{}'.format(i),))
t.start()
t_list.append(t)
# Wait for the thread to end
for t in t_list:
t.join()
# Calculate the usage time
end = time() - start
print(end)
2. The second kind of method packaging
# Class wrapper - Start multithreading
from threading import Thread
from time import sleep, time
class MyThread(Thread):
def __init__(self,name):
Thread.__init__(self)
self.name =name
def run(self):
print("Threading:{} start".format(self.name))
sleep(3)
print("Threading:{} end".format(self.name))
if __name__ == '__main__':
# Start time
start = time()
# Create a Thread List
t_list = []
# Loop to create threads
for i in range(10):
t = MyThread('t{}'.format(i))
t.start()
t_list.append(t)
# Wait for the thread to end
for t in t_list:
t.join()
# Calculate the usage time
end = time() - start
print(end)Note:
The main thread will not wait for the child thread to finish running. If you need to wait, you can use the join () method not to start the thread immediately after join (), which will easily lead to serial running and concurrent failure
5. Daemon and child threads
1. Threads are divided into:
Main thread: The program itself
Sub-thread: A thread opened in a program
2. Daemon thread
The main feature is its life cycle. The main thread dies, and it dies with it
# Class wrapper - Start multithreading
from threading import Thread
from time import sleep, time
class MyThread(Thread):
def __init__(self,name):
Thread.__init__(self)
self.name =name
def run(self):
print("Threading:{} start".format(self.name))
sleep(3)
print("Threading:{} end".format(self.name))
if __name__ == '__main__':
# Start time
start = time()
# Loop to create threads
for i in range(10):
t = MyThread('t{}'.format(i))
t.setDaemon(True)
t.start()
# Calculate the usage time
end = time() - start
print(end)
Step 6 Lock
from threading import Thread
def func1(name):
print('Threading:{} start'.format(name))
global num
for i in range(50000000): # Have a problem
#for i in range(5000): # No problem
num += 1
print('Threading:{} end num={}'.format(name, num))
if __name__ == '__main__':
num =0
# Create a Thread List
t_list = []
# Loop to create threads
for i in range(5):
t = Thread(target=func1, args=('t{}'.format(i),))
t.start()
t_list.append(t)
# Wait for the thread to end
for t in t_list:
t.join()When Python uses threads, GIL locks will be released regularly, and sleep will be released at this time, so the above problems will occur. Facing this problem, if we want to solve this problem, we can use Lock lock to solve this problem (the purpose of locking is to ensure data security)
from threading import Thread,Lock
def func1(name):
# Acquisition lock
lock.acquire()
with lock:
global count
for i in range(100000):
count += 1
# Release lock
lock.release()
if __name__ == "__main__":
count = 0
t_list = []
# Create a lock object
lock = Lock()
for i in range(10):
t = Thread(target=func1,args=(f't{i+1}',))
t.start()
t_list.append(t)
for t in t_list:
t.join()
print(count)
7. Deadlock
from threading import Thread, Lock #Lock Lock Synchronous lock Mutual exclusion lock
from time import sleep
def fun1():
lock1.acquire()
print('fun1 Get the keyboard ')
sleep(2)
lock2.acquire()
print('fun1 Get the mouse ')
lock2.release()
print('fun1 Release the mouse ')
lock1.release()
print('fun1 Release the keyboard ')
def fun2():
lock2.acquire()
print('fun2 Get the mouse ')
lock1.acquire()
print('fun2 Get the keyboard ')
lock1.release()
print('fun2 Release the keyboard ')
lock2.release()
print('fun2 Release the mouse ')
if __name__ == '__main__':
lock1 = Lock()
lock2 = Lock()
t1 = Thread(target=fun1)
t2 = Thread(target=fun2)
t1.start()
t2.start()
from threading import RLock
'''
Lock Lock Synchronous lock Mutual exclusion lock
RLock Recursive lock
'''
def func1():
lock.acquire()
print('func1 Acquisition lock ')
func2()
lock.release()
print('func1 Release lock ')
def func2():
lock.acquire()
print('func2 Acquisition lock ')
lock.release()
print('func2 Release lock ')
def func3():
func1()
func2()
if __name__ == "__main__":
#lock = Lock() Errors will be generated
lock = RLock()
func3()
8. Semaphore (Semaphore)
We all know that in the case of locking, the program becomes serial, that is, single-threaded, and sometimes, when we don't need to consider data security, we want to avoid opening too many threads in order to avoid business. We can set the specified number of threads through the semaphore (Semaphore). (For example, if the elevator can only carry 3 people at a time, then only 3 people can take it at the same time, and others can only wait for others to finish it.)
from time import sleep
from threading import Thread
from threading import BoundedSemaphore
def index(num):
lock.acquire()
print(f' No. 1 {num} Individual ride! ! ')
sleep(2)
lock.release()
if __name__ == "__main__":
lock = BoundedSemaphore(3)
for i in range(10):
t = Thread(target=index,args=(f'{i+1}',))
t.start()
9. Event (Event)
Event () can create an event management flag, which defaults to False (event), and event objects have four main methods that can be called:
1.
event.wait(timeout=None)：
The thread calling this method will be blocked. If timeout parameter is set, the thread will stop blocking and continue to execute after timeout.
2.
event.set()：
Set the flag of event to True, and all threads that call the wait method will wake up;
3.
event.clear()：
Set the flag of event to False and all threads calling the wait method will be blocked;
4.
event.is_set()：
It is judged whether the flag of event is True.
10. Thread Communication-Queue
Thread safety is a concept in computer program code when multithreaded programming. In a program with multiple threads executing in parallel with shared data, thread-safe code can ensure that each thread can execute normally and correctly through synchronization mechanism, and there will be no unexpected situations such as data pollution
Benefits of the queue used by 1:
1. Security
2. Decoupling
STEP 3 Increase efficiency
Common methods in 2Queue module:
Synchronized, thread-safe queue classes are provided in the Queue module of Python, including FIFO (first-in, first-out) queue Queue, LIFO (last-in, first-out) queue LifoQueue, and priority queue PriorityQueue. These queues implement lock primitives and can be used directly in multiple threads. You can use queues to synchronize between threads
Queue.qsize()
Returns the size of the queue
Queue.empty()
If the queue is empty, True is returned, otherwise False
Queue.full()
If the queue is full, return True, otherwise False
Queue.full
And
maxsize
Size correspondence
Queue.get([block[, timeout]])
Get queue, timeout wait time
event.set()：0
Quite
Queue.get(False)
Queue.put(item)
Write queue, timeout wait time
Queue.put_nowait(item
) Equivalent to Queue. put (item, False)
Queue.task_done()
After 1 task is completed, the Queue.task_done () function sends a signal to the queue where the task has been completed
Queue.join()
It actually means waiting until the queue is empty before performing other operations
101. Producer and consumer models
Producer-consumer model is to solve the strong coupling problem between producer and consumer through a container. Producers and consumers do not communicate directly with each other, The communication is carried out through the blocking queue, so the producer does not have to wait for the consumer to process the data after producing it, and throws it directly to the blocking queue. The consumer does not ask the producer for the data, but takes it directly from the blocking queue. The blocking queue is equivalent to a buffer, which balances the processing capacity of the producer and the consumer.
from threading import Thread
from queue import Queue
from time import sleep
def producer():
num = 1
while True:
print(f' Produced {num} Pikachu ')
qe.put(f'{num} Pikachu ')
num += 1
sleep(1)
def consumer():
print(' Buy {}'.format(qe.get()))
sleep(2)
if __name__ == "__main__":
# Container for sharing data
qe= Queue(maxsize=5)
# Create a producer thread
t1 = Thread(target = producer)
# Create a consumer thread
t2 = Thread(target = consumer)
# Create a consumer thread
t3 = Thread(target = consumer)
# Begin work
t1.start()
t2.start()
t3.start()
Summarize
This article is here, I hope to give you help, but also hope that you can pay more attention to this site more content!