本文目录导读：
我来详细解释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()
]);
}
}
关键要点
- 补偿事务设计: 每个步骤必须有对应的补偿操作
- 幂等性: 确保操作可以安全重试
- 日志记录: 详细记录每个步骤的执行情况
- 错误处理: 完善的异常处理和回滚机制
- 监控告警: 监控Saga执行状态并设置告警
这样实现的Saga模式能够保证分布式事务的一致性,同时提供了良好的错误处理和恢复机制。
