Ayuda con multiprocesamiento en Java

Question

Muy bien, esta es una clase de pregunta de doble propósito. Lo principal que espero sacar de esto es más conocimiento de multihilo. Soy un novato completo cuando se trata de multihilo, y este es mi primer intento real de hacer algo en múltiples hilos. La otra cosa que espero quitar es algo de ayuda con una tarea que estoy convirtiendo en un proyecto franken mucho más complicado para la diversión y el aprendizaje. En la primera parte de esta pregunta voy a detallar mi pensamiento y metodología para los hilos en los que he estado trabajando en mi tarea. Si estoy haciendo algo que es una mala práctica, necesita ser reparado, lo que sea, házmelo saber para que pueda aprender. De nuevo, no sé prácticamente nada sobre multihilo.

En primer lugar, actualmente estoy tomando un curso de informática que, para ser agradable, tiene tareas que utilizan técnicas y estructuras de datos que ya he aprendido y que, por lo tanto, no representan un desafío. Para tratar de no aburrirme por completo, intento tomar un proyecto simple (crear una lista enlazada y una lista enlazada ordenada) y convertirlo en un programa franken multiproceso. Me gustaría que el método para agregar elementos nuevos esté en un hilo separado y obtenga entrada de una cola (no del todo necesario para una lista no ordenada, pero será más útil para el orden correcto), y para la lista ordenada quiero un hilo separado que patrullará la lista para asegurarse de que todos los elementos estén en orden (uno de los métodos que NO PUEDO cambiar devuelve referencias a nodos completos, que tienen datos completamente públicos). Esas son las dos cosas principales que me gustaría crear, pero definitivamente no son las únicas cosas que querría probar y descubrir. Esto pretende ser simplemente una prueba para aprender sobre el multihilo, por lo que no estoy especialmente diseñando este proyecto por su practicidad. Si desea dejar un comentario sobre lo que es una buena práctica de programación para lo que debe ser enhebrado o no, sería muy apreciado.

Como un paso que tomé hace poco, cuya idea fue obtenida al leer una publicación en StackOverflow, he creado una clase maestra que tiene Threads pasados ​​para que pueda detenerlos y limpiar todo al final Del programa. Esto se hace mediante el uso de someThread.interrupt() y con el método run() del Thread para verificar esa excepción. Sin embargo, he encontrado un problema con este método: el bucle que se ejecuta a través de los hilos y llamando a la interrupción(), la mayoría de las veces, no se ejecuta. Aquí está el código para el método:

public static void stopAllThreads() 
{ 
    boolean stopped = false; 
    while(!stopped) 
    { 
     System.out.println("Stopping all threads"); 

     synchronized(threads) 
     { 
      System.out.println("Threads being stopped: " + threads.size()); 
      for(int i = 0; i < threads.size(); i++) 
      { 
       System.out.println("Stopping: " + threads.get(i)); 
       threads.get(i).interrupt(); 
      } 
      threads.clear(); 
      stopped = true; 
     } 
    } 
} 

Al intentar depurar, lo puso en el bucle while para tratar de forzar el método a ejecutar finalmente el bucle, pero todo lo que sucede es que da salida a "Detener todo hilos "y luego nunca veo nada más. No sé si esta es una mala forma de codificar o qué está mal con este código. Se agradece cualquier ayuda para descubrir cómo hacer que esto funcione (si necesita ver más de la clase, hágamelo saber. Realmente no sé qué necesita ver, pero no quiero copiar y pegar varios archivos completos de Java).

Además de este problema, también he determinado que cuando ejecuto mi programa, llegará a la última línea e intentará detener todos los hilos antes de que el hilo pase por la cola de números para agregar a la lista. . Esto también es un problema cuando se intenta generar la lista a través de un método toString directamente después de agregar números a la cola, ya que no se agregará nada y, por lo tanto, no se generará nada en la consola. Aquí es mi método principal que muestra el orden de las cosas que suceden (aunque imagino que es menos importante cuando se trata de enhebrar):

