LaravelSaga模式怎么实现

本文目录导读：

1. Saga模式基础概念
2. 基础架构设计
3. 订单处理Saga示例
4. 使用Orchestrator管理Saga
5. 数据库表设计
6. 高级特性实现
7. 使用示例
8. 异步Saga实现
9. 关键要点

我来详细解释Laravel中的Saga模式实现方法。

Saga模式基础概念

Saga模式用于管理分布式事务,通过一系列本地事务和补偿操作来保证数据一致性。

基础架构设计

<?php
namespace App\Sagas;
abstract class Saga
{
    protected array $steps = [];
    protected array $executedSteps = [];
    protected array $compensations = [];
    protected ?\Throwable $lastException = null;
    abstract public function execute(): bool;
    protected function addStep(string $name, callable $action, callable $compensation): void
    {
        $this->steps[$name] = [
            'action' => $action,
            'compensation' => $compensation,
        ];
    }
    protected function runStep(string $name): bool
    {
        try {
            $result = call_user_func($this->steps[$name]['action']);
            $this->executedSteps[] = $name;
            $this->compensations[$name] = $this->steps[$name]['compensation'];
            return true;
        } catch (\Throwable $e) {
            $this->lastException = $e;
            $this->rollback();
            return false;
        }
    }
    protected function rollback(): void
    {
        $failedSteps = array_reverse($this->executedSteps);
        foreach ($failedSteps as $step) {
            try {
                if (isset($this->compensations[$step])) {
                    call_user_func($this->compensations[$step]);
                }
            } catch (\Throwable $e) {
                // 记录补偿失败日志
                logger()->error("Saga compensation failed for step: {$step}", [
                    'error' => $e->getMessage()
                ]);
            }
        }
    }
    public function getLastException(): ?\Throwable
    {
        return $this->lastException;
    }
}

订单处理Saga示例

<?php
namespace App\Sagas;
use App\Models\Order;
use App\Models\Payment;
use App\Models\Inventory;
use App\Services\ShippingService;
use App\Services\NotificationService;
use Illuminate\Support\Facades\DB;
class OrderSaga extends Saga
{
    private Order $order;
    private array $data;
    public function __construct(array $data)
    {
        $this->data = $data;
        $this->initializeSteps();
    }
    protected function initializeSteps(): void
    {
        // 步骤1: 创建订单
        $this->addStep(
            'create_order',
            function () {
                $this->order = Order::create([
                    'user_id' => $this->data['user_id'],
                    'items' => $this->data['items'],
                    'total' => $this->data['total'],
                    'status' => 'pending'
                ]);
                return $this->order;
            },
            function () {
                $this->order->delete();
            }
        );
        // 步骤2: 扣减库存
        $this->addStep(
            'reserve_inventory',
            function () {
                foreach ($this->data['items'] as $item) {
                    $inventory = Inventory::findOrFail($item['product_id']);
                    if ($inventory->quantity < $item['quantity']) {
                        throw new \Exception("Insufficient inventory for product: {$item['product_id']}");
                    }
                    $inventory->decrement('quantity', $item['quantity']);
                }
            },
            function () {
                foreach ($this->data['items'] as $item) {
                    Inventory::where('id', $item['product_id'])
                        ->increment('quantity', $item['quantity']);
                }
            }
        );
        // 步骤3: 处理支付
        $this->addStep(
            'process_payment',
            function () {
                $payment = Payment::create([
                    'order_id' => $this->order->id,
                    'amount' => $this->data['total'],
                    'status' => 'pending'
                ]);
                // 调用支付网关
                $result = $this->callPaymentGateway($payment);
                if (!$result) {
                    throw new \Exception("Payment failed");
                }
                $payment->update(['status' => 'completed']);
                return $payment;
            },
            function () {
                Payment::where('order_id', $this->order->id)
                    ->update(['status' => 'refunded']);
                // 调用支付网关退款
                $this->refundPayment($this->order->id);
            }
        );
        // 步骤4: 安排发货
        $this->addStep(
            'schedule_shipping',
            function () {
                app(ShippingService::class)->scheduleDelivery([
                    'order_id' => $this->order->id,
                    'address' => $this->data['shipping_address'],
                    'items' => $this->data['items']
                ]);
            },
            function () {
                app(ShippingService::class)->cancelDelivery($this->order->id);
            }
        );
        // 步骤5: 发送通知
        $this->addStep(
            'send_notification',
            function () {
                app(NotificationService::class)->sendOrderConfirmation($this->order);
            },
            function () {
                // 通知补偿：发送取消通知
                app(NotificationService::class)->sendOrderCancellation($this->order);
            }
        );
    }
    public function execute(): bool
    {
        $steps = ['create_order', 'reserve_inventory', 'process_payment', 
                   'schedule_shipping', 'send_notification'];
        foreach ($steps as $step) {
            if (!$this->runStep($step)) {
                $this->order->update(['status' => 'failed']);
                return false;
            }
        }
        $this->order->update(['status' => 'completed']);
        return true;
    }
    private function callPaymentGateway($payment): bool
    {
        // 模拟支付网关调用
        return true;
    }
    private function refundPayment(int $orderId): void
    {
        // 实现退款逻辑
    }
}

