====== Futurs a Java ======

{{tag> #FpInfor #Dam #DamMp09 #DamMp09Uf2 #DamMp09Uf02}}


----

==== Exemples ====

{{ ::futurs_java_exemple.zip |}}

==== Futurs ====

La clase ‘Futur’ de Java, representa el resultat futur d’un càlcul asíncron. 


És a dir, un resultat o unes dades que obtindrem en un futur i cal esperar sense aturar el funcionament normal del programa.


És habitual fer servir futurs en càlculs molt complexes o quan s’esperen dades d’un servidor que pot trigar en respondre.


En realitat es fan amb fils paral·lels (threads), però la implementació ens allibera de definir-los un a un


==== Futurs, un sol Futur amb “newSingleThreadExecutor” ====

**ExecutorService**: crea un thread automàticament per executar la tasca que ha de retornar un futur


**.submit()**: permet fer el càlcul en paral·lel i retornar el Futur quan ha acabat


**.shutdown()**: atura el thread paral·lel


En aquest exemple es força una espera d’1 segon per simular una operació lenta

<code java>
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class NumeroQuadratCalcul {

   private ExecutorService executor =        
           Executors.newSingleThreadExecutor();

    public Future<Integer> calculate(Integer input) {
       
       return executor.submit(() -> {
           Thread.sleep(1000);
           return input * input;
       });
   }
   public void shutdown () { executor.shutdown(); }
}

</code>

<code java>
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class NumeroQuadratMain {
    public static void main(String args[]) {
        NumeroQuadratCalcul obj = new NumeroQuadratCalcul();
        Future<Integer> future = obj.calculate(7);

        while (!future.isDone()) {
            System.out.println("Calculant...");
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) { e.printStackTrace(); }
        }

        try {
            Integer result = future.get();
            System.out.println("Resultat: " + result);
            obj.shutdown();
        } catch (InterruptedException e) { e.printStackTrace();
        } catch (ExecutionException e) { e.printStackTrace(); }
    }
}

</code>


Quan demanem el càlcul (lent), rebem un futur enlloc de la dada


Hem d’esperar a que el Futur estigui disponible “**.isDone()**”


Quan ja està llest, afagem la dada resultant amb “**.get()**”

==== Futurs, aturar un futur amb .cancel() ==== 

A vegades esperar un futur triga més del què podem acceptar, per aturar-lo podem fer “.cancel()”

En aquest exemple es crea un futur que triga entre 2 i 4 segons a acabar, el programa principal l’atura si triga més de 3.

<code java>
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class EsperaSegonsCalcul {
    private ExecutorService executor = Executors.newSingleThreadExecutor();
  
    public Future<Integer> calculate(Integer input) {        
        return executor.submit(() -> {
            long millis = (new Random()).nextInt(3000) + 1000;
            System.out.println("Trigarà: " + (millis / 1000.0) + " milisegons");
            Thread.sleep(millis);
            return input * input;
        });
    }

    public void shutdown () { executor.shutdown(); }
}

</code>

<code java>
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class EsperaSegonsMain {
    public static void main(String args[]) {
        EsperaSegonsCalcul obj = new EsperaSegonsCalcul();
        Future<Integer> future = obj.calculate(7);
        long start = System.currentTimeMillis();
        long finish = 0;

        while (!future.isDone()) {
            finish = System.currentTimeMillis();
            double segons = (finish - start) / 1000.0;
            System.out.println("Esperant ... " + segons + " segons");
            if (segons > 2) {
                // Si esperem més de 2 segons, ja no volem el resultat
                future.cancel(true);
            }
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) { e.printStackTrace(); }
        }

        try {
            if (future.isCancelled()) {
                System.out.println("El resultat ha trigat massa");
            } else {
                Integer result = future.get();
                System.out.println("Resultat: " + result);
            }
            obj.shutdown();
        } catch (InterruptedException e) { e.printStackTrace();
        } catch (ExecutionException e) { e.printStackTrace(); }
    }  
}

</code>

==== Futurs, multiples futurs amb ‘newFixedThreadPool’ ====

Per executar múltiples futurs alhora, cal iniciar “executor” amb “newfixedThreadPool(X)” on X representa el número de fils que funcionen en paral·lel

<code java>
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class MultiplesFilsCalcul {
    private ExecutorService executor = Executors.newFixedThreadPool(4);

    public Future<Integer> calculate(Integer input) {        
        return executor.submit(() -> {
            long millis = (new Random()).nextInt(4000) + 500;
            System.out.println("Calcular " + input + " trigarà " + (millis / 1000.0) + " milisegons");
            Thread.sleep(millis);
            return input * input;
        });
    }
    public void shutdown () { executor.shutdown(); }
}
</code>