public static void main(String[] args) 
{ 
    // always make sure this is the first thing created 
    ActiveThreads bla = new ActiveThreads(); 

    Comparator<Integer> temp = new Comparator<Integer>() 
      { 
       public int compare(Integer i, Integer j) 
       { 
        if(i > j) return 1; 
        if(i < j) return -1; 
        return 0; 
       } 
      }; 
    List<Integer> woogle = new List<Integer>(temp); 

    woogle.add(1); 
    woogle.add(2); 
    woogle.add(3); 
    woogle.add(4); 
    System.out.println(woogle); 

    ActiveThreads.stopAllThreads(); 
} 

Por último, aquí es el hilo a medida que he hecho que se supone para comprobar la cola de elementos, a continuación, añadir todos los elementos en la cola a mi lista real:

private class AddThread implements Runnable 
    { 
     public AddThread() 
     { 

     } 

     public void run() 
     { 
      while(running) 
      { 
       synchronized(addQueue) 
       { 
        while(!addQueue.isEmpty()) 
        { 
         System.out.println("Going through queue"); 
         T temp = addQueue.poll(); 
         if(head == null) 
         { 
          head = new Node<T>(); 
          last = head; 
          head.data = temp; 
          largest = temp; 
          smallest = temp; 
         } 
         else 
         { 
          last.next = new Node<T>(); 
          last.next.data = temp; 
          last = last.next; 
          if(mComparator.compare(temp, largest) == 1) 
          { 
           largest = temp; 
          } 
          else if(mComparator.compare(temp, smallest) == -1) 
          { 
           smallest = temp; 
          } 
         } 
         ++size; 
        } 
       } 

       System.out.println("Pausing " + addThread); 
       synchronized(addThread) 
       { 
        pause(200); 
       } 
      } 
     } 

     private void pause(long time) 
     { 
      try 
      { 
       addThread.wait(time); 
      } 
      catch (InterruptedException e) 
      { 
       running = false; 
       addThread.notify(); 
      } 
     } 

     private boolean running = true; 
    } 

Ahora, el absoluto último que me gustaría pedir son algunos de los cebadores y las directrices básicas y como por una buena multihilo.He encontrado una buena cantidad de tutoriales, pero hay algunas cosas que las personas no mencionan en los tutoriales, y la mayoría de los tutoriales son regulares. Si conoces algunos buenos tutoriales, agradecería tener enlaces a ellos. Si hay algunas buenas metodologías de codificación para multihilo, eso sería muy apreciado también. Cualquier tipo de información que piense que podría ayudarme a enfocar mi cerebro en este concepto, y me aseguro de que al implementarlo lo haga de una forma menos propensa a los errores y la gente no me grite por lo malo. prácticas.

EDITAR: Para responder a las preguntas que matt b puso en un comentario, addThread es puramente una referencia al hilo que mira la cola que tiene elementos para agregar a la lista y los coloca en la lista. Es un hilo privado dentro de la clase que contiene la clase AddThread (uso inteligente de nombres, ¿eh?). Los hilos se construyen de una manera que se publicó en otra pregunta en StackOverflow. Aquí está un ejemplo (y también un fragmento de código que muestra lo que es addThread):

addThread = new Thread(new AddThread(), "Thread for adding elements"); 
addThread.setPriority(6); 
addThread.start(); 
ActiveThreads.addThread(addThread); 

sus temas se crean de una manera similar. En cuanto a lo que se supone que está haciendo el hilo que publiqué, lo expliqué en detalle anteriormente, pero trataré de poner una sinopsis aquí.

La clase AddThread es una clase interna que se crea en el momento de la construcción de la clase de lista vinculada que estoy tratando de hacer. Mira una cola privada que puede tener o no elementos, y cuando lo hace, coloca esos elementos en la lista vinculada. La cola también se agrega cuando se llama al método add (T newElt) de la lista vinculada.

