帮助Java多线程

Question

好吧,这是一个双重目的的问题.我希望从中获得的主要内容是更多的多线程知识.在多线程方面,我是一个完全新手,这是我第一次尝试在多线程中做任何事情.我希望带走的另一件事是对一件作业的一些帮助,我正在变成一个更复杂的用于娱乐和学习的非常项目.在这个问题的第一部分,我将详细介绍我在家庭作业中一直在研究的主题的思路和方法.如果我正在做任何不好的做法,需要修复,无论如何,请让我知道,这样我就可以学习.我再一次对多线程几乎一无所知.

首先,我目前正在修读一门计算机科学课程,为此,我要做的是使用我已经学过的技术和数据结构的家庭作业,因此没有挑战性.为了尝试不完全厌倦我的头骨,我试图采取一个简单的项目(创建一个链接列表和一个排序的链接列表),并将其转换为一个多线程的法兰克计划.我想将添加新元素的方法放在一个单独的线程中并从队列中获取输入(对于无序列表不是完全必要的,但对于正确的排序更有用),对于有序列表,我想要一个单独的线程将巡视列表以确保所有元素都按顺序排列(我不允许更改的方法之一返回对整个节点的引用,这些节点具有完全公共数据).这些是我想要创造的两件主要事情,但绝对不是我想要尝试和弄清楚的唯一事情.这只是一个了解多线程的测试,所以我并不是特别为了实用性而设计这个项目.如果你想对什么是优秀的编程实践留下评论应该是什么线程,那么我们将非常感激.

作为我最近采取的一个步骤,通过阅读StackOverflow本身的帖子收集到的想法,我创建了一个主类,其中传递了Threads,因此它可以在程序结束时停止它们并清理所有内容.这是通过使用someThread.interrupt()并使用Thread的run()方法检查该异常来完成的.但是,我发现这个方法存在问题:大多数情况下,通过线程运行并调用interrupt()的循环实际上并不运行.以下是该方法的代码:

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;
        }
    }
}

在尝试调试它时,我放入while循环尝试强制该方法最终运行for循环,但所有发生的是输出"Stopping all threads"然后我再也看不到任何其他内容了.我不知道这是一种不好的编码方式,也不知道这段代码有什么问题.任何有关如何使这项工作有所帮助的人都会受到赞赏(如果您需要查看更多课程,请告诉我.我真的不知道您需要看什么,但我不想复制和粘贴几个整个java文件).

除了这个问题之外,我还确定在运行我的程序时,它将到达最后一行并尝试在线程通过要添加到列表的数字队列之前停止所有线程.在向队列添加数字后直接尝试通过toString方法输出列表时,这也是一个问题,因为还没有添加任何内容,因此不会向控制台输出任何内容.这是我的主要方法,它显示了事情发生的顺序(虽然我认为在处理线程时它不那么重要):

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();
}

最后,这里是我所做的自定义线程,它应该检查队列中的元素,然后将队列中的所有元素添加到我的实际列表中:

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;
  }

现在,我想要求的绝对最后一件事是一些基本的引物和指南,以及良好的多线程.我已经找到了很多教程,但有一些人们在教程中没有提及,以及大多数教程都只是马马虎虎.如果你知道任何好的教程,我会很感激他们的链接.如果有多线程的良好编码方法,那也是非常值得赞赏的.你认为任何类型的信息都可以帮助我围绕这个概念,并确保在实现它时,我以一种不易出错的方式做到这一点,人们不会因为坏事而对我大喊大叫实践.

编辑:为了回答matt b放入注释中的问题,addThread纯粹是对线程的引用,该线程查看队列中包含要添加到列表中的元素并将它们放入列表的队列.它是包含AddThread类的类内部的私有线程(聪明地使用名称,嗯？).线程的构建方式是在StackOverflow的另一个问题中发布的.这是一个示例(以及显示addThread的代码片段):

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

所有线程都以类似的方式创建.至于我发布的Thread应该做什么,我上面详细解释过,但我会尝试在这里进行一个概要.

AddThread类是一个内部类,它在构建我想要创建的链表类时创建.它查看一个私有队列,其中可能包含或不包含元素,当它执行时,它会将这些元素放入链接列表中.当调用链表的add(T newElt)方法时,队列也会被添加.

编辑编辑:为了充分披露,对于那些希望花时间的人,我将在这里发布两个相关的java文件的全部内容.

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

很难看到整体情况,但有一些一般性的观点,甚至更一般的建议.

在尝试调试它时,我放入while循环尝试强制该方法最终运行for循环,但所有发生的是它输出"Stopping all threads"然后我再也看不到任何其他东西

这可能是因为另一个Thread使用相同的锁定位于同步块中并阻止您的代码运行.最简单的方法是在调试器中运行它,并在您认为它可能被卡住时暂停/中断程序.您应该能够检查线程并检查其状态.寻找BLOCKED状态.

该while( !stopped )中stopAllThreads是多余的,因为它可以永远循环.

在AddThread你有这样的

private boolean running = true;

当使用布尔值作为停止标志(我认为这是你想要实现的)并且停止标志由一个线程轮询但由另一个线程设置然后你必须做它volatile.Java多线程编码的整个区域涉及数据可见性,volatile是用于确保正确性的工具之一.

在没有"正确"的多线程逻辑的情况下,程序通常在大多数情况下工作"很好".但它们很破碎,并且最有可能在最不方便的时候中断(通常只要客户掌握它)!(如果你记得我的回答中的一件事,请记住上一句话:D)

java.util.concurrent尽可能多地利用包裹.虽然理解基础知识也非常重要,但是这个包有许多非常有用的结构,由非常聪明的人设计,可以解决许多常见的并发问题.

阅读Java Concurrent In Practice.对于它描述的主题(可能非常干燥),它以可访问的方式很好地解释了这些概念,并提供了大量示例.它是由负责该java.util.concurrent程序包的同一组撰写的.

根据我自己的经验,我认为并发程序中数据的"可见性"是Java线程编程中最重要和最不被理解的部分.如果你能够完成任务,那么你将会很顺利. JCIP可以帮助您解决这个问题.

希望以上有所帮助,祝你好运!

编辑:发现此构造的附加代码中的另一个问题

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

在像这样的列表的索引扫描中执行remove()将无法按预期工作,因为remove()会干扰列表中其他项的索引(所有后续项目都会向下移动一个索引).

Answer 2

发现问题：有一个嵌套的同步块在该块的最里面部分调用了 wait()，而没有释放 Threads 的 ArrayList 上的锁。

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;
                }

是这个问题，删除线程同步内的threadsMonitor 同步可以修复该问题，如下所示。（对于格式问题，我深表歉意，复制+粘贴是一个善变的野兽）

            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 );
                    }
                }
            }
        }

感谢 Mike Q 提供的多线程资源。我还没有时间对它做太多事情，但我肯定会看一看并希望能学到很多东西。除了停止线程之外，我还将针对我遇到的问题提出一个单独的问题。