Esercizi di Multiprocessing in Python
Impara a gestire il multiprocessing in Python con questi esercizi introduttivi.
Esercizio 1
Creare un processo che stampa i numeri da 1 a 10.
from multiprocessing import Process
def print_numbers(): for i in range(1, 11): print(i)
if __name__ == '__main__': process = Process(target=print_numbers) process.start() process.join()
Esercizio 2
Scrivere un programma che avvia due processi; uno stampa numeri pari e l'altro numeri dispari fino a 10.
from multiprocessing import Process
def print_even(): for i in range(2, 11, 2): print(f"Pari: {i}")
def print_odd(): for i in range(1, 11, 2): print(f"Dispari: {i}")
if __name__ == '__main__': process1 = Process(target=print_even) process2 = Process(target=print_odd)
process1.start() process2.start()
process1.join() process2.join()
Esercizio 3
Implementare un processo che esegue un countdown da 10 a 1 e stampa "Fine!" al termine.
from multiprocessing import Process
def countdown(): for i in range(10, 0, -1): print(i) print("Fine!")
if __name__ == '__main__': process = Process(target=countdown) process.start() process.join()
Esercizio 4
Creare un processo che esegue un'operazione matematica semplice e utilizza una Pipe per inviare il risultato al processo principale.
from multiprocessing import Process, Pipe
def calculate_sum(conn): total = sum(range(1, 11)) conn.send(total) conn.close()
if __name__ == '__main__': parent_conn, child_conn = Pipe() process = Process(target=calculate_sum, args=(child_conn,)) process.start() print("La somma è:", parent_conn.recv()) process.join()
Esercizio 5
Utilizzare un processo per monitorare lo stato di un'altra variabile modificata nel programma principale attraverso un Manager.
from multiprocessing import Process, Manager
def monitor_status(d): while d['run']: print("Monitoraggio in corso...") time.sleep(1) print("Monitoraggio terminato.")
if __name__ == '__main__': with Manager() as manager: d = manager.dict() d['run'] = True process = Process(target=monitor_status, args=(d,)) process.start()
time.sleep(5) d['run'] = False
process.join()
Esercizio 6
Scrivere un programma che avvia tre processi per stampare "Hello, World!" 5 volte ciascuno.
from multiprocessing import Process
def say_hello(): for _ in range(5): print("Hello, World!")
if __name__ == '__main__': processes = [Process(target=say_hello) for _ in range(3)]
for p in processes: p.start()
for p in processes: p.join()
Esercizio 7
Creare un processo che stampa i primi 20 numeri della sequenza di Fibonacci.
from multiprocessing import Process
def fibonacci(): a, b = 0, 1 for _ in range(20): print(a) a, b = b, a + b
if __name__ == '__main__': process = Process(target=fibonacci) process.start() process.join()
Esercizio 8
Implementare due processi che modificano una lista condivisa aggiungendo elementi, utilizzando un Manager per gestire la lista.
from multiprocessing import Process, Manager
def add_items(shared_list): for i in range(5): shared_list.append(i) print(f"Elemento aggiunto: {i}")
if __name__ == '__main__': with Manager() as manager: shared_list = manager.list()
p1 = Process(target=add_items, args=(shared_list,)) p2 = Process(target=add_items, args=(shared_list,))
p1.start() p2.start()
p1.join() p2.join()
print("Lista finale:", list(shared_list))
Esercizio 9
Utilizzare multiprocessing per calcolare simultaneamente somme di parti diverse di una stessa lista.
from multiprocessing import Process, Array
def partial_sum(arr, start, end, result, index): result[index] = sum(arr[start:end])
if __name__ == '__main__': arr = list(range(1, 101)) result = Array('i', 2)
p1 = Process(target=partial_sum, args=(arr, 0, 50, result, 0)) p2 = Process(target=partial_sum, args=(arr, 50, 100, result, 1))
p1.start() p2.start()
p1.join() p2.join()
total_sum = sum(result) print("Somma totale:", total_sum)
Esercizio 10
Creare un programma che utilizza i processi per stampare alternativamente "Ping" e "Pong" usando le Queue.
from multiprocessing import Process, Queueimport time
def ping(queue): while True: queue.put("Ping") time.sleep(1)
def pong(queue): while True: print(queue.get()) print("Pong") time.sleep(1)
if __name__ == '__main__': queue = Queue()
p1 = Process(target=ping, args=(queue,)) p2 = Process(target=pong, args=(queue,))
p1.start() p2.start()
p1.join(timeout=5) p2.join(timeout=5)
p1.terminate() p2.terminate()
print("Ping Pong terminato.")