I podrem crear un “Futur” per cada un dels fils que tenim disponibles. Cada un d’aquests futurs pot trigar diferent temps a executar-se.


<code java>
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class MultiplesFilsMain {

    public static void main(String args[]) {
        MultiplesFilsCalcul obj = new MultiplesFilsCalcul();
        Future<Integer> f0 = obj.calculate(1);
        Future<Integer> f1 = obj.calculate(2);
        Future<Integer> f2 = obj.calculate(4);
        Future<Integer> f3 = obj.calculate(8);
        String txt = "";

        while (!f0.isDone() || !f1.isDone() || !f2.isDone() || !f3.isDone()) {
            // Esperar a que estiguin llestos
            txt = String.format(
                "Estats: f0 %s, f1 %s, f2 %s, f3 %s", 
                f0.isDone() ? "llest" : "calculant", 
                f1.isDone() ? "llest" : "calculant",
                f2.isDone() ? "llest" : "calculant",
                f3.isDone() ? "llest" : "calculant"
            );
            System.out.println(txt);
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) { e.printStackTrace(); }
        }    

        try {
            txt = String.format(
                "Resultat: f0=%s, f1=%s, f2=%s, f3=%s", 
                f0.get(), f1.get(), f2.get(), f3.get()
            );
            System.out.println(txt);
            obj.shutdown();
        } catch (InterruptedException e) { e.printStackTrace();
        } catch (ExecutionException e) { e.printStackTrace(); }
    }
}

</code>

==== Futurs, Completable Future ====

**CompletableFuture** permet definir els processos sense necessitat de ‘**executors**’, bloquejant accions futures fins que han acabat els processos o fins i tot encadenant futurs. Es declaren amb:

**.runAsync** quan no tenen valor de retorn

**.supplyAsync** quan tenen valor de retorn

<code java>
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class CompletableMain {
    public static void main(String args[]) {

        // Declarar els processos asíncrons al codi directament
        CompletableFuture<Void> futureA = CompletableFuture.runAsync(getRunnable("A"));
        CompletableFuture<Void> futureB = CompletableFuture.runAsync(getRunnable("B"));

        // Posar els 'CompletableFuture' en una llista per esperar-los tots alhora
        // tots han de retornar el mateix tipus (en aquest cas String)
        System.out.println("Esperant futurs");
        List<CompletableFuture<Void>> futureList = new ArrayList<>();
        futureList.add(futureA);
        futureList.add(futureB);
        futureList.forEach(CompletableFuture::join); 
        System.out.println("Futurs llestos");
    }

    static Runnable getRunnable (String type) {
        return new Runnable () {
            @Override
            public void run () {
                try {
                    long millis = (new Random()).nextInt(4000) + 500;
                    System.out.println("futur" + type + " trigarà " + (millis / 1000.0) + " milisegons");
                    int cnt = 10;
                    while (cnt > 0) {
                        Thread.sleep(millis / 10);
                        System.out.println(" - " + type + " " + cnt);
                        cnt = cnt - 1;
                    }
                    System.out.println("future" + type + " acabat");
                    
                } catch (InterruptedException e) { e.printStackTrace(); }
            }
        };
    }
}

</code>

A més, permet fer una llista d’objectes tipus ‘CompletableFuture’ i esperar a que tots han acabat l’execució per continuar.

Tots han de retornar el mateix tipus, en aquest cas ‘String’

<code java>
System.out.println("Esperant resultats");
List<CompletableFuture<String>> futureList 
                                       = new ArrayList<>();
futureList.add(futureA);
futureList.add(futureB);
System.out.println("Resultats llestos");

</code>

==== Futurs, CompletableFuture amb retorn i sense retorn ====

Quan un Futur gestionat per CompletableFuture no té retorn, fem servir ‘.**runAsync**’ i li passem una funció que retorni un objecte **Runnable** sobreescrivint la runció ‘**run**’ que executa el càlcul

<code java>
CompletableFuture<Void> f0 = CompletableFuture.runAsync(getRunnable("ABC"));
</code>