EDITAR EDITAR: En aras de la divulgación completa, y para aquellos que deseen tomarse el tiempo, voy a publicar la totalidad de los dos archivos java relevantes aquí.

List.java

public class List<T> implements Iterable<T>, Comparable< List<T> > 
{ 
    // you may not change this inner class! 
    class Node<D> 
    { 
     D data; 
     Node<D> next; 
    } 

    public List(Comparator<T> comparator) 
    { 
     mComparator = comparator; 
     head = null; 
     last = null; 
     size = 0; 
     addQueue = new LinkedList<T>(); 

     addThread = new Thread(new AddThread(), "Thread for adding elements"); 
     addThread.setPriority(6); 
     addThread.start(); 
     ActiveThreads.addThread(addThread); 
    } 

    public void add(T newElt) 
    { 
     synchronized(addQueue) 
     { 
      addQueue.add(newElt); 
     } 
    } 

    public T get(int index) 
    { 
     if(index > size) 
      throw new IndexOutOfBoundsException(); 

     Node<T> temp = head; 
     for(int i = 0; i < index; i++) 
     { 
      temp = temp.next; 
     } 
     return temp.data; 
    } 

    public Node<T> lookup(T element) 
    { 
     Node<T> temp = head; 
     for(int i = 0; i < size; i++) 
     { 
      if(temp.data == element) 
       return temp; 
      temp = temp.next; 
     } 
     throw new NoSuchElementException(); 
    } 

    public int size() 
    { 
     return size; 
    } 

    public boolean isEmpty() 
    { 
     return head == null; 
    } 

    public void delete(T element) 
    { 
     throw new UnsupportedOperationException("You must implement this method."); 
    } 

    public void replace(T oldElt, T newElt) 
    { 
     try 
     { 
      Node<T> temp = lookup(oldElt); 
      temp.data = newElt; 
     } 
     catch(NoSuchElementException e) 
     { 
      throw e; 
     } 
    } 

    public T getLargest() 
    { 
     return largest; 
    } 

    public T getSmallest() 
    { 
     return smallest; 
    } 

    public List<T> copy() 
    { 
     throw new UnsupportedOperationException("You must implement this method."); 
    } 

    public String toString() 
    { 
     StringBuffer ret = new StringBuffer(); 
     int i = 0; 
     for(T x : this) 
     { 
      System.out.println("Loop: " + i++); 
      ret.append(x + " "); 
     } 
     return ret.toString(); 
    } 

    public int compareTo(List<T> other) 
    { 
     throw new UnsupportedOperationException("You must implement this method."); 
    } 

    public ListIterator<T> iterator() 
    { 
     return new ListIterator<T>(head); 
    } 

    private class ListIterator<E> implements Iterator<E> 
    { 
     private ListIterator(Node<E> head) 
     { 
      cur = head; 
     } 

     public boolean hasNext() 
     { 
      if(cur == null) 
       return false; 

      System.out.println("Iterator: " + cur.data); 
      return cur.next == null; 
     } 

     @SuppressWarnings("unchecked") 
     public E next() 
     { 
      Node<E> temp = cur; 
      cur = cur.next; 
      return (E)temp.data; 
     } 

     public void remove() 
     { 
      throw new UnsupportedOperationException("You do NOT need to implement " + 
                "this method."); 
     } 

     private Node<E> cur; 
    } 

    private class AddThread implements Runnable 
    { 
     public AddThread() 
     { 

     } 

     public void run() 
     { 
      while(running) 
      { 
       synchronized(addQueue) 
       { 
        while(!addQueue.isEmpty()) 
        { 
         System.out.println("Going through queue"); 
         T temp = addQueue.poll(); 
         if(head == null) 
         { 
          head = new Node<T>(); 
          last = head; 
          head.data = temp; 
          largest = temp; 
          smallest = temp; 
         } 
         else 
         { 
          last.next = new Node<T>(); 
          last.next.data = temp; 
          last = last.next; 
          if(mComparator.compare(temp, largest) == 1) 
          { 
           largest = temp; 
          } 
          else if(mComparator.compare(temp, smallest) == -1) 
          { 
           smallest = temp; 
          } 
         } 
         ++size; 
        } 
       } 

       System.out.println("Pausing " + addThread); 
       synchronized(addThread) 
       { 
        pause(200); 
       } 
      } 
     } 

