Multithreading

Python provides the threading module to control thread processing, making it easier to complete multithreaded tasks

Common properties and methods of thread modules

active_count(): Returns the number of thread objects currently alive
enumerate(): Returns a list of thread objects that are currently alive
current_thread(): Returns the current thread object
main_thread()Returns the main thread object
get_ident(): Returns the thread identifier of the current thread
stack_size([size]): returns the stack size used to create the thread
TIMEOUT_MAX: Specifies blocking functions (e.gacquire,wait,wait_forEtc.) the maximum value of the timeout parameter

import threading

thread_number = threading.active_count()
print(thread_number) # 1

curr_thread = threading.main_thread()
main_thread = threading.current_thread()
print(curr_thread == main_thread) # True
Copy the code

2. Create thread objects

threading.Thread(group, target, name, args, kwargs, *, daemon)
Copy the code

group: in order to developThreadGroupThe default value is ReservedNone
target: Call object, default isNoneTo indicate that no method needs to be called
name: Specifies the thread nameThread-NFormat name of
args: The argument (positional argument) passed to the calling object, default to(a)
kwargs: The argument (keyword argument) passed to the calling object. The default is{}
daemon: Indicates whether to set it as a daemon thread. The default value is daemonsNoneRepresents inheriting the properties of the caller

Differences between non-daemons and daemons:

When a program exits, if there are still non-daemons running, the program will wait for all non-daemons to finish before actually exiting
When the program exits, if there are still daemon threads running, the program will force all daemon threads to terminate, causing resources to be improperly released

Common properties and methods of thread objects

name: Thread name
ident: Thread identifier
daemon: Indicates whether it is a daemon thread
is_alive(): Whether the thread is alive

import threading

main_thread = threading.current_thread()
print(main_thread.name) # MainThread
print(main_thread.ident) # 1084
print(main_thread.daemon) # False
print(main_thread.is_alive()) # True
Copy the code

start(): Creates a new thread and calls the method from the new thread, with different threads executing simultaneously

import threading
import time

def sleep(wait) :
    name = threading.current_thread().name
    print(name, 'Start')
    time.sleep(wait)
    print(name, 'Terminated')

def main() :
    name = threading.current_thread().name
    print(name, 'Start')
    worker = threading.Thread(target = sleep, args = (2,))
    worker.start() # start the worker Thread thread-1
    print(name, 'Terminated')

if __name__ == '__main__':
    main() Start MainThread

# Execution result
# MainThread Start
# Thread-1 Start
# MainThread Terminated
# Sleep for 2 seconds
# Thread-1 Terminated
Copy the code

join(): blocks the caller’s thread until the called thread terminates

import threading
import time

def sleep(wait) :
    name = threading.current_thread().name
    print(name, 'Start')
    time.sleep(wait)
    print(name, 'Terminated')

def main() :
    name = threading.current_thread().name
    print(name, 'Start')
    worker = threading.Thread(target = sleep, args = (2,))
    worker.start()
    worker.join() Block MainThread until Thread 1 terminates
    print(name, 'Terminated')

if __name__ == '__main__':
    main()

# Execution result
# MainThread Start
# Thread-1 Start
# Sleep for 2 seconds
# Thread-1 Terminated
# MainThread Terminated
Copy the code

run(): Does not create a thread, equivalent to calling a method directly

import threading
import time

def sleep(wait) :
    name = threading.current_thread().name
    print(name, 'Start')
    time.sleep(wait)
    print(name, 'Terminated')

def main() :
    name = threading.current_thread().name
    print(name, 'Start')
    worker = threading.Thread(target = sleep, args = (2,))
    worker.run() This is equivalent to calling sleep directly from MainThread
    print(name, 'Terminated')

if __name__ == '__main__':
    main()

# Execution result
# MainThread Start
# MainThread Start
# Sleep for 2 seconds
# MainThread Terminated
# MainThread Terminated
Copy the code

setDaemon(): Whether to set it as a daemon thread

import threading
import time

