Równoległe wykonywanie zadań asynchronicznych

W Parku Jurajskim, jednoczesne zarządzanie wieloma systemami i monitorowanie dziesiątek dinozaurów wymaga wydajnego podejścia do wykonywania wielu zadań jednocześnie. W tym ćwiczeniu skupimy się na technikach równoległego wykonywania zadań asynchronicznych w JavaScript, co pozwoli nam zwiększyć wydajność naszych systemów zarządzania parkiem.

Dlaczego równoległe wykonywanie jest ważne?

Wyobraźmy sobie, że musimy sprawdzić status wszystkich 50 wybiegów dinozaurów przed otwarciem parku. Jeśli każde sprawdzenie zajmuje około 1 sekundy i wykonujemy je sekwencyjnie (jedno po drugim), cały proces zajmie co najmniej 50 sekund - zdecydowanie za długo przed wpuszczeniem odwiedzających!

Wykonując te zadania równolegle, możemy znacznie skrócić całkowity czas potrzebny na przeprowadzenie wszystkich operacji, ograniczając go do czasu trwania najdłuższej pojedynczej operacji (plus pewien narzut).

Promise.all() - podstawowe narzędzie do równoległego wykonywania

Jak już widzieliśmy w poprzednich lekcjach,

Promise.all()

jest podstawowym narzędziem do równoległego wykonywania wielu zadań asynchronicznych:

1async function sprawdźStatusWszystkichWybiegów() {
2  const listaWybiegów = await pobierzListęWybiegów();
3  console.log(`Sprawdzanie statusu ${listaWybiegów.length} wybiegów...`);
4
5  const startTime = Date.now();
6
7  // Uruchomienie sprawdzania wszystkich wybiegów równolegle
8  const wyniki = await Promise.all(
9    listaWybiegów.map(wybieg => sprawdźStatusWybiegu(wybieg.id))
10  );
11
12  const czasWykonania = (Date.now() - startTime) / 1000;
13  console.log(`Sprawdzenie wszystkich wybiegów zajęło ${czasWykonania.toFixed(2)} sekund`);
14
15  // Analizuj wyniki
16  const bezpieczneWybiegi = wyniki.filter(w => w.status === 'secure');
17  const problematyczneWybiegi = wyniki.filter(w => w.status !== 'secure');
18
19  console.log(`Bezpieczne wybiegi: ${bezpieczneWybiegi.length}`);
20  console.log(`Problematyczne wybiegi: ${problematyczneWybiegi.length}`);
21
22  if (problematyczneWybiegi.length > 0) {
23    console.warn("Znaleziono problemy w następujących wybiegach:");
24    problematyczneWybiegi.forEach(w => {
25      console.warn(`- Wybieg #${w.id}: ${w.status} (${w.message})`);
26    });
27  }
28
29  return {
30    wszystkie: wyniki.length,
31    bezpieczne: bezpieczneWybiegi.length,
32    problematyczne: problematyczneWybiegi,
33    czasWykonania
34  };
35}

W powyższym przykładzie, używamy

Promise.all()

do równoległego sprawdzenia statusu wszystkich wybiegów. Zamiast czekać na zakończenie każdego sprawdzenia przed rozpoczęciem następnego, wszystkie sprawdzenia są inicjowane jednocześnie, a my czekamy na zakończenie wszystkich.

Wyzwania związane z równoległym wykonywaniem

Równoległe wykonywanie zadań asynchronicznych przynosi wiele korzyści, ale wiąże się także z pewnymi wyzwaniami:

1. Zarządzanie zasobami

Uruchomienie zbyt wielu operacji równolegle może przeciążyć dostępne zasoby systemowe (pamięć, połączenia sieciowe, itp.).

2. Obsługa błędów

Pamiętaj, że

Promise.all()

zakończy się niepowodzeniem, jeśli którakolwiek z obietnic zostanie odrzucona. W przypadku dużej liczby równoległych operacji, jeden błąd może spowodować niepowodzenie całego procesu.

3. Zbyt dużo równoległości

Nie zawsze uruchamianie wszystkich zadań równolegle jest najlepszym rozwiązaniem. Czasami potrzebujemy kontrolować maksymalną liczbę równoległych operacji.

Kontrolowanie liczby równoległych operacji

W sytuacjach, gdy mamy do wykonania wiele zadań, ale nie chcemy uruchamiać ich wszystkich jednocześnie, możemy zastosować technikę "równoległości z ograniczeniem":

