תור ייבוא מוצרים ברקע עם Laravel + Redis

תור ייבוא מוצרים ברקע (עיבוד אסינכרוני)

תארו לעצמכם העלאת מחירון עם 50,000 מוצרים. ייבוא סינכרוני נתקע למשך 2 דקות ונכשל עם שגיאת 500. התוצאה: אובדן נתונים ותסכול למשתמש. אנו מתכננים תורי משימות שאינם חוסמים את הממשק ואינם מאבדים נתונים. הקובץ מתקבל, המשימה נכנסת לתור, ומזהה ייבוא מוחזר מיד. המשתמש רואה התקדמות דרך WebSocket ואינו ממתין.

ייבוא אסינכרוני על Laravel + Redis פותר זאת: הוא מעבד את הקובץ ללא חסימה, ואתם רואים התקדמות בזמן אמת. חיסכון בזמן—עד 70% בהשוואה לפתרונות סינכרוניים.

סקירת ארכיטקטורה

HTTP Upload (файл/URL) ↓
Import Job (запись в очередь) ↓
Queue (Redis / SQS / RabbitMQ) ↓
Worker Process (отдельный процесс/контейнер) ↓
Chunk Processing (батчи по 500 товаров) ↓
Database (upsert) ↓
Progress Event (WebSocket / SSE → UI)

כיצד להשיג סובלנות לתקלות בתור?

אנו משתמשים בLaravel Queues עם לוגיקת ניסיון חוזר: במקרה של כשל, העובד מנסה שוב את הנתח עד 3 פעמים עם השהיה. אם כל הניסיונות נכשלים, המשימה מסומנת כנכשלה, ואנו מודיעים למנהל. נתונים לעולם לא אובדים—מאושר על ידי התיעוד הרשמי.

// Контроллер — принимает файл и ставит задачу
class ProductImportController extends Controller
{
    public function upload(Request $request): JsonResponse
    {
        $path = $request->file('file')->store('imports');
        $import = ImportJob::create([
            'file_path' => $path,
            'status' => 'pending',
            'total' => 0,
            'processed' => 0,
            'errors' => 0,
        ]);
        ProcessProductImport::dispatch($import->id);
        return response()->json(['import_id' => $import->id]);
    }
}

// Job — обработка в фоне
class ProcessProductImport implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue;

    public int $timeout = 3600; // 1 час
    public int $tries = 3;

    public function handle(): void
    {
        $import = ImportJob::findOrFail($this->importId);
        $import->update(['status' => 'processing', 'started_at' => now()]);

        $reader = new CsvReader(storage_path('app/' . $import->file_path));
        $total = $reader->count();
        $import->update(['total' => $total]);

        foreach ($reader->chunk(500) as $chunkIndex => $rows) {
            try {
                DB::transaction(function () use ($rows) {
                    foreach ($rows as $row) {
                        Product::updateOrCreate(
                            ['sku' => $row['sku']],
                            $this->mapRow($row)
                        );
                    }
                });

                $processed = ($chunkIndex + 1) * 500;
                $import->update(['processed' => min($processed, $total)]);

                // Событие прогресса
                event(new ImportProgressUpdated($import->id, min($processed, $total), $total));
            } catch (\Exception $e) {
                $import->increment('errors');
                Log::error("Import chunk failed", ['chunk' => $chunkIndex, 'error' => $e->getMessage()]);
            }
        }

        $import->update(['status' => 'completed', 'finished_at' => now()]);
    }
}

טיפול בשגיאות ייבוא

שגיאות שורה אינן עוצרות את התהליך. כל שגיאה מתועדת בטבלת HTTP Upload (файл/URL) ↓ Import Job (запись в очередь) ↓ Queue (Redis / SQS / RabbitMQ) ↓ Worker Process (отдельный процесс/контейнер) ↓ Chunk Processing (батчи по 500 товаров) ↓ Database (upsert) ↓ Progress Event (WebSocket / SSE → UI) :

