在Laravel项目里，生产者消费者模式常用于处理耗时任务，比如批量数据导出、消息通知发送等场景，Eloquent Attribute TransferQueue可以基于Eloquent模型的属性变化来触发队列任务，简化数据传输流程。
环境准备与依赖安装
首先需要在Laravel项目中安装Eloquent Attribute TransferQueue相关依赖，打开终端执行以下命令：
composer require vendor/eloquent-attribute-transfer-queue
安装完成后，需要发布配置文件，执行命令：
php artisan vendor:publish --provider="EloquentAttributeTransferQueueTransferQueueServiceProvider"
配置文件中可以设置默认的队列连接、失败重试次数等参数，通常保持默认配置即可满足基础需求。
定义模型与传输规则
我们需要先创建一个Eloquent模型，比如Order订单模型，然后定义该模型需要传输的属性规则。首先在app/Models目录下创建Order.php：
<?php
namespace AppModels;
use IlluminateDatabaseEloquentModel;
use EloquentAttributeTransferQueueTraitsHasTransferQueue;
class Order extends Model
{
use HasTransferQueue;
// 定义需要传输的属性
protected $transferAttributes = ['id', 'order_no', 'total_amount', 'status'];
// 定义传输触发条件，当status变为2时触发传输
protected $transferConditions = [
'status' => 2
];
protected $fillable = ['order_no', 'total_amount', 'status'];
}
上述代码中，$transferAttributes指定了需要传递给消费者的属性，$transferConditions设置了触发传输的条件，当订单状态变为2时，会自动将对应属性推送到传输队列。
实现生产者逻辑
生产者负责生成需要处理的任务数据，这里我们在订单状态更新时自动触发数据传输。可以创建一个订单更新服务类：
<?php
namespace AppServices;
use AppModelsOrder;
class OrderService
{
public function updateOrderStatus(int $orderId, int $status)
{
$order = Order::findOrFail($orderId);
// 更新订单状态，满足传输条件时自动触发队列推送
$order->update(['status' => $status]);
return $order;
}
}
当调用updateOrderStatus方法将订单状态改为2时，Eloquent Attribute TransferQueue会自动将$transferAttributes中定义的属性封装成消息推送到队列中，完成生产者的数据投递工作。
实现消费者逻辑
消费者负责从队列中获取数据并处理业务逻辑，我们需要创建一个队列任务类。执行以下命令生成任务：
php artisan make:job ProcessOrderTransfer
然后编辑生成的app/Jobs/ProcessOrderTransfer.php文件：
<?php
namespace AppJobs;
use IlluminateBusQueueable;
use IlluminateContractsQueueShouldQueue;
use IlluminateFoundationBusDispatchable;
use IlluminateQueueInteractsWithQueue;
use IlluminateQueueSerializesModels;
use EloquentAttributeTransferQueueContractsTransferQueueMessage;
class ProcessOrderTransfer implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
protected $message;
public function __construct(TransferQueueMessage $message)
{
$this->message = $message;
}
public function handle()
{
// 获取传输过来的属性数据
$orderData = $this->message->getData();
// 这里编写消费者业务逻辑，比如发送订单通知、同步数据到其他系统等
// 示例：记录订单处理日志
Log::info('收到订单传输数据', $orderData);
// 如果处理失败，可以抛出异常触发重试
// throw new Exception('处理失败');
}
}
接下来需要在配置文件中指定该任务类作为消费者处理类，打开config/transfer-queue.php，修改consumer_job配置：
<?php
return [
'consumer_job' => AppJobsProcessOrderTransfer::class,
'queue_connection' => 'redis', // 使用redis作为队列驱动
'retry_times' => 3,
];
运行与测试
首先启动队列监听进程，执行命令：
php artisan queue:work
然后可以编写一个测试路由来验证功能，在routes/web.php中添加：
<?php
use AppServicesOrderService;
use IlluminateSupportFacadesRoute;
Route::get('/test-transfer', function () {
$service = new OrderService();
// 创建一个测试订单
$order = AppModelsOrder::create([
'order_no' => 'TEST_' . time(),
'total_amount' => 100.00,
'status' => 1
]);
// 更新状态触发传输
$service->updateOrderStatus($order->id, 2);
return '触发传输成功，请查看队列日志';
});
访问该路由后，生产者会推送订单数据到队列，消费者进程会自动获取数据处理，查看日志文件可以看到对应的订单数据输出。
异常场景处理
如果消费者处理失败，队列会根据配置的retry_times自动重试，超过重试次数后会把任务放入失败队列。可以通过以下命令查看失败任务：
php artisan queue:failed
如果需要重新执行失败任务，可以使用命令：
php artisan queue:retry 任务ID
另外需要注意，传输的属性数据如果需要包含关联关系，可以在模型中自定义getTransferData方法来扩展传输内容：
<?php
namespace AppModels;
use IlluminateDatabaseEloquentModel;
use EloquentAttributeTransferQueueTraitsHasTransferQueue;
class Order extends Model
{
use HasTransferQueue;
protected $transferAttributes = ['id', 'order_no', 'total_amount', 'status'];
protected $transferConditions = [
'status' => 2
];
protected $fillable = ['order_no', 'total_amount', 'status'];
// 自定义传输数据，添加关联的用户信息
public function getTransferData(): array
{
$baseData = $this->only($this->transferAttributes);
$baseData['user'] = $this->user ? $this->user->only(['id', 'name']) : null;
return $baseData;
}
public function user()
{
return $this->belongsTo(User::class);
}
}
PHPLaravelEloquent_Attribute_TransferQueue生产者消费者模式修改时间：2026-07-13 02:18:34