1/**
2 * Wykonuje tablicę zadań z ograniczoną liczbą równoległych operacji
3 * @param {Function[]} zadania - Tablica funkcji, które zwracają Promise
4 * @param {number} maxRównoległe - Maksymalna liczba równoległych operacji
5 * @returns {Promise<any[]>} - Promise z tablicą wyników
6 */
7async function wykonajZOgraniczonaRownoleglością(zadania, maxRównoległe) {
8  // Tablica dla wszystkich wyników
9  const wyniki = [];
10  // Tablica dla aktywnych zadań
11  const aktywneZadania = new Set();
12  // Indeks bieżącego zadania
13  let indeksZadania = 0;
14
15  // Funkcja pomocnicza do uruchamiania zadań
16  const uruchomZadanie = async (indeks) => {
17    // Dodanie zadania do aktywnych
18    const zadanie = zadania[indeks];
19    aktywneZadania.add(indeks);
20
21    try {
22      // Wykonanie zadania
23      const wynik = await zadanie();
24      wyniki[indeks] = { status: 'fulfilled', value: wynik };
25    } catch (error) {
26      wyniki[indeks] = { status: 'rejected', reason: error };
27    } finally {
28      // Usunięcie zadania z aktywnych
29      aktywneZadania.delete(indeks);
30
31      // Jeśli są jeszcze zadania do wykonania, uruchom następne
32      if (indeksZadania < zadania.length) {
33        uruchomZadanie(indeksZadania++);
34      } else if (aktywneZadania.size === 0) {
35        // Jeśli nie ma więcej zadań i wszystkie aktywne zakończone, zakończ
36        return;
37      }
38    }
39  };
40
41  // Uruchom początkową pulę zadań
42  const początkowe = Math.min(maxRównoległe, zadania.length);
43  for (let i = 0; i < początkowe; i++) {
44    uruchomZadanie(indeksZadania++);
45  }
46
47  // Czekaj na zakończenie wszystkich zadań
48  await new Promise(resolve => {
49    const interval = setInterval(() => {
50      if (indeksZadania >= zadania.length && aktywneZadania.size === 0) {
51        clearInterval(interval);
52        resolve();
53      }
54    }, 100);
55  });
56
57  return wyniki;
58}
59
60// Przykład użycia
61async function monitorujDinozaury() {
62  const listaDinozaurów = await pobierzListęDinozaurów();
63
64  // Tworzymy tablicę funkcji, które zwracają Promise
65  const zadania = listaDinozaurów.map(dino => {
66    return async () => {
67      console.log(`Rozpoczynam monitorowanie dinozaura #${dino.id}...`);
68      const wynik = await monitorujDinozaura(dino.id);
69      console.log(`Zakończono monitorowanie dinozaura #${dino.id}`);
70      return wynik;
71    };
72  });
73
74  console.log(`Monitorowanie ${zadania.length} dinozaurów z maksymalnie 5 równolegle...`);
75  const wyniki = await wykonajZOgraniczonaRownoleglością(zadania, 5);
76
77  // Analiza wyników
78  const udane = wyniki.filter(w => w.status === 'fulfilled').length;
79  const nieudane = wyniki.filter(w => w.status === 'rejected').length;
80
81  console.log(`Ukończono monitorowanie: ${udane} udanych, ${nieudane} nieudanych`);
82
83  return {
84    wszystkie: wyniki.length,
85    udane,
86    nieudane,
87    wyniki
88  };
89}

W powyższym przykładzie,

wykonajZOgraniczonaRownoleglością

zapewnia, że w dowolnym momencie wykonywanych jest maksymalnie

maxRównoległe

zadań. Gdy jedno zadanie zostaje zakończone, automatycznie uruchamiane jest kolejne z kolejki.

Biblioteki pomagające w zarządzaniu równoległością

Istnieje kilka popularnych bibliotek, które ułatwiają zarządzanie równoległym wykonywaniem zadań:

p-limit

Biblioteka p-limit pozwala na łatwe ograniczenie liczby równoległych operacji:

