これらの手法はすべてデータのコピーを伴うため、共有メモリほど効率的ではありません。しかし、ロック管理などの複雑さがない分、使い勝手は良いでしょう。機能的には似ていますが、それぞれのシナリオにおけるパフォーマンスは異なります。
以下にテストコードを示します。ロジックは概ね同じです。今回のテストデータは、指定されたサイズの整数リストです。
ここでは2つのテストケースを用意しました。一つは「少量頻繁」(小さなデータを多く回送信)で、もう一つは「多量少数」(大きなデータを少数回送信)です。
(ZeroMQに関して補足します。ZeroMQはリストを直接送信できないため、シリアライズして送信する必要があります。このため、シリアライズのオーバーヘッドもテスト結果に含まれています。)
結果：
結果を表にまとめようとしましたが、差はそれほど大きくありませんでした。他のデータでもテストしましたが、結果はかなりばらつきました。(もちろん、この実験は厳密なものではありません)
Pipeのテストコード
from multiprocessing import Process, Pipe
import time
import random
# テストパラメータ
data_size = int(input("データサイズ: "))
num_iterations = int(input("テスト回数: "))
# テストデータの生成
payload = [[random.randint(0, 1000) for _ in range(data_size)] for _ in range(num_iterations)]
def consumer(conn):
"""データを受信するプロセス"""
start_time = time.time()
for _ in range(num_iterations):
conn.recv() # データを受信
print(f"処理時間: {time.time() - start_time}")
# Pipeの接続を確立
sender_conn, receiver_conn = Pipe()
# コンシューマプロセスを開始
Process(target=consumer, args=(receiver_conn,)).start()
# データを送信
for _ in range(num_iterations):
sender_conn.send(payload)
Queueのテストコード
from multiprocessing import Process, Queue
import time
import random
# テストパラメータ
data_size = int(input("データサイズ: "))
num_iterations = int(input("テスト回数: "))
# テストデータの生成
payload = [[random.randint(0, 1000) for _ in range(data_size)] for _ in range(num_iterations)]
def consumer(task_queue: Queue):
"""データを取得するプロセス"""
start_time = time.time()
for _ in range(num_iterations):
task_queue.get() # データを取得
print(f"処理時間: {time.time() - start_time}")
# タスクキューを初期化
task_queue = Queue()
# コンシューマプロセスを開始
Process(target=consumer, args=(task_queue,)).start()
# データをキューに投入
for _ in range(num_iterations):
task_queue.put(payload)
ZeroMQのテストコード
import zmq
import time
import pickle
import random
from multiprocessing import Process
# テストパラメータ
data_size = int(input("データサイズ: "))
num_iterations = int(input("テスト回数: "))
# テストデータの生成
payload = [[random.randint(0, 1000) for _ in range(data_size)] for _ in range(num_iterations)]
def consumer():
"""データを受信するプロセス"""
# ZeroMQコンテキストとソケットの初期化
ctx = zmq.Context()
sub_socket = ctx.socket(zmq.PULL)
sub_socket.connect('tcp://127.0.0.1:5557')
init_time = time.time()
print(f"初期化時間: {time.time() - init_time}")
start_time = time.time()
for _ in range(num_iterations):
# 受信したデータをデシリアライズ
pickle.loads(sub_socket.recv())
print(f"処理時間: {time.time() - start_time}")
# コンシューマプロセスを開始
Process(target=consumer).start()
# パブリッシャーの設定
ctx = zmq.Context()
pub_socket = ctx.socket(zmq.PUSH)
pub_socket.bind("tcp://*:5557")
# データを送信 (シリアライズが必要)
for _ in range(num_iterations):
pub_socket.send(pickle.dumps(payload))