def sleep(wait) :
    name = threading.current_thread().name
    print(name, 'Start')
    time.sleep(wait)
    print(name, 'Terminated')

def main() :
    name = threading.current_thread().name
    print(name, 'Start')
    worker = threading.Thread(target = sleep, args = (2,))
    worker.setDaemon(True) Set worker Thread thread-1 to daemon
    worker.start()
    print(name, 'Terminated') When the program ends, the daemon thread is forcibly terminated

if __name__ == '__main__':
    main()

# Execution result
# MainThread Start
# Thread-1 Start
# MainThread Terminated
Copy the code

4. Maintain thread safety

Because different threads are parallel, if multiple threads modify the same data at the same time, the results will be unpredictable

import threading
import time

num = 0

def add(val) :
    global num
    time.sleep(1)
    num += val
    print(num)

def main() :
    for index in range(1.9):
        worker = threading.Thread(target = add, args = (index,))
        worker.start()

if __name__ == '__main__':
    main()
Copy the code

The result of each run of the program is unknown, so we need some mechanism to make the thread work the way we want it to

(1) Lock objects: threading.lock and threading.rlock

A threading.lock object has only two states, unlocked and unlocked.

Any thread can use the acquire() method to set the lock object to a locked state (called acquiring the lock)

If another thread calls acquire(), it will be blocked

Until any other thread uses the release() method to make the lock object unlocked (called releasing the lock)

But if the lock object is unlocked when the release() method is called, an exception is thrown

Lock object state	Method called	The results of
unlocked	acquire	Sets the lock object to the locked state
locked	acquire	Blocking the current thread
locked	release	Set the lock object to unlocked
unlocked	release	An exception is thrown

Threading.lock has two common methods, acquire() and release().

Acquire (blocking = True, timeout = -1) : Acquire the lock
- blocking: Whether to block the thread. The default value isTrue, which blocks the current thread if no lock is acquired
- timeout: Maximum blocking time. The default value is- 1Is blocked until the lock is released
Release () : Releases the lock

import threading
import time

num = 0
lock = threading.Lock() Declare the lock object

def add(val) :
    lock.acquire() Set the lock object to lock before modifying the data
    global num
    time.sleep(1)
    num += val
    print(num)
    lock.release() Set the lock object to unlocked state after modifying the data

def main() :
    for index in range(1.8):
        worker = threading.Thread(target = add, args = (index,))
        worker.start()

if __name__ == '__main__':
    main()
Copy the code

Threading.rlock has roughly the same functionality as threading.Lock, but what makes threading.rlock special is that:

Multiple calls within the same threadacquire()Method that does not block the thread
How many timesacquire()Method must be used as many times as necessary to obtain the lockrelease()Method to release the lock
A thread passesacquire()Method to obtain the lock and allow only the thread to pass throughrelease()Method to release the lock

(2) Semaphore objects: threading.semaphore

A threading.Semaphore object internally maintains a counter that cannot be less than 0

Any thread can use acquire() to decrease the counter by 1

If the counter is already 0 at this point, the current thread is blocked until the counter is greater than 0

Any thread can use the release() method to increment the counter by 1

counter	Method called	The results of
Greater than zero	acquire	Decrement the counter by 1
Equal to zero	acquire	Blocking the current thread
Greater than or equal to 0	release	Increment the counter by 1

Threading.semaphore has two common methods, acquire() and release().

Acquire (blocking = True, timeout = -1) : Disable the counter by 1
- blocking: Whether to block the thread. The default value isTrue, which means that the current thread will be blocked when the counter is 0
- timeout: Maximum blocking time. The default value is- 1, blocking until the counter is greater than 0
Release () : increments the counter by 1

import threading
import time

num = 0
semaphore = threading.Semaphore(1) The initial value of the counter can be specified. Default is 1

def add(val) :
    semaphore.acquire() # decrease the counter by 1
    global num
    time.sleep(1)
    num += val
    print(num)
    semaphore.release() # increment the counter by 1