1const pLimit = require('p-limit');
2
3async function monitorujWszystkieDinozaury() {
4  const listaDinozaurów = await pobierzListęDinozaurów();
5
6  // Utwórz instancję limitera z maksymalnie 5 równoległymi operacjami
7  const limit = pLimit(5);
8
9  // Utwórz tablicę obietnic z ograniczoną równoległością
10  const obietnice = listaDinozaurów.map(dino => {
11    return limit(() => monitorujDinozaura(dino.id));
12  });
13
14  // Czekaj na zakończenie wszystkich operacji
15  const wyniki = await Promise.all(obietnice);
16
17  return wyniki;
18}

async

Biblioteka async oferuje szeroki zestaw narzędzi do zarządzania zadaniami asynchronicznymi:

1const async = require('async');
2
3async function uruchomDiagnostykęSystemów() {
4  const systemy = [
5    { id: 'ogrodzenie', nazwa: 'System ogrodzeń elektrycznych' },
6    { id: 'kamery', nazwa: 'System monitoringu wideo' },
7    { id: 'czujniki', nazwa: 'System czujników ruchu' },
8    // ... więcej systemów
9  ];
10
11  // Wywołaj diagnostykę dla maksymalnie 3 systemów równolegle
12  const wyniki = await new Promise((resolve, reject) => {
13    async.mapLimit(
14      systemy,
15      3,
16      async (system) => {
17        console.log(`Uruchamiam diagnostykę systemu: ${system.nazwa}...`);
18        const wynik = await diagnostykaSystemu(system.id);
19        console.log(`Zakończono diagnostykę systemu: ${system.nazwa}`);
20        return { system, wynik };
21      },
22      (err, results) => {
23        if (err) reject(err);
24        else resolve(results);
25      }
26    );
27  });
28
29  return wyniki;
30}

Strategie równoległego przetwarzania

Poniżej znajduje się kilka użytecznych strategii równoległego przetwarzania, które można zastosować w różnych scenariuszach:

1. Przetwarzanie wsadowe

Zamiast przetwarzać wszystkie elementy naraz, możemy podzielić je na mniejsze partie i przetwarzać każdą partię równolegle:

1async function przetworzyćWszystkieDaneWPartie(elementy, wielkośćPartii = 10) {
2  const wyniki = [];
3
4  // Podziel elementy na partie
5  for (let i = 0; i < elementy.length; i += wielkośćPartii) {
6    const partia = elementy.slice(i, i + wielkośćPartii);
7    console.log(`Przetwarzanie partii ${Math.floor(i / wielkośćPartii) + 1}/${Math.ceil(elementy.length / wielkośćPartii)}...`);
8
9    // Przetwórz wszystkie elementy z bieżącej partii równolegle
10    const wynikiPartii = await Promise.all(
11      partia.map(element => przetwórzElement(element))
12    );
13
14    wyniki.push(...wynikiPartii);
15  }
16
17  return wyniki;
18}
19
20// Przykład użycia
21async function aktualizujDaneWszystkichDinozaurów(noweDane) {
22  const listaDinozaurów = await pobierzListęDinozaurów();
23
24  console.log(`Aktualizowanie danych dla ${listaDinozaurów.length} dinozaurów w partiach po 20...`);
25
26  const wyniki = await przetworzyćWszystkieDaneWPartie(
27    listaDinozaurów.map(dino => ({ id: dino.id, ...noweDane })),
28    20
29  );
30
31  console.log(`Zakończono aktualizację danych dla ${wyniki.length} dinozaurów`);
32
33  return wyniki;
34}

2. Kolejkowanie zadań priorytetowych

Czasami niektóre zadania są ważniejsze od innych i powinny być wykonywane w pierwszej kolejności:

1class KolejkaPriorytetowa {
2  constructor() {
3    // Mapa kolejek dla różnych priorytetów
4    this.kolejki = new Map();
5    // Aktualnie przetwarzane zadania
6    this.aktywne = new Set();
7    // Maksymalna liczba równoległych zadań
8    this.maxRównoległe = 5;
9    // Czy kolejka jest aktywna
10    this.aktywna = false;
11  }
12
13  // Dodaj zadanie do kolejki
14  dodajZadanie(zadanie, priorytet = 1) {
15    if (!this.kolejki.has(priorytet)) {
16      this.kolejki.set(priorytet, []);
17    }
18
19    this.kolejki.get(priorytet).push(zadanie);
20
21    // Jeśli kolejka jest aktywna, spróbuj uruchomić zadania
22    if (this.aktywna) {
23      this.przetwarzajZadania();
24    }
25  }
26
27  // Rozpocznij przetwarzanie zadań
28  rozpocznij() {
29    this.aktywna = true;
30    this.przetwarzajZadania();
31  }
32
33  // Zatrzymaj przetwarzanie zadań
34  zatrzymaj() {
35    this.aktywna = false;
36  }
37
38  // Przetwarzaj zadania
39  async przetwarzajZadania() {
40    if (!this.aktywna) return;
41
42    // Sprawdź, czy możemy uruchomić więcej zadań
43    while (this.aktywne.size < this.maxRównoległe) {
44      // Pobierz zadanie o najwyższym priorytecie
45      const zadanie = this.pobierzNastępneZadanie();
46
47      if (!zadanie) {
48        // Brak zadań do przetworzenia
49        break;
50      }
51
52      // Utwórz unikalny identyfikator zadania
53      const id = Symbol();
54
55      // Dodaj zadanie do aktywnych
56      this.aktywne.add(id);
57
58      // Uruchom zadanie
59      zadanie()
60        .catch(error => {
61          console.error("Błąd zadania:", error);
62        })
63        .finally(() => {
64          // Usuń zadanie z aktywnych
65          this.aktywne.delete(id);
66
67          // Kontynuuj przetwarzanie
68          this.przetwarzajZadania();
69        });
70    }
71  }
72
73  // Pobierz zadanie o najwyższym priorytecie
74  pobierzNastępneZadanie() {
75    // Sortuj priorytety malejąco
76    const priorytety = [...this.kolejki.keys()].sort((a, b) => b - a);
77
78    for (const priorytet of priorytety) {
79      const kolejka = this.kolejki.get(priorytet);
80
81      if (kolejka.length > 0) {
82        return kolejka.shift();
83      }
84    }
85
86    return null;
87  }
88
89  // Sprawdź, czy wszystkie zadania zostały zakończone
90  czyWszystkoZakończone() {
91    return this.aktywne.size === 0 &&
92           [...this.kolejki.values()].every(kolejka => kolejka.length === 0);
93  }
94
95  // Czekaj na zakończenie wszystkich zadań
96  async czekajNaZakończenie() {
97    if (this.czyWszystkoZakończone()) {
98      return;
99    }
100
101    return new Promise(resolve => {
102      const interwał = setInterval(() => {
103        if (this.czyWszystkoZakończone()) {
104          clearInterval(interwał);
105          resolve();
106        }
107      }, 100);
108    });
109  }
110}
111
112// Przykład użycia
113async function monitorujParkZPriorytetami() {
114  const kolejka = new KolejkaPriorytetowa();
115
116  // Dodawanie zadań z różnymi priorytetami
117
118  // Priorytet najwyższy (3) - systemy bezpieczeństwa
119  kolejka.dodajZadanie(
120    async () => {
121      console.log("Sprawdzanie systemów bezpieczeństwa...");
122      return await sprawdźSystemyBezpieczeństwa();
123    },
124    3
125  );
126
127  // Priorytet średni (2) - drapieżniki
128  const drapieżniki = await pobierzListęDrapieżników();
129  drapieżniki.forEach(dino => {
130    kolejka.dodajZadanie(
131      async () => {
132        console.log(`Monitorowanie drapieżnika #${dino.id}...`);
133        return await monitorujDinozaura(dino.id);
134      },
135      2
136    );
137  });
138
139  // Priorytet normalny (1) - roślinożercy
140  const roślinożercy = await pobierzListęRoślinożerców();
141  roślinożercy.forEach(dino => {
142    kolejka.dodajZadanie(
143      async () => {
144        console.log(`Monitorowanie roślinożercy #${dino.id}...`);
145        return await monitorujDinozaura(dino.id);
146      },
147      1
148    );
149  });
150
151  // Rozpocznij przetwarzanie
152  console.log("Rozpoczynam monitorowanie parku z priorytetyzacją zadań...");
153  kolejka.rozpocznij();
154
155  // Czekaj na zakończenie wszystkich zadań
156  await kolejka.czekajNaZakończenie();
157
158  console.log("Monitorowanie parku zakończone");
159}

3. Przetwarzanie strumieniowe

W przypadku dużych zbiorów danych, które pojawiają się w czasie, przetwarzanie strumieniowe może być bardziej efektywne:

1async function* generatorDanychDinozaurów() {
2  const strumień = await otwórzStrumieńDanych('/var/log/dino-sensors.log');
3
4  for await (const linia of strumień) {
5    // Przetwarzanie i parsowanie linii danych
6    const dane = JSON.parse(linia);
7    yield dane;
8  }
9}
10
11async function monitorujStrumieńDanychDinozaurów() {
12  const wyniki = [];
13  const aktywnePromisy = new Set();
14  const maxRównoległe = 10;
15
16  async function przetwórzDane(dane) {
17    try {
18      // Przetwarzanie danych i aktualizacja statusu
19      const wynik = await aktualizujStatusDinozaura(dane.dinoId, dane);
20      wyniki.push({ id: dane.dinoId, status: 'success', result: wynik });
21    } catch (error) {
22      wyniki.push({ id: dane.dinoId, status: 'error', error: error.message });
23    }
24  }
25
26  for await (const dane of generatorDanychDinozaurów()) {
27    // Czekaj, jeśli osiągnięto maksymalną liczbę równoległych operacji
28    if (aktywnePromisy.size >= maxRównoległe) {
29      await Promise.race(aktywnePromisy);
30    }
31
32    // Utwórz nowy promise i śledź go
33    const promisa = przetwórzDane(dane);
34    aktywnePromisy.add(promisa);
35
36    // Po zakończeniu, usuń z aktywnych
37    promisa.finally(() => {
38      aktywnePromisy.delete(promisa);
39    });
40  }
41
42  // Czekaj na zakończenie wszystkich aktywnych obietnic
43  if (aktywnePromisy.size > 0) {
44    await Promise.all(aktywnePromisy);
45  }
46
47  return wyniki;
48}

Równoległe przetwarzanie z obsługą błędów

Jak wspomnieliśmy wcześniej,

Promise.all()

zawiedzie, jeśli którakolwiek z obietnic zostanie odrzucona. W niektórych przypadkach chcemy kontynuować przetwarzanie, nawet jeśli niektóre operacje zakończą się niepowodzeniem.

W ECMAScript 2020, wprowadzono

Promise.allSettled()

, które rozwiązuje ten problem:

1async function uruchomDiagnostykęWszystkichSystemów() {
2  const systemy = await pobierzWszystkieSystemy();
3
4  console.log(`Uruchamianie diagnostyki dla ${systemy.length} systemów...`);
5
6  // Uruchom diagnostykę wszystkich systemów równolegle i poczekaj na zakończenie wszystkich
7  const wyniki = await Promise.allSettled(
8    systemy.map(async (system) => {
9      try {
10        console.log(`Diagnostyka systemu ${system.name}...`);
11        const wynik = await diagnostykaSystemu(system.id);
12        console.log(`Diagnostyka systemu ${system.name} zakończona`);
13        return { systemId: system.id, status: 'success', result: wynik };
14      } catch (error) {
15        console.error(`Błąd diagnostyki systemu ${system.name}:`, error);
16        return { systemId: system.id, status: 'error', error: error.message };
17      }
18    })
19  );
20
21  // Analiza wyników
22  const udane = wyniki.filter(w => w.status === 'fulfilled' && w.value.status === 'success');
23  const nieudane = wyniki.filter(w => w.status === 'rejected' || (w.status === 'fulfilled' && w.value.status === 'error'));
24
25  console.log(`Diagnostyka zakończona: ${udane.length} udanych, ${nieudane.length} nieudanych`);
26
27  if (nieudane.length > 0) {
28    console.warn("Systemy z problemami:");
29    nieudane.forEach(w => {
30      if (w.status === 'rejected') {
31        console.warn(`- System #${w.systemId}: Błąd krytyczny: ${w.reason}`);
32      } else {
33        console.warn(`- System #${w.value.systemId}: ${w.value.error}`);
34      }
35    });
36  }
37
38  return {
39    wszystkie: wyniki.length,
40    udane: udane.length,
41    nieudane: nieudane.length,
42    wyniki
43  };
44}

Zaawansowane techniki równoległego przetwarzania

1. Worker Threads w Node.js

W Node.js, możemy użyć modułu

worker_threads

do wykonywania obliczeń intensywnych w oddzielnych wątkach:

1const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
2
3// Główny kod
4if (isMainThread) {
5  async function analizujDaneDinozaurów() {
6    const dane = await pobierzDużeZbioryDanych();
7
8    // Podziel dane na części, jedną dla każdego rdzenia CPU
9    const ilośćWątków = require('os').cpus().length;
10    const częściDanych = podzielDane(dane, ilośćWątków);
11
12    // Utwórz worker dla każdej części danych
13    const workers = częściDanych.map((część, indeks) => {
14      return new Promise((resolve, reject) => {
15        const worker = new Worker(__filename, {
16          workerData: { dane: część, indeks }
17        });
18
19        worker.on('message', resolve);
20        worker.on('error', reject);
21        worker.on('exit', (code) => {
22          if (code !== 0) {
23            reject(new Error(`Worker stopped with exit code ${code}`));
24          }
25        });
26      });
27    });
28
29    // Czekaj na zakończenie wszystkich workerów
30    const wyniki = await Promise.all(workers);
31
32    // Połącz wyniki
33    return połączWyniki(wyniki);
34  }
35} else {
36  // Kod wykonywany w każdym wątku pracownika
37  const { dane, indeks } = workerData;
38
39  // Przeprowadź złożone obliczenia na danych
40  const wyniki = przetwórzDane(dane);
41
42  // Wyślij wyniki z powrotem do głównego wątku
43  parentPort.postMessage(wyniki);
44}

2. Web Workers w przeglądarce

W środowisku przeglądarki, możemy użyć Web Workers do wykonywania zadań w tle:

1// main.js - główny kod w przeglądarce
2function uruchomAnalizęDNA() {
3  const dnaSamples = pobierzPróbkiDNA();
4
5  // Podziel próbki na grupy
6  const groupSize = Math.ceil(dnaSamples.length / 4); // 4 grupy
7  const groups = [];
8
9  for (let i = 0; i < dnaSamples.length; i += groupSize) {
10    groups.push(dnaSamples.slice(i, i + groupSize));
11  }
12
13  // Utwórz worker dla każdej grupy
14  const workers = groups.map((group, index) => {
15    return new Promise((resolve, reject) => {
16      const worker = new Worker('dna-worker.js');
17
18      worker.onmessage = function(e) {
19        resolve(e.data);
20        worker.terminate();
21      };
22
23      worker.onerror = function(e) {
24        reject(new Error(e.message));
25        worker.terminate();
26      };
27
28      worker.postMessage({
29        samples: group,
30        groupId: index
31      });
32    });
33  });
34
35  // Czekaj na wyniki ze wszystkich workerów
36  return Promise.all(workers)
37    .then(results => {
38      // Połącz wyniki
39      return results.flat();
40    });
41}
42
43// dna-worker.js - kod workera
44self.onmessage = function(e) {
45  const { samples, groupId } = e.data;
46
47  // Wykonaj złożoną analizę DNA
48  const results = samples.map(sample => analyzeDNASample(sample));
49
50  // Wyślij wyniki z powrotem do głównego wątku
51  self.postMessage(results);
52};
53
54function analyzeDNASample(sample) {
55  // Symulacja złożonej analizy
56  //
57  return {
58    sampleId: sample.id,
59    species: determineSpecies(sample.dna),
60    genomeMap: mapGenome(sample.dna),
61    //
62  };
63}

Strategie zarządzania zasobami

Gdy wykonujemy wiele zadań równolegle, ważne jest zarządzanie dostępnymi zasobami:

1. Monitorowanie zużycia zasobów