CREATE TABLE import_errors (
    id BIGSERIAL PRIMARY KEY,
    import_id BIGINT,
    row_number INT,
    row_data JSONB,
    error_msg TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

לאחר השלמה, המשתמש מוריד דוח עם שורות שגויות. אנו גם מגדירים התראות כאשר סף השגיאות נחצה (לדוגמה, >5% מכלל השורות).

WebSocket / SSE להתקדמות

// Laravel Broadcasting: событие прогресса
class ImportProgressUpdated implements ShouldBroadcast
{
    public function broadcastOn(): Channel
    {
        return new PrivateChannel("import.{$this->importId}");
    }

    public function broadcastWith(): array
    {
        return [
            'processed' => $this->processed,
            'total' => $this->total,
            'percent' => round($this->processed / $this->total * 100),
        ];
    }
}

בצד הלקוח—הירשמו דרך Laravel Echo או EventSource מקורי (SSE). אנו מיישמים את שניהם לפי בחירתכם.

השוואה: ייבוא סינכרוני מול אסינכרוני

ייבוא אסינכרוני מהיר פי 3–5 מייבוא סינכרוני בזכות עיבוד מקבילי וללא פסקי זמן.

השוואת מערכות תור

עבור 90% מהפרויקטים, Redis מספיק—מהיר, פשוט, ומובנה ב-Laravel. עבור נתונים קריטיים או נפחים > 1 מיליון רשומות, בחרו ב-SQS או RabbitMQ. אם אינכם בטוחים בבחירה, קבלו ייעוץ—נעזור לבחור את הדרייבר האופטימלי.

גודל נתח אופטימלי

גודל הנתח הוא פרמטר ביצועים מרכזי. קטן מדי (50 שורות) יוצר תקורה גבוהה של עסקאות, גדול מדי (5000) מסכן את גבולות הזיכרון. אנו מתאימים את גודל הנתח לשרת שלכם: בדרך כלל 500–1000 שורות לעובד. להאצה, השתמשו בעובדים מקביליים עם Laravel Batches.

ניטור תור עם Laravel Horizon

Laravel Horizon מספק לוח מחוונים יפה לניטור התור: מספר משימות, זמן ביצוע, מספר שגיאות. אנו מגדירים התראות בטלגרם או Slack בעת חציית סף. זה מאפשר תגובה מהירה לתקלות ושומר על שליטה.

עיבוד מקבילי (Laravel Batches)

עבור קבצים גדולים מאוד (100,000+ מוצרים), אנו מפצלים לחלקים עצמאיים עם עובדים מקביליים:

class DispatchImportChunks implements ShouldQueue {
    public function handle(): void {
        $chunks = $this->splitFile($this->filePath, chunkSize: 1000);
        Bus::batch(
            array_map(fn($chunk) => new ProcessImportChunk($chunk), $chunks)
        )
            ->then(fn(Batch $batch) => $this->onComplete($batch))
            ->catch(fn(Batch $batch, Throwable $e) => $this->onError($batch, $e))
            ->dispatch();
    }
}

זה מאיץ את הייבוא פי 3–5 בהשוואה לעיבוד רציף. תוצאה אופיינית: זמן השבתת שרת מופחת ב-40%.

מה כלול

• עיצוב ארכיטקטורת תור (Redis / SQS / RabbitMQ)
• הטמעת העלאת קבצים (CSV, Excel, XML, CommerceML)
• הגדרת עובדים עם עיבוד נתחים (גודל נתח מותאם לשרת שלכם)
• מערכת ניסיון חוזר וטיפול בשגיאות (תיעוד + התראות)
• התקדמות בזמן אמת (WebSocket או SSE)
• תיעוד על הפעלת עובדים וניטור
• אינטגרציה עם הממשק שלכם (API להתחלה וקבלת סטטוס)

לוח זמנים

הטמעה בסיסית (פורמט קובץ אחד, עובד אחד)—4–6 ימי עבודה. עם עיבוד מקבילי ומספר פורמטים—8–10 ימי עבודה. נאמוד את הפרויקט שלכם לאחר בחינת הקובץ והדרישות שלכם.

הניסיון שלנו

לצוות שלנו ניסיון רב במסחר אלקטרוני ולוגיסטיקה עם תורי רקע, לאחר שהשלמנו 30+ פרויקטים. אנו משתמשים בתבניות מוכחות: Repository, BFF, Event Sourcing. אנו מבטיחים ללא אובדן נתונים גם במקרה של כשל בעובד.

קבלו ייעוץ—נציע ארכיטקטורה והערכה מדויקת. צרו קשר כדי לדון בצרכי הייבוא שלכם.