Python 機能編①

Pythonで使用する機能について解説します。

目次

1. マルチプロセッシング（Multiprocessing）
2. パイプ（Pipe）
3. シグナル（Signal）
4. バリア（Barrier）
5. セマフォ（Semaphore）
6. デーモンスレッド（Daemon Thread）
7. ノンブロッキングI/O（Non-blocking I/O）
8. ストリーム処理（Stream Processing）
9. 継続（Continuation）
10. 依存性注入（Dependency Injection）

1. マルチプロセッシング（Multiprocessing）
文字列は、テキストデータを格納するためのデータ型です。シングルクォートかダブルクォートで囲んで表します。

Multiprocessing.py

from multiprocessing import Process

def print_square(num):
    print(f"Square: {num * num}")

def print_cube(num):
    print(f"Cube: {num * num * num}")

if __name__ == "__main__":
    p1 = Process(target=print_square, args=(10,))
    p2 = Process(target=print_cube, args=(10,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()
# Output:
# Square: 100
# Cube: 1000

2. パイプ（Pipe）
文字列は、テキストデータを格納するためのデータ型です。シングルクォートかダブルクォートで囲んで表します。

Pipe.py

from multiprocessing import Process, Pipe

def sender(conn):
    conn.send("Hello from child process")
    conn.close()

def receiver(conn):
    print(conn.recv())
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p1 = Process(target=sender, args=(child_conn,))
    p2 = Process(target=receiver, args=(parent_conn,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()
# Output: Hello from child process

3. シグナル（Signal）
シグナルは、プロセス間で非同期のイベント通知を行うための仕組みです。Pythonの signal モジュールを使用します。

Signal.py

import signal
import time

def handler(signum, frame):
    print("Signal handler called with signal", signum)

signal.signal(signal.SIGALRM, handler)
signal.alarm(2)

print("Waiting for signal...")
time.sleep(3)
# Output: Signal handler called with signal 14
# Waiting for signal...

4. バリア（Barrier）
バリアは、スレッドやプロセスが同期するための同期プリミティブです。threading モジュールの Barrier クラスを使用します。

Barrier.py

from threading import Barrier, Thread
import time

def worker(barrier, i):
    print(f"Worker {i} waiting at barrier")
    barrier.wait()
    print(f"Worker {i} passed the barrier")

barrier = Barrier(3)
threads = [Thread(target=worker, args=(barrier, i)) for i in range(3)]

for t in threads:
    t.start()

for t in threads:
    t.join()
# Output:
# Worker 0 waiting at barrier
# Worker 1 waiting at barrier
# Worker 2 waiting at barrier
# Worker 0 passed the barrier
# Worker 1 passed the barrier
# Worker 2 passed the barrier

5. セマフォ（Semaphore）
セマフォは、同時にアクセスできるリソースの数を制御するための同期プリミティブです。

Semaphore.py

from threading import Semaphore, Thread
import time

def worker(semaphore, i):
    semaphore.acquire()
    print(f"Worker {i} acquired semaphore")
    time.sleep(1)
    print(f"Worker {i} releasing semaphore")
    semaphore.release()

semaphore = Semaphore(2)
threads = [Thread(target=worker, args=(semaphore, i)) for i in range(4)]

for t in threads:
    t.start()

for t in threads:
    t.join()
# Output:
# Worker 0 acquired semaphore
# Worker 1 acquired semaphore
# Worker 0 releasing semaphore
# Worker 2 acquired semaphore
# Worker 1 releasing semaphore
# Worker 3 acquired semaphore
# Worker 2 releasing semaphore
# Worker 3 releasing semaphore

6. デーモンスレッド（Daemon Thread）
デーモンスレッドは、プログラムが終了するときに強制的に終了されるスレッドです。

Daemon Thread.py

import threading
import time

def daemon():
    while True:
        print("Daemon thread is running...")
        time.sleep(1)

d = threading.Thread(target=daemon)
d.setDaemon(True)
d.start()

time.sleep(3)
print("Main thread exiting")
# Output:
# Daemon thread is running...
# Daemon thread is running...
# Daemon thread is running...
# Main thread exiting

7. ノンブロッキングI/O（Non-blocking I/O）
ノンブロッキングI/Oは、I/O操作が完了するまで待たずに他の作業を続行できるようにする方法です。selectors モジュールを使用します。

Non-blocking I/O.py

import selectors
import socket

sel = selectors.DefaultSelector()

def accept(sock, mask):
    conn, addr = sock.accept()
    print('Accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    data = conn.recv(1000)
    if data:
        print('Received', repr(data), 'from', conn)
    else:
        print('Closing', conn)
        sel.unregister(conn)
        conn.close()

sock = socket.socket()
sock.bind(('localhost', 12345))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)

while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)

8. ストリーム処理（Stream Processing）
ストリーム処理は、連続するデータの流れをリアルタイムで処理する方法です。Pythonでは itertools や pandas などのライブラリが使用されます。

Stream Processing.py

import itertools

# Infinite data stream
stream = itertools.cycle([1, 2, 3, 4, 5])
for i in range(10):
    print(next(stream))
# Output: 1 2 3 4 5 1 2 3 4 5

9. 継続（Continuation）
ストリーム処理は、連続するデータの流れをリアルタイムで処理する方法です。Pythonでは itertools や pandas などのライブラリが使用されます。

Continuation.py

def generator():
    yield "First part"
    yield "Second part"
    yield "Third part"

g = generator()
print(next(g))  # Output: First part
print(next(g))  # Output: Second part
print(next(g))  # Output: Third part

10. 依存性注入（Dependency Injection）
ストリーム処理は、連続するデータの流れをリアルタイムで処理する方法です。Pythonでは itertools や pandas などのライブラリが使用されます。

Dependency Injection.py

class Service:
    def do_something(self):
        print("Service is doing something")

class Client:
    def __init__(self, service):
        self.service = service

    def run(self):
        self.service.do_something()

service = Service()
client = Client(service)
client.run()
# Output: Service is doing something

最後まで読んでいただき、ありがとうございました。