$v > 0)); if (empty($userIds)) { throw new BusinessException('分配失败:业务员列表为空'); } $inquiries = Db::table('inquiry_log') ->where('status', 0) ->where('recipient', 0) ->where('channel', $channelId) ->orderBy('id') ->limit($limit) ->get(); if ($inquiries->isEmpty()) { return 0; } $now = time(); $count = 0; foreach ($inquiries as $row) { $inquiryId = (int)$row->id; $recipientUserId = $this->nextRecipientId($userIds, $roundRobinKey); Db::transaction(function () use ($inquiryId, $recipientUserId, $now, &$count) { $affected = Db::table('inquiry_log') ->where('id', $inquiryId) ->update([ 'recipient' => $recipientUserId, 'status' => 2, 'r_time' => $now, 't_time' => $now, 'updated_at' => $now, ]); if ($affected === 0) { // 可能被其他进程抢先处理了,直接跳过 return; } $res = [ 'id' => $inquiryId, 'recipient' => $recipientUserId, 'dispatched_at' => $now, 'source' => 'fb_leads_inquiry', ]; // 队列推送 —— 用你原来的写法 RedisQueueRedis::send('inquiry-to-crm-new', $res); $count++; }); } return $count; } /** * 轮询获取下一个业务员ID */ protected function nextRecipientId(array $userIds, string $groupKey): int { $userIds = array_values(array_filter(array_map('intval', $userIds), fn($v) => $v > 0)); if (empty($userIds)) { throw new BusinessException('轮询分配失败:业务员列表为空'); } $count = count($userIds); $key = 'inquiry:assign:round_robin:' . $groupKey; $val = Redis::incr($key); // 原子自增 $index = ($val - 1) % $count; return $userIds[$index]; } }