インストール手順

インストール前に以下の点に注意してください：

• ErlangとRabbitMQのバージョン互換性を確認すること（RabbitMQ公式ドキュメント参照）
• インストールパスにスペースや日本語文字が含まれないようにすること
• 必要なソフトウェアは公式サイトからダウンロード可能

Erlang環境が正しく設定されているか確認するには、コマンドプロンプトでerlコマンドを実行し、正常に起動することを確認します。

基本的な利用方法

RabbitMQはメッセージブローカーとして機能し、複数の通信パターンをサポートしています。以下に主要なパターンを紹介します。

Hello World パターン

最も基本的なメッセージ送受信パターンです。

メッセージ送信側コード例：

import pika

# 接続設定
connection_params = pika.ConnectionParameters(host='localhost')
connection = pika.BlockingConnection(connection_params)
channel = connection.channel()

# キュー作成
channel.queue_declare(queue='basic_queue')

# メッセージ送信
channel.basic_publish(
    exchange='',
    routing_key='basic_queue',
    body=b'Hello RabbitMQ!'
)

connection.close()

メッセージ受信側コード例：

import pika

def process_message(channel, method, properties, body):
    print(f'Message received: {body}')
    channel.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='basic_queue')

channel.basic_consume(queue='basic_queue', on_message_callback=process_message)
channel.start_consuming()

ワークキュー パターン

複数のワーカー間でタスクを分配するパターンです。

タスク生成コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 永続化キュー作成
channel.queue_declare(queue='task_queue', durable=True)

# 永続化メッセージ送信
channel.basic_publish(
    exchange='',
    routing_key='task_queue',
    body=b'Processing task',
    properties=pika.BasicProperties(delivery_mode=2)
)

connection.close()

ワーカーコード：

import time
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)

def handle_task(channel, method, properties, body):
    print(f'Processing: {body}')
    time.sleep(5)  # 処理時間シミュレーション
    channel.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=handle_task)
channel.start_consuming()

パブリッシュ・サブスクライブ パターン

ファンアウト交換を使用して、すべてのサブスクライバーに同じメッセージを配布します。

メッセージ発行コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# ファンアウト交換作成
channel.exchange_declare(exchange='broadcast_exchange', exchange_type='fanout')

channel.basic_publish(
    exchange='broadcast_exchange',
    routing_key='',
    body=b'Broadcast message'
)

connection.close()

サブスクリプションコード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='broadcast_exchange', exchange_type='fanout')
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='broadcast_exchange', queue=queue_name)

def receive_broadcast(channel, method, properties, body):
    print(f'Broadcast received: {body}')

channel.basic_consume(queue=queue_name, on_message_callback=receive_broadcast)
channel.start_consuming()

ルーティング パターン

ダイレクト交換を使用して、特定のルーティングキーに基づいてメッセージを配布します。

ルーティング発行コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

severity = 'warning'
message_body = b'Warning message'

channel.basic_publish(
    exchange='direct_logs',
    routing_key=severity,
    body=message_body
)

connection.close()

ルーティング受信コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
queue_result = channel.queue_declare(queue='', exclusive=True)
queue_name = queue_result.method.queue

binding_keys = ['warning', 'error']
for binding_key in binding_keys:
    channel.queue_bind(
        exchange='direct_logs',
        queue=queue_name,
        routing_key=binding_key
    )

def handle_routing_message(channel, method, properties, body):
    print(f'Routing message: {body}')

channel.basic_consume(queue=queue_name, on_message_callback=handle_routing_message)
channel.start_consuming()

トピック パターン

ワイルドカードを使用した柔軟なルーティングが可能なパターンです。

トピック発行コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

routing_pattern = 'user.activity.login'
message_data = b'User login event'

channel.basic_publish(
    exchange='topic_logs',
    routing_key=routing_pattern,
    body=message_data
)

connection.close()

トピック受信コード：

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
temp_queue = channel.queue_declare(queue='', exclusive=True)
temp_queue_name = temp_queue.method.queue

patterns = ['user.*', '*.activity.*']
for pattern in patterns:
    channel.queue_bind(
        exchange='topic_logs',
        queue=temp_queue_name,
        routing_key=pattern
    )

def handle_topic_message(channel, method, properties, body):
    print(f'Topic message: {body}')

channel.basic_consume(queue=temp_queue_name, on_message_callback=handle_topic_message)
channel.start_consuming()

RPC パターン

リモートプロシージャコールをメッセージキューで実装するパターンです。

RPCサーバーコード：

import pika
import uuid

class RpcServer:
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='rpc_queue', durable=True)
        
    def calculate_response(self, request):
        # 処理ロジック
        return int(request) * 2
        
    def process_request(self, channel, method, props, body):
        request_data = body.decode('utf-8')
        response = self.calculate_response(request_data)
        
        channel.basic_publish(
            exchange='',
            routing_key=props.reply_to,
            properties=pika.BasicProperties(correlation_id=props.correlation_id),
            body=str(response).encode('utf-8')
        )
        
        channel.basic_ack(delivery_tag=method.delivery_tag)

    def start_server(self):
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(queue='rpc_queue', on_message_callback=self.process_request)
        self.channel.start_consuming()

server = RpcServer()
server.start_server()