AMQP协议本身无批量原语，php-amqplib和Enqueue的publish()或send()均为单条语义，批量发送需业务层控制聚合节奏并配合传输层实现，否则仍是线性网络往返。
php-amqplib 和 enqueue 都支持批量发送，但默认不开启；真正在生产环境做消息聚合，得靠业务层控制节奏 + 传输层配合，不能只依赖框架自动“合并”。
为什么直接调用 publish() 不等于批量发送
AMQP 协议本身没有“批量 publish”原语。像 php-amqplib 的 AMQPExchange::publish() 每次调用都是一次独立的 AMQP 方法帧（method frame），哪怕你循环发 100 条，也是 100 次网络往返（未启用 channel 批量确认时）。
常见错误现象：publish() 调用耗时随消息数线性增长，CPU 和 RabbitMQ 连接数飙升，消费者端看到消息是“散装抵达”，根本没聚合效果。
- AMQP 是面向单条消息设计的，批量语义必须由上层封装
-
enqueue的ProducerInterface::send()同样是单条语义，即使传入数组也不会自动拆包或合并 - Redis 驱动（如
redis-lpush）看似能LPUSH key v1 v2 v3，但 PHP 消费者通常仍按单条BRPOP取出，聚合逻辑仍在业务代码里
think-queue 中实现任务聚合的实操路径
ThinkPHP 的 think-queue 本身不提供“攒批投递”功能，但你可以通过以下方式在生产者侧主动聚合：
立即学习“PHP免费学习笔记（深入）”；
- 用
cache或static数组暂存待发任务，设置触发阈值（如 50 条 or 2 秒超时） - 聚合后调用
Queue::push()一次，把整个数组序列化为单个任务体 - 消费者收到后，先
json_decode再遍历处理，而不是当成单条业务逻辑 - 注意：失败重试会重试整批，需在消费逻辑里加细粒度错误隔离（比如记录失败子项 ID，不中断整批）
示例片段：
// 生产者端攒批
$batch = [];
for ($i = 0; $i < 100; $i++) {
$batch[] = ['user_id' => $i, 'action' => 'notify'];
}
Queue::push('App\Job\BatchNotifyJob', ['batch' => $batch]);
RabbitMQ 死信 + TTL 实现延迟聚合分发
如果你需要“等齐一批再统一触发”，又不想长期 hold 内存，可以用 RabbitMQ 的 TTL + 死信队列机制模拟定时聚合：
- 所有单条消息发往一个带
x-message-ttl=5000的临时队列 A - 队列 A 声明
x-dead-letter-exchange指向主交换器 - 5 秒内到达的消息会一起“过期”，被批量路由到下游队列 B
- 消费者监听 B，拿到的就是自然聚合成批的消息（前提是 RabbitMQ 版本 ≥ 3.8，且启用
dead-letter-routing-key精确控制）
⚠️ 注意：这不是严格意义上的“批量发送”，而是利用消息生命周期制造聚合窗口；TTL 精度受 RabbitMQ 调度器影响，实际延迟可能浮动 ±200ms。
Redis Stream 驱动下更自然的批量读写
相比 List 模型，Redis Stream 天然支持批量操作，更适合聚合场景：
- 生产者用
XADD stream_key * field1 value1 field2 value2单条写入，但可配合 pipeline 批量提交 - 消费者用
XREAD COUNT 100 STREAMS stream_key >一次拉取最多 100 条 -
think-queue目前不原生支持 Stream，但你可以绕过队列组件，直接用Redis::xadd()/Redis::xread()封装自己的聚合消费者 - Stream 还支持 consumer group、pending list、ACK 机制，比 List 更可靠，适合中高并发聚合任务
真正容易被忽略的一点：聚合不是越大批量越好。单次处理 1000 条可能卡住 worker 进程超过 30 秒，导致心跳超时、任务重复投递。建议从 20–50 条起步压测，结合你的任务平均耗时和超时配置来定界。