1async function wykonajZMonitorowaniemZasobów(zadania, opcje = {}) {
2  const {
3    maxRównoległe = 5,
4    maxPamięć = 1024 * 1024 * 1024, // 1GB
5    interwałMonitorowania = 1000 // 1s
6  } = opcje;
7
8  const aktywneZadania = new Set();
9  const zakończoneZadania = [];
10  let indeksZadania = 0;
11
12  // Uruchom początkowe zadania
13  for (let i = 0; i < Math.min(maxRównoległe, zadania.length); i++) {
14    uruchomZadanie(indeksZadania++);
15  }
16
17  // Monitorowanie zasobów
18  const interwał = setInterval(() => {
19    const wykorzystaniePamięci = process.memoryUsage().heapUsed;
20
21    // Jeśli zużycie pamięci jest zbyt wysokie, wstrzymaj uruchamianie nowych zadań
22    if (wykorzystaniePamięci > maxPamięć && aktywneZadania.size > 1) {
23      console.warn(`Wysokie zużycie pamięci: ${(wykorzystaniePamięci / 1024 / 1024).toFixed(2)} MB. Zwalnianie...`);
24      // Nie uruchamiaj nowych zadań, dopóki pamięć nie zostanie zwolniona
25    }
26
27    // Sprawdź, czy wszystkie zadania zostały zakończone
28    if (indeksZadania >= zadania.length && aktywneZadania.size === 0) {
29      clearInterval(interwał);
30    }
31  }, interwałMonitorowania);
32
33  async function uruchomZadanie(indeks) {
34    if (indeks >= zadania.length) {
35      return;
36    }
37
38    const identyfikator = Symbol();
39    aktywneZadania.add(identyfikator);
40
41    try {
42      const wynik = await zadania[indeks]();
43      zakończoneZadania[indeks] = { status: 'fulfilled', value: wynik };
44    } catch (error) {
45      zakończoneZadania[indeks] = { status: 'rejected', reason: error };
46    } finally {
47      aktywneZadania.delete(identyfikator);
48
49      // Sprawdź wykorzystanie pamięci przed uruchomieniem kolejnego zadania
50      const wykorzystaniePamięci = process.memoryUsage().heapUsed;
51
52      if (wykorzystaniePamięci < maxPamięć || aktywneZadania.size === 0) {
53        // Uruchom kolejne zadanie, jeśli pamięć pozwala lub nie ma aktywnych zadań
54        if (indeksZadania < zadania.length) {
55          uruchomZadanie(indeksZadania++);
56        }
57      }
58    }
59  }
60
61  // Czekaj na zakończenie wszystkich zadań
62  while (zakończoneZadania.length < zadania.length) {
63    await new Promise(resolve => setTimeout(resolve, 100));
64  }
65
66  clearInterval(interwał);
67
68  return zakończoneZadania;
69}

2. Dynamiczne dostosowywanie równoległości

W zależności od bieżącego obciążenia systemu, możemy dynamicznie dostosowywać poziom równoległości:

1class AdaptiveTaskProcessor {
2  constructor(options = {}) {
3    this.minParallelism = options.minParallelism || 1;
4    this.maxParallelism = options.maxParallelism || 10;
5    this.currentParallelism = options.startParallelism || 5;
6    this.tasks = [];
7    this.activeTasks = new Set();
8    this.completedTasks = [];
9    this.running = false;
10    this.performanceHistory = [];
11    this.adaptationInterval = options.adaptationInterval || 5000; // 5s
12  }
13
14  addTask(task) {
15    this.tasks.push(task);
16    if (this.running) {
17      this.processQueue();
18    }
19  }
20
21  async start() {
22    this.running = true;
23    this.processQueue();
24
25    // Start adaptation
26    this.adaptationTimer = setInterval(() => {
27      this.adaptParallelism();
28    }, this.adaptationInterval);
29  }
30
31  stop() {
32    this.running = false;
33    clearInterval(this.adaptationTimer);
34  }
35
36  async processQueue() {
37    if (!this.running) return;
38
39    while (this.activeTasks.size < this.currentParallelism && this.tasks.length > 0) {
40      const task = this.tasks.shift();
41      await this.executeTask(task);
42    }
43  }
44
45  async executeTask(task) {
46    const id = Symbol();
47    this.activeTasks.add(id);
48
49    const startTime = Date.now();
50
51    try {
52      const result = await task();
53      this.completedTasks.push({
54        status: 'fulfilled',
55        value: result,
56        duration: Date.now() - startTime
57      });
58    } catch (error) {
59      this.completedTasks.push({
60        status: 'rejected',
61        reason: error,
62        duration: Date.now() - startTime
63      });
64    } finally {
65      this.activeTasks.delete(id);
66
67      // Add to performance history
68      this.performanceHistory.push({
69        parallelism: this.currentParallelism,
70        duration: Date.now() - startTime
71      });
72
73      // Trim history to last 100 tasks
74      if (this.performanceHistory.length > 100) {
75        this.performanceHistory.shift();
76      }
77
78      this.processQueue();
79    }
80  }
81
82  adaptParallelism() {
83    if (this.performanceHistory.length < 10) {
84      // Not enough data to adapt
85      return;
86    }
87
88    // Calculate average task duration for current parallelism
89    const recentTasks = this.performanceHistory.slice(-20);
90    const avgDuration = recentTasks.reduce((sum, t) => sum + t.duration, 0) / recentTasks.length;
91
92    // Check if we've tracked different parallelism levels
93    const levels = new Set(this.performanceHistory.map(t => t.parallelism));
94
95    if (levels.size < 2) {
96      // Try a different parallelism level
97      this.currentParallelism = Math.max(
98        this.minParallelism,
99        Math.min(
100          this.maxParallelism,
101          this.currentParallelism + (Math.random() > 0.5 ? 1 : -1)
102        )
103      );
104      console.log(`Adjusting parallelism to ${this.currentParallelism} (exploration)`);
105      return;
106    }
107
108    // Group by parallelism level
109    const byLevel = {};
110    this.performanceHistory.forEach(t => {
111      if (!byLevel[t.parallelism]) {
112        byLevel[t.parallelism] = [];
113      }
114      byLevel[t.parallelism].push(t.duration);
115    });
116
117    // Calculate average duration per parallelism level
118    const avgByLevel = {};
119    Object.entries(byLevel).forEach(([level, durations]) => {
120      avgByLevel[level] = durations.reduce((sum, d) => sum + d, 0) / durations.length;
121    });
122
123    // Find the best parallelism level
124    const bestLevel = Object.entries(avgByLevel)
125      .sort(([, durationA], [, durationB]) => durationA - durationB)[0][0];
126
127    const newParallelism = parseInt(bestLevel, 10);
128
129    if (newParallelism !== this.currentParallelism) {
130      console.log(`Adjusting parallelism to ${newParallelism} (optimization)`);
131      this.currentParallelism = newParallelism;
132    }
133  }
134
135  async waitForAll() {
136    this.stop();
137
138    while (this.tasks.length > 0 || this.activeTasks.size > 0) {
139      await new Promise(resolve => setTimeout(resolve, 100));
140    }
141
142    return this.completedTasks;
143  }
144}
145
146// Przykład użycia
147async function wykonajZadaniaZAdaptacją() {
148  const processor = new AdaptiveTaskProcessor({
149    minParallelism: 1,
150    maxParallelism: 20,
151    startParallelism: 5
152  });
153
154  // Dodaj zadania
155  for (let i = 0; i < 100; i++) {
156    processor.addTask(async () => {
157      const zadanieId = i;
158      console.log(`Rozpoczęcie zadania #${zadanieId}`);
159
160      // Symulacja zadania o różnej złożoności
161      const złożoność = Math.random() * 1000 + 500;
162      await new Promise(resolve => setTimeout(resolve, złożoność));
163
164      console.log(`Zakończenie zadania #${zadanieId}`);
165      return { zadanieId, złożoność };
166    });
167  }
168
169  // Uruchom przetwarzanie
170  processor.start();
171
172  // Czekaj na zakończenie
173  const wyniki = await processor.waitForAll();
174
175  console.log(`Zakończono wszystkie zadania. Wyniki: ${wyniki.length}`);
176
177  return wyniki;
178}

