microservices-patterns

Name: microservices-patterns
Author: secondsky

github.com/secondsky開発240392 回インストール5 週間前に更新

リポジトリ

microservices-patterns について

マイクロサービスや分散システムを設計する開発者向けのスキルです。サービス分割、通信、データ管理、回復性のパターンを例や注意点とともに扱います。

主な機能

01サービス分割の戦略
02同期・非同期通信パターン
03サービスごとのデータとSagaトランザクション
04サーキットブレーカー、リトライ、バルクヘッド
05ベストプラクティスとよくある問題

活用シーン

01モノリスのサービス分割
02サービス境界と契約の定義
03通信・データパターンの選択
04サービス障害への対応計画

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/secondsky/claude-skills --skill microservices-patterns

インストール後、Claude Code で/microservices-patternsを入力して読み込まれたか確認します。

詳細

マイクロサービスのパターン

分散システムを構築するための、サービス境界、サービス間通信、データ管理、レジリエンスのパターンなど、マイクロサービスアーキテクチャのパターンを習得します。

このスキルを使用する場面

モノリスをマイクロサービスに分解する
サービス境界とコントラクトを設計する
サービス間通信を実装する
分散データとトランザクションを管理する
レジリエントな分散システムを構築する
サービスディスカバリーとロードバランシングを実装する
イベント駆動型アーキテクチャを設計する

コアコンセプト

1. サービス分解の戦略

ビジネスケイパビリティ別

ビジネス機能を中心にサービスを編成する
各サービスが独自のドメインを所有する
例：OrderService、PaymentService、InventoryService

サブドメイン別（DDD）

コアドメインと支援サブドメイン
境界づけられたコンテキストをサービスに対応づける
明確な所有権と責任

ストラングラーフィグパターン

モノリスから段階的に切り出す
新しい機能はマイクロサービスとして実装する
プロキシが旧システムまたは新システムにルーティングする

2. 通信パターン

同期（リクエスト／レスポンス）

REST API
gRPC
GraphQL

非同期（イベント／メッセージ）

イベントストリーミング（Kafka）
メッセージキュー（RabbitMQ、SQS）
Pub/Subパターン

3. データ管理

サービスごとのデータベース

各サービスが独自のデータを所有する
データベースを共有しない
疎結合

Sagaパターン

分散トランザクション
補償アクション
結果整合性

4. レジリエンスパターン

サーキットブレーカー

エラーが繰り返し発生した場合、速やかに失敗させる
障害の連鎖を防止する

バックオフ付き再試行

一時的な障害への対処
指数バックオフ

バルクヘッド

リソースを分離する
障害の影響を制限する

サービス分解パターン

パターン1：ビジネスケイパビリティ別

# Order Service
class OrderService:
    async def create_order(self, order_data: dict) -> Order:
        order = Order.create(order_data)
        await self.event_bus.publish(
            OrderCreatedEvent(order_id=order.id, customer_id=order.customer_id)
        )
        return order

# Payment Service (separate service)
class PaymentService:
    async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult:
        result = await self.payment_gateway.charge(
            amount=payment_request.amount,
            customer=payment_request.customer_id
        )
        if result.success:
            await self.event_bus.publish(
                PaymentCompletedEvent(order_id=payment_request.order_id)
            )
        return result

# Inventory Service (separate service)
class InventoryService:
    async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult:
        for item in items:
            available = await self.inventory_repo.get_available(item.product_id)
            if available < item.quantity:
                return ReservationResult(success=False, error=f"Insufficient inventory")

        reservation = await self.create_reservation(order_id, items)
        await self.event_bus.publish(InventoryReservedEvent(order_id=order_id))
        return ReservationResult(success=True, reservation=reservation)

パターン2：APIゲートウェイ

from fastapi import FastAPI
import httpx

class APIGateway:
    """Central entry point for all client requests."""

    def __init__(self):
        self.order_service_url = "http://order-service:8000"
        self.payment_service_url = "http://payment-service:8001"
        self.http_client = httpx.AsyncClient(timeout=5.0)

    @circuit(failure_threshold=5, recovery_timeout=30)
    async def call_order_service(self, path: str, method: str = "GET", **kwargs):
        """Call order service with circuit breaker."""
        response = await self.http_client.request(
            method, f"{self.order_service_url}{path}", **kwargs
        )
        response.raise_for_status()
        return response.json()

    async def create_order_aggregate(self, order_id: str) -> dict:
        """Aggregate data from multiple services."""
        order, payment, inventory = await asyncio.gather(
            self.call_order_service(f"/orders/{order_id}"),
            self.call_payment_service(f"/payments/order/{order_id}"),
            self.call_inventory_service(f"/reservations/order/{order_id}"),
            return_exceptions=True
        )

        result = {"order": order}
        if not isinstance(payment, Exception):
            result["payment"] = payment
        if not isinstance(inventory, Exception):
            result["inventory"] = inventory
        return result