     private void pause(long time) 
     { 
      try 
      { 
       addThread.wait(time); 
      } 
      catch (InterruptedException e) 
      { 
       running = false; 
       addThread.notify(); 
      } 
     } 

     private volatile boolean running = true; 
    } 

    private Comparator<T> mComparator; 
    private Node<T> head, last; 
    private Thread addThread; 
    // replace this with your own created class later 
    private Queue<T> addQueue; 
    private int size; 
    private T largest, smallest; 
} 

ActiveThreads.java

public class ActiveThreads 
{ 
public ActiveThreads() 
{ 
    if(threads == null) 
    { 
     threads = new ArrayList<Thread>(); 
    } 

    if(threadsMonitor == null) 
    { 
     ThreadsMonitor monitor = new ThreadsMonitor(); 
     Thread thread = new Thread(monitor, "Active Threads Monitor"); 
     thread.setPriority(3); 
     thread.start(); 
     threadsMonitor = thread; 
    } 
} 

public static void stopMonitoring() 
{ 
    synchronized(threadsMonitor) 
    { 
     threadsMonitor.interrupt(); 
    } 
} 

public static void addThread(Thread t) 
{ 
    synchronized(threads) 
    { 
     threads.add(t); 
     System.out.println("Added thread: " + t); 
    } 
} 

public static void addThread(Runnable r, String s) 
{ 
    Thread t = new Thread(r, s); 
    t.start(); 
    addThread(t); 
} 

public static void stopThread(Thread t) 
{ 
    synchronized(threads) 
    { 
     for(int i = 0; i < threads.size(); i++) 
     { 
      if(threads.get(i) == t) 
      { 
       threads.get(i).interrupt(); 
       threads.remove(i); 
      } 
     } 
    } 
} 

public static void stopAllThreads() 
{ 
    boolean stopped = false; 
    while(stopped == false) 
    { 
     System.out.println("Stopping all threads"); 

     synchronized(threads) 
     { 
      System.out.println("Threads being stopped: " + threads.size()); 
      for(int i = 0; i < threads.size(); i++) 
      { 
       System.out.println("Stopping: " + threads.get(i)); 
       threads.get(i).interrupt(); 
      } 
      threads.clear(); 
      stopped = true; 
     } 
    } 
} 

private static ArrayList<Thread> threads = null; 
private static Thread threadsMonitor = null; 

private class ThreadsMonitor implements Runnable 
{ 
    public ThreadsMonitor() 
    { 

    } 

    public void run() 
    { 
     while(true) 
     { 
      synchronized(threads) 
      { 
       for(int i = 0; i < threads.size(); i++) 
       { 
        if(!threads.get(i).isAlive()) 
        { 
         threads.get(i).interrupt(); 
         threads.remove(i); 
        } 

        synchronized(threadsMonitor) 
        { 
         try 
         { 
          threadsMonitor.wait(5000); 
         } 
         catch(InterruptedException e) 
         { 
          threadsMonitor.interrupted(); 
          return; 
         } 
        } 
       } 
      } 
     } 
    } 
} 
} 

Answer 1

problema encontrado: Hubo un bloque sincronizado anidada que entró en una llamada a wait() en la parte interior del bloque, sin liberar el bloqueo en el ArrayList de Temas.