<code java>
static Runnable getRunnable (String cadena) {
       return new Runnable () {
           @Override
           public void run() {
...
</code>

En canvi quan ha de retornar valors, fem servir ‘.**supplyAsync**’ i li passem una funció que retorni un objecte **Supplier<tipus>** sobreescrivint la funció ‘**get**’ que executa el càlcul

<code java>
CompletableFuture<String> f0 = CompletableFuture.supplyAsync(getSupplier("ABC"));
</code>

<code java>
static Supplier<String> getSupplier (String cadena) {
       return new Supplier<String> () {
           @Override
           public String get() {

...
</code>

==== Futurs, CompletableFuture amb expressions lambda ====

CompletableFuture permet definir els processos paral·lels directament al codi, a través del què s'anomena expressions ‘lambda’ que s’escriuen amb () -> { }

<code java>import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableLambdaMain {
    public static void main(String args[]) {

        // Declarar els processos asíncrons al codi directament
        CompletableFuture<String> futureA = CompletableFuture.supplyAsync(() -> {
            try {
                long millis = (new Random()).nextInt(4000) + 500;
                System.out.println("futurA trigarà " + (millis / 1000.0) + " milisegons");
                int cnt = 10;
                while (cnt > 0) {
                    Thread.sleep(millis / 10);
                    System.out.println(" - A " + cnt);
                    cnt = cnt - 1;
                }                    
            } catch (InterruptedException e) { e.printStackTrace(); }
            return "Done A";
        });

        CompletableFuture<String> futureB = CompletableFuture.supplyAsync(() -> {
            try {
                long millis = (new Random()).nextInt(4000) + 500;
                System.out.println("futurB trigarà " + (millis / 1000.0) + " milisegons");
                int cnt = 10;
                while (cnt > 0) {
                    Thread.sleep(millis / 10);
                    System.out.println(" - B " + cnt);
                    cnt = cnt - 1;
                }                    
            } catch (InterruptedException e) { e.printStackTrace(); }
            return "Done B";
        });

        // Posar els 'CompletableFuture' en una llista per esperar-los tots alhora
        System.out.println("Esperant resultats");
        List<CompletableFuture<String>> futureList = new ArrayList<>();
        futureList.add(futureA);
        futureList.add(futureB);
        futureList.forEach(CompletableFuture::join); 
        System.out.println("Resultats llestos");

        // Mostrar resultats
        try {
            String resultA = futureA.get(); 
            String resultB = futureB.get(); 
            System.out.println("Resultat A = " + resultA + ", Resultat B = " + resultB);
        } catch (InterruptedException e) { e.printStackTrace();
        } catch (ExecutionException e) { e.printStackTrace(); }
    }
}


</code>

==== CompletableFuture i dades compartides ====

Cada futur en realitat és un thread que es pot executar en paral·lel a altres futurs.

Si aquests futurs accedeixen a les mateixes dades, s’ha de tenir algun mètode de bloqueig, per tal que els altres threads no modifiquin les dades simultàniament i per tant aquestes siguin errònies.

Aquest bloqueig es pot aconseguir amb "ReentrantLock", tal i com mostra el següent exemple.

En aquest codi es veu com les dades sense bloqueig només poden comptar fins a 10 perquè els fils es trepitgen entre ells, però les que implementen el bloqueig aconsegueixen comptar fins a 50.

<code java>
import java.util.ArrayList;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.ReentrantLock;

public class FutMain {

    // Vector de dades
    private static int dadesSns = 0;
    private static int dadesBqg = 0;

    // Bloquejador de recursos
    private static ReentrantLock mutex = new ReentrantLock();

    // Main
    public static void main(String[] args) {
        
        // Creacio de futurs
        ArrayList<CompletableFuture<Void>> futureList = new ArrayList<>();
        for (int cnt = 0; cnt < 5; cnt++){
            futureList.add(CompletableFuture.runAsync((getRunnable(cnt))));
        }

        // Execucio dels futurs
        System.out.println("Esperant resultats");
        futureList.forEach(CompletableFuture::join); 

        // Mostrar resultats
        System.out.println("Resultats:");
        System.out.println("Dades sense bloqueig: " + dadesSns);
        System.out.println("Dades amb bloqueig: " + dadesBqg);
    }

    //Codi executat pels futurs
    static Runnable getRunnable (int pos) {
        return new Runnable () {
            @Override public void run () {

                // Modificar les dades sense bloqueig
                for (int cnt = 0; cnt < 10; cnt++) {
                    int val = dadesSns;
                    try {Thread.sleep(10);} 
                    catch (InterruptedException e) {e.printStackTrace();}
                    dadesSns = val + 1;
                }

                try {
                    // Bloquejem els recursos per assegurar la seva exclusivitat
                    mutex.lock();

                    // Modificar les dades amb bloqueig 
                    for (int cnt = 0; cnt < 10; cnt++) {
                        int val = dadesBqg;
                        try {Thread.sleep(10);} 
                        catch (InterruptedException e) {e.printStackTrace();}
                        dadesBqg = val + 1;
                    }

                } finally {
                    mutex.unlock();
                }
            }
        };
    }
}
</code>