Podsumowanie

Równoległe wykonywanie zadań asynchronicznych jest kluczowym narzędziem w optymalizacji wydajności aplikacji JavaScript. W Parku Jurajskim, gdzie monitorowanie i zarządzanie wieloma systemami i dinozaurami wymaga szybkiej reakcji, właściwe wykorzystanie technik równoległego przetwarzania może znacznie poprawić responsywność i skalowalność systemu.

Poznaliśmy kilka ważnych technik i wzorców:

1. Promise.all() - podstawowe narzędzie do równoległego wykonywania wielu zadań asynchronicznych
2. Kontrolowanie liczby równoległych operacji - techniki ograniczania maksymalnej równoległości
3. Przetwarzanie wsadowe - przetwarzanie danych w partiach
4. Kolejkowanie zadań z priorytetami - wykonywanie ważniejszych zadań w pierwszej kolejności
5. Przetwarzanie strumieniowe - przetwarzanie danych w miarę ich napływania
6. Worker Threads i Web Workers - wykonywanie zadań w oddzielnych wątkach
7. Adaptacyjne zarządzanie zasobami - dynamiczne dostosowywanie poziomu równoległości

Wybór odpowiedniej techniki zależy od konkretnego przypadku użycia, wymagań wydajnościowych oraz dostępnych zasobów. Pamiętaj, że nie zawsze więcej równoległości oznacza lepszą wydajność - zbyt wiele równoległych operacji może prowadzić do przeciążenia systemu, a w konsekwencji do pogorszenia ogólnej wydajności.

W praktyce, najlepsze wyniki często osiąga się przez eksperymentowanie z różnymi strategiami i poziomami równoległości, monitorowanie wydajności i odpowiednie dostosowywanie parametrów.