synchronized(threads) 
      { 
       for(int i = 0; i < threads.size(); i++) 
       { 
        if(!threads.get(i).isAlive()) 
        { 
         threads.get(i).interrupt(); 
         threads.remove(i); 
        } 
       } 
      } 

      synchronized(threadsMonitor) 
      { 
       try 
       { 
        threadsMonitor.wait(5000); 
       } 
       catch(InterruptedException e) 
       { 
        threadsMonitor.interrupted(); 
        return; 
       } 

era el problema, la eliminación de la sincronización a threadsMonitor dentro de la sincronización a las roscas corrige el problema, como se ve a continuación. (Me disculpo por problemas de formato, copiar + pegar es una bestia voluble)

  synchronized(threadsMonitor) 
      { 
       try 
       { 
        threadsMonitor.wait(5000); 
       } 
       catch(InterruptedException e) 
       { 
        threadsMonitor.interrupted(); 
        return; 
       } 
      } 

synchronized(threads) 
       { 
       for(int i = 0; i < threads.size(); i++) 
       { 
        if(!threads.get(i).isAlive()) 
        { 
         threads.get(i).interrupt(); 
         threads.remove(i); 
        } 
       } 
      } 
     } 

Gracias a Mike Q para el recurso multihilo. Todavía no he tenido tiempo de hacer mucho con eso, pero definitivamente voy a echar un vistazo y espero aprender mucho. También haré una pregunta por separado para un problema que estoy teniendo con esto aparte de detener los hilos.

Answer 2

Es difícil ver la imagen completa, pero algunos puntos generales y sugerencias incluso más generales.

Al intentar depurar, lo pusieron en el bucle while para tratar de forzar el método para funcionar con el tiempo el bucle, pero todo lo que sucede es que da salida a "Detener todas las discusiones" y luego no ver nada más

Esto es probablemente debido a que otro hilo está sentado en un bloque sychronized utilizando la misma cerradura y la prevención de su código se ejecute. La forma más fácil de averiguarlo es ejecutarlo en un depurador y pausar/interrumpir el programa cuando creas que puede estar bloqueado. Debería poder inspeccionar los hilos y verificar su estado. Busque el estado BLOQUEADO.

El while(!stopped) en stopAllThreads es redundante porque nunca puede repetirse.

En AddThread usted tiene esta

private boolean running = true; 

Cuando se utiliza un booleano como un indicador de parada (que creo que es lo que está tratando de lograr) y que la bandera de parada es recogido por otro hilo, pero asignadas por otro entonces DEBE hacerlo volatile. Existe un área completa de codificación multiproceso de Java que se ocupa de la visibilidad de los datos y volatile es una de las herramientas utilizadas para garantizar la corrección.

A menudo, los programas de trabajo "bien la mayor parte del tiempo" sin lógica multihilo "correcta". Pero ESTÁN rotas y es muy probable que se rompan en el momento más inoportuno (por lo general, tan pronto como un cliente se apodere de él). (Si usted recuerda una cosa en mi respuesta recuerda la frase anterior: D)

apalancamiento del paquete java.util.concurrent lo más que pueda. Aunque es muy importante comprender los fundamentos también, este paquete tiene muchas construcciones muy útiles diseñadas por personas muy inteligentes que resolverán muchos problemas comunes de simultaneidad.

Leer Java Concurrent In Practice. Para el tema que describe (potencialmente muy seco) explica muy bien los conceptos de una manera accesible con muchos ejemplos. Fue escrito por el mismo grupo que trabajó en el paquete java.util.concurrent.

Desde mi propia experiencia creo que la "visibilidad" de los datos en los programas concurrentes es el más importante y el menos comprendido parte de la programación Java hilo. Si puede darse cuenta de que estará bien en su camino. JCIP puede ayudarlo con esto.

Espero que lo anterior ayude y buena suerte!

EDIT: manchado otro problema en su código adicional con esta construcción

for(int i = 0; i < threads.size(); i++) 
{ 
    if(!threads.get(i).isAlive()) 
    { 
     threads.get(i).interrupt(); 
     threads.remove(i); 
    } 
} 

Haciendo un remove() dentro de una exploración de índice de una lista como esta no va a funcionar como se esperaba debido a que el remove() interfiere con la índices de otros elementos en la lista (todos los elementos subsecuentes se desplazan hacia abajo un índice).