通信パターン

パターン1：同期REST通信

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ServiceClient:
    """HTTP client with retries and timeout."""

    def __init__(self, base_url: str):
        self.base_url = base_url
        self.client = httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=2.0))

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    async def get(self, path: str, **kwargs):
        """GET with automatic retries."""
        response = await self.client.get(f"{self.base_url}{path}", **kwargs)
        response.raise_for_status()
        return response.json()

payment_client = ServiceClient("http://payment-service:8001")
result = await payment_client.get("/payments/123")

パターン2：非同期イベント駆動型

from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
import json

class EventBus:
    """Event publishing and subscription."""

    async def publish(self, event: DomainEvent):
        """Publish event to Kafka topic."""
        await self.producer.send_and_wait(
            event.event_type,
            value=asdict(event),
            key=event.aggregate_id.encode()
        )

    async def subscribe(self, topic: str, handler: callable):
        """Subscribe to events."""
        consumer = AIOKafkaConsumer(topic, bootstrap_servers=self.bootstrap_servers)
        await consumer.start()
        async for message in consumer:
            await handler(message.value)

# Order Service publishes
await event_bus.publish(OrderCreatedEvent(order_id=order.id))

# Inventory Service subscribes
async def handle_order_created(event_data: dict):
    await reserve_inventory(event_data["order_id"], event_data["items"])

パターン3：Sagaパターン（分散トランザクション）

class OrderFulfillmentSaga:
    """Orchestrated saga for order fulfillment."""

    def __init__(self):
        self.steps = [
            SagaStep("create_order", self.create_order, self.cancel_order),
            SagaStep("reserve_inventory", self.reserve_inventory, self.release_inventory),
            SagaStep("process_payment", self.process_payment, self.refund_payment),
            SagaStep("confirm_order", self.confirm_order, self.cancel_order_confirmation)
        ]

    async def execute(self, order_data: dict) -> SagaResult:
        completed_steps = []
        context = {"order_data": order_data}

        try:
            for step in self.steps:
                result = await step.action(context)
                if not result.success:
                    await self.compensate(completed_steps, context)
                    return SagaResult(status=SagaStatus.FAILED, error=result.error)

                completed_steps.append(step)
                context.update(result.data)

            return SagaResult(status=SagaStatus.COMPLETED, data=context)
        except Exception as e:
            await self.compensate(completed_steps, context)
            return SagaResult(status=SagaStatus.FAILED, error=str(e))

    async def compensate(self, completed_steps: List[SagaStep], context: dict):
        """Execute compensating actions in reverse order."""
        for step in reversed(completed_steps):
            await step.compensation(context)

レジリエンスパターン

サーキットブレーカーパターン

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"   # Normal operation
    OPEN = "open"       # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing recovery

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.opened_at = None

    async def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
            else:
                raise CircuitBreakerOpenError("Circuit breaker is open")

        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            self.opened_at = datetime.now()

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30)
result = await breaker.call(payment_client.process_payment, payment_data)

ベストプラクティス

サービス境界：ビジネスケイパビリティに合わせる
サービスごとのデータベース：データベースを共有しない
APIコントラクト：バージョン管理し、後方互換性を保つ
可能な限り非同期にする：直接呼び出しよりイベントを優先する
サーキットブレーカー：サービス障害時に速やかに失敗させる
分散トレーシング：サービス間のリクエストを追跡する
サービスレジストリ：動的なサービスディスカバリー
ヘルスチェック：生存プローブと準備完了プローブ

よくある落とし穴

分散モノリス：サービスが密結合している
おしゃべりなサービス：サービス間呼び出しが多すぎる
共有データベース：データを介した密結合
サーキットブレーカーがない：障害が連鎖する
すべて同期処理：密結合になり、レジリエンスが低下する
時期尚早なマイクロサービス化：最初からマイクロサービスで始める
ネットワーク障害を無視する：ネットワークが常に信頼できると想定する
補償ロジックがない：失敗したトランザクションを取り消せない

Markdown 対応

ログインしてコメント

基本情報

作者: secondsky
リポジトリ: secondsky/claude-skills
カテゴリ: 開発
インストール数: 392
GitHub スター: 240

問題を報告