def main() :
    for index in range(1.8):
        worker = threading.Thread(target = add, args = (index,))
        worker.start()

if __name__ == '__main__':
    main()
Copy the code

Using semaphores also enables multiple threads to modify a single data at the same time

import threading
import time

semaphore = threading.Semaphore(3)

def run() :
    semaphore.acquire()
    time.sleep(1)
    print(threading.current_thread().name, 'Running')
    semaphore.release()
    
def main() :
    for _ in range(7):
        worker = threading.Thread(target = run)
        worker.start()

if __name__ == '__main__':
    main()
Copy the code

(3) Condition object: threading.condition

The Condition object is wrapped from the lock object. The threading.Condition method is as follows:

Acquire () : Acquire the Lock and call the underlying function (Lock or RLock)
Release () : releases the Lock and calls the underlying function (Lock or RLock)
wait(timeout = None)

After calling this method, release() is called to release the lock, and the current thread is blocked until another thread calls Notify () to wake it up

Then, after being awakened, call Acquire () to try to acquire the lock

If timeout is set, the current thread is automatically woken up after a timeout, even if no other thread calls notify() to wake it up
wait_for(predicate, timeout = None)

After calling this method, the predicate is called first, and if it returns True, execution continues

If False is returned, call release() to release the lock, and then block the current thread until another thread calls Notify () to wake it up

Then, after being woken up, the predicate will also be called, and if it returns False, the predicate will block forever

If True is returned, call Acquire () to attempt to acquire the lock

If timeout is set, the current thread is automatically woken up after a timeout, even if no other thread calls notify() to wake it up
Notify (n = 1) : Wakes up n threads
Notify_all () : Wakes up all threads

import threading
import time

data = 1
condition = threading.Condition()

def isEven() :
    global data
    return data % 2= =0

def wait() :
    condition.acquire()        The lock must be acquired before it can be released
    print('wait_thread enters wait ')
    condition.wait_for(isEven) # release the lock, block the current thread, wait to wake up, regain the lock, and continue execution
    print('wait_thread continues execution ')
    condition.release()        Remember to release the lock once you have regained it

def wake() :
    global data
    condition.acquire() Get the lock first, then modify the data
    data = 2
    print('wake up wait_thread')
    condition.notify()
    condition.release() Once the lock is acquired, release the lock

def main() :
    wait_thread = threading.Thread(target = wait)
    wake_thread = threading.Thread(target = wake)
    wait_thread.start()
    time.sleep(1)
    wake_thread.start()

if __name__ == '__main__':
    main()

# Execution result
# wait_thread enters the wait
# awaken wait_thread
# wait_thread continues execution
Copy the code

(4) Event object: threading.Event

An threading.Event object internally maintains a tag, initially False by default

Threading.event commonly used methods are as follows:

set(): sets the flag toTrue
clear(): sets the flag toFalse
wait(): blocks the current thread until the flag becomesTrue
is_set(): Indicates whether the flag isTrue

import threading
import time

event = threading.Event()

def wait() :
    print(threading.current_thread().name, 'On hold')
    event.wait()
    print(threading.current_thread().name, 'Carry on')

def wake() :
    print('Wake up all threads')
    event.set(a)def main() :
    for _ in range(5):
        wait_thread = threading.Thread(target = wait)
        wait_thread.start()
    time.sleep(1)
    wake_thread = threading.Thread(target = wake)
    wake_thread.start()

if __name__ == '__main__':
    main()

# Execution result
Thread 1 enters the wait
Thread 2 enters the wait
Thread 3 enters the wait
Thread 4 enters the wait
Thread 5 enters the wait
Wake up all threads
# thread-1 continues execution
# thread-2 continues execution
# thread-5 continue execution
# thread-4 continues execution
# thread-3 continue execution
Copy the code

Common properties and methods of thread modules

2. Create thread objects

Common properties and methods of thread objects

4. Maintain thread safety

Related Posts

Elasticsearch (Lucene) index data process

Kubernetes exercises and procedures for Helm

23. Spring Boot database configuration and account information security protection