使用Orchestrator管理Saga

<?php
namespace App\Services;
use App\Sagas\OrderSaga;
use App\Models\SagaLog;
use Illuminate\Support\Facades\DB;
class SagaOrchestrator
{
    public function executeOrderSaga(array $data): array
    {
        $sagaLog = SagaLog::create([
            'type' => 'order',
            'status' => 'started',
            'payload' => $data
        ]);
        try {
            DB::beginTransaction();
            $saga = new OrderSaga($data);
            $result = $saga->execute();
            if ($result) {
                $sagaLog->update(['status' => 'completed']);
                DB::commit();
                return ['success' => true, 'message' => 'Order processed successfully'];
            } else {
                $sagaLog->update([
                    'status' => 'failed',
                    'error' => $saga->getLastException()?->getMessage()
                ]);
                DB::rollBack();
                return ['success' => false, 'message' => 'Saga execution failed'];
            }
        } catch (\Throwable $e) {
            DB::rollBack();
            $sagaLog->update([
                'status' => 'failed',
                'error' => $e->getMessage()
            ]);
            throw $e;
        }
    }
}

数据库表设计

// 创建Saga日志表
Schema::create('saga_logs', function (Blueprint $table) {
    $table->id();
    $table->string('type');
    $table->string('status');
    $table->json('payload')->nullable();
    $table->json('executed_steps')->nullable();
    $table->text('error')->nullable();
    $table->timestamps();
});
// 创建Saga步骤日志表
Schema::create('saga_step_logs', function (Blueprint $table) {
    $table->id();
    $table->foreignId('saga_log_id')->constrained();
    $table->string('step_name');
    $table->string('status');
    $table->json('input')->nullable();
    $table->json('output')->nullable();
    $table->text('error')->nullable();
    $table->timestamps();
});

高级特性实现

<?php
namespace App\Sagas\Traits;
trait RetryableStep
{
    private int $maxRetries = 3;
    private int $retryDelay = 1;
    protected function runWithRetry(callable $action, string $stepName): bool
    {
        $attempts = 0;
        while ($attempts < $this->maxRetries) {
            try {
                $attempts++;
                $result = call_user_func($action);
                $this->logStepAttempt($stepName, $attempts, true);
                return true;
            } catch (\Throwable $e) {
                $this->logStepAttempt($stepName, $attempts, false, $e);
                if ($attempts < $this->maxRetries) {
                    sleep($this->retryDelay);
                    $this->retryDelay *= 2; // 指数退避
                }
            }
        }
        return false;
    }
    private function logStepAttempt(string $step, int $attempt, bool $success, ?\Throwable $e = null): void
    {
        // 记录重试日志
        logger()->info("Saga step attempt", [
            'step' => $step,
            'attempt' => $attempt,
            'success' => $success,
            'error' => $e?->getMessage()
        ]);
    }
}

使用示例

<?php
namespace App\Http\Controllers;
use App\Services\SagaOrchestrator;
use Illuminate\Http\Request;
class OrderController extends Controller
{
    private SagaOrchestrator $orchestrator;
    public function __construct(SagaOrchestrator $orchestrator)
    {
        $this->orchestrator = $orchestrator;
    }
    public function placeOrder(Request $request)
    {
        $data = $request->validate([
            'user_id' => 'required|exists:users,id',
            'items' => 'required|array',
            'total' => 'required|numeric',
            'shipping_address' => 'required|string'
        ]);
        try {
            $result = $this->orchestrator->executeOrderSaga($data);
            if ($result['success']) {
                return response()->json([
                    'message' => 'Order placed successfully'
                ], 201);
            }
            return response()->json([
                'error' => $result['message'],
                'retry' => true
            ], 500);
        } catch (\Throwable $e) {
            return response()->json([
                'error' => 'Order processing failed',
                'details' => $e->getMessage()
            ], 500);
        }
    }
}

异步Saga实现

<?php
namespace App\Jobs;
use App\Sagas\OrderSaga;
use App\Models\SagaLog;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
class ProcessOrderSagaJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    private array $data;
    private SagaLog $sagaLog;
    public function __construct(array $data)
    {
        $this->data = $data;
    }
    public function handle()
    {
        $saga = new OrderSaga($this->data);
        $result = $saga->execute();
        if ($result) {
            // Saga成功，可以触发后续操作
            dispatch(new PostOrderProcessingJob($saga->getOrder()));
        }
    }
    public function failed(\Throwable $exception)
    {
        // 记录失败并可能发送警报
        logger()->error('Order saga failed', [
            'data' => $this->data,
            'error' => $exception->getMessage()
        ]);
    }
}

关键要点

1. 补偿事务设计: 每个步骤必须有对应的补偿操作
2. 幂等性: 确保操作可以安全重试
3. 日志记录: 详细记录每个步骤的执行情况
4. 错误处理: 完善的异常处理和回滚机制
5. 监控告警: 监控Saga执行状态并设置告警

这样实现的Saga模式能够保证分布式事务的一致性,同时提供了良好的错误处理和恢复机制。