多线程集合队列等等

女爷i 2022-07-11 11:07 148阅读 0赞

对于许多线程问题，可以通过使用一个或多个队列以优雅且安全的方式将其形式化。当试图向队列添加元素而队列已满，或是想从队列移出元素而队列为空的时候， 阻塞队列（ blocking queue）导致线程阻塞。在协调多个线程之间的合作时，阻塞队列是一个有用的工具。工作者线程可以周期性地将中间结果存储在阻塞队列中。其他的工作者线程移出中间结果并进一步加以修改。队列会自动地平衡负载。

add 添加一个元素 如果队列满，则抛出IllegalStateException异常  
element 返回队列的头元素 如果队列空，抛出NoSuchElementException异常  
offer 添加一个元素并返回true 如果队列满,返回false  
peek 返回队列的头元素 如果队列空，则返回null  
poll 移出并返回队列的头元素 如果队列空，则返回null  
put 添加一个元素 如果队列满，则阻塞  
remove 移出并返回头元素 如果队列空，则抛出NoSuchElementException异常  
take 移出并返回头元素 如果队列空，则阻塞  
  
当试图向满的队列中添加或从空的队列中移出元素时， add、 remove和element操作抛出异常。一定要使用offer、 poll和peek方法作为替代。向这些队列中插入null值是非法的。  
默认情况下，LinkedBlockingQueue的容量是没有上边界的，但是，也可以选择指定最大容量。 LinkedBlockingDeque是一个双端的版本。ArrayBlockingQueue在构造时需要指定容量，并且有一个可选的参数来指定是否需要公平性。  
PriorityBlockingQueue是一个带优先级的队列，而不是先进先出队列。元素按照它们的优先级顺序被移出。该队列是没有容量上限，但是，如果队列是空的，取元素的操作会阻塞。注意，不需要显式的线程同步。我们使用队列数据结构作为一种同步机制。  
  
如果多线程要并发地修改一个数据结构，例如散列表，那么很容易会破坏这个数据结构。可以通过提供锁来保护共享数据结构，但是选择线程安全的实现作为替代可能更容易些。阻塞队列就是线程安全的集合。  
java.util.concurrent包提供了映像、有序集和队列的高效实现： ConcurrentHashMap、ConcurrentSkipListMap、 ConcurrentSkipListSet和ConcurrentLinkedQueue。这些集合使用复杂的算法，通过允许并发地访问数据结构的不同部分来使竞争极小化。与大多数集合不同， size方法不必在常量时间内操作。确定这样的集合当前的大小通常需要遍历。  
CopyOnWriteArrayList和CopyOnWriteArraySet是线程安全的集合。

List<T> list =  Collections.synchronizedList(new ArrayList<T>());
    Map<K, V> map = Collections.synchronizedMap(new HashMap<K, V>());

结果集合的方法使用锁加以保护，提供了线程的安全访问。应该确保没有任何线程通过原始的非同步方法访问数据结构。最便利的方法是确保不保存任何指向原始对象的引用，简单地构造一个集合并立即传递给包装器。最好使用java.util.concurrent包中定义的集合，不使用同步包装器中的。

Runnable封装一个异步运行的任务，可以把它想像成为一个没有参数和返回值的异步方法。Callable与Runnable类似，但是有返回值。 Callable接口是一个参数化的类型，只有一个方法call。

interface Callable<V>{
          V call() throws Exception;
    }

Future保存异步计算的结果。可以启动一个计算，将Future对象交给某个线程，然后忘掉它。 Future对象的所有者在结果计算好之后就可以获得它。

Future<V>{
     V get();
     v get(time,Timeunit);
     void cancel(boolean);
     boolean isCancelled();
     boolean isDone();
    }

第一个get方法的调用被阻塞，直到计算完成。如果在计算完成之前，第二个方法的调用超时，抛出一个TimeoutException异常。如果运行该计算的线程被中断，两个方法都将抛出InterruptedException。如果计算已经完成，那么get方法立即返回.

如果计算还在进行，isDone方法返回false；如果完成了，则返回true。可以用cancel方法取消该计算。如果计算还没有开始，它被取消且不再开始。如果计算处于运行之中，那么如果boolean参数为true，它就被中断。  
  
FutureTask包装器是一种非常便利的机制，可将Callable转换成Future和Runnable，它同时实现二者的接口。

Callable<V> a = new......;
    FutureTask<v> b = new FutureTask<V>(a);
    Thread c = new Thread(b);这里是runnable
    c.start();
    .....
    V d = b.get();//这里是future

构建一个新的线程是有一定代价的，因为涉及与操作系统的交互。如果程序中创建了大量的生命期很短的线程，应该使用线程池（threadpool）。一个线程池中包含许多准备运行的空闲线程。将Runnable对象交给线程池，就会有一个线程调用run方法。当run方法退出时，线程不会死亡，而是在池中准备为下一个请求提供服务。  
另一个使用线程池的理由是减少并发线程的数目。创建大量线程会大大降低性能甚至使虚拟机崩溃。如果有一个会创建许多线程的算法，应该使用一个线程数“固定的”线程池以限制并发线程的总数。

执行器（ Executor）类有许多静态工厂方法用来构建线程池  
newCachedThreadPool 必要时创建新线程；空闲线程会被保留60秒  
newFixedThreadPool 该池包含固定数量的线程；空闲线程会一直被保留  
newSingleThreadExecutor 只有一个线程的“池”，该线程顺序执行每一个提交的任务  
newScheduledThreadPool 用于预定执行而构建的固定线程池，替代java.util.Timer  
newSingleThreadScheduledExecutor 用于预定执行而构建的单线程“池

newCachedThreadPool方法构建了一个线程池，对于每个任务，如果有空闲线程可用，立即让它执行任务，如果没有可用的空闲线程，则创建一个新线程。 newFixedThreadPool方法构建一个具有固定大小的线程池。如果提交的任务数多于空闲的线程数，那么把得不到服务的任务放置到队列中。当其他任务完成以后再运行它们。 newSingleThreadExecutor是一个退化了的大小为1的线程池：由一个线程执行提交的任务，一个接着一个。这 3个方法返回实现了ExecutorService接口的ThreadPoolExecutor类的对象。

可用下面的方法之一将一个Runnable对象或Callable对象提交给ExecutorService：

Future<?> submit(runnable); Future<T> submit(runnable,T result); Future<T> submit(Callable);

该池会在方便的时候尽早执行提交的任务。调用submit时，会得到一个Future对象，可用来查询该任务的状态。  
第一个submit方法返回一个奇怪样子的Future<?>。可以使用这样一个对象来调用isDone、cancel或isCancelled。但是， get方法在完成的时候只是简单地返回null。  
第二个的Submit也提交一个Runnable，并且Future的get方法在完成的时候返回指定的result对象。  
第三个的Submit提交一个Callable，并且返回的Future对象将在计算结果准备好的时候得到它。  
 当用完一个线程池的时候，调用shutdown。该方法启动该池的关闭序列。被关闭的执行器不再接受新的任务。当所有任务都完成以后，线程池中的线程死亡。另一种方法是调用shutdownNow。该池取消尚未开始的所有任务并试图中断正在运行的线程。

下面总结了在使用连接池时应该做的事：  
1）调用Executors类中静态的方法newCachedThreadPool或newFixedThreadPool。  
2）调用submit提交Runnable或Callable对象。  
3）如果想要取消一个任务，或如果提交Callable对象，那就要保存好返回的Future对象。  
4）当不再提交任何任务时，调用 shutdown.

ExecutorService pool = Executors.newCachedThreadPool(); Future<V> a = pool.submit(new Runnable()); Future<V> b = pool.submit(new Runnable(),5); Future<V> c = pool.submit(new Callable()); pool.shutdown(); ((ThreadPoolExecutor)pool).getLargestPoolSize();//返回线程池在该执行器生命周期中的最大尺寸。

ScheduledExecutorService接口具有为预定执行（ Scheduled Execution）或重复执行任务而设计的方法。它是一种允许使用线程池机制的java.util.Timer的泛化。Executors类的newScheduledThreadPool和newSingleThreadScheduledExecutor方法将返回实现了  
ScheduledExecutorService接口的对象。

可以预定Runnable或Callable在初始的延迟之后只运行一次。也可以预定一个Runnable对象周期性地运行。  
  
invokeAll方法提交所有对象到一个Callable对象的集合中，并返回一个Future对象的列表，代表所有任务的解决方案。当计算结果可获得时，可以像下面这样对结果进行处理：

List<Callable<V>>a=...;
    List<Future<V>>list = pool.invokeAll(a);
    for(){
        xxx.get();
    }

这个方法的缺点是如果第一个任务恰巧花去了很多时间，则可能不得不进行等待。以结果按可获得的顺序保存起来更有实际意义。可以用ExecutorCompletionService来进行排列. 用常规的方法获得一个执行器。然后，构建一个ExecutorCompletionService，提交任务给完成服务（ completion service）。该服务管理Future对象的阻塞队列，其中包含已经提交的任务的执行结果（当这些结果成为可用时）。  
  
同步器

CyclicBarrier允许线程集等待直至其中预定数目的线程到达一个公共障栅（ barrier），然后可以选择执行一个处理障栅的动作，

什么时候使用：当大量的线程需要在它们的结果可用之前完成时。  
  
CountDownLatch允许线程集等待直到计数器减为0  
  
使用：当一个或多个线程需要等待直到指定数目的事件发生  
  
Exchanger允许两个线程在要交换的对象准备好时交换对象  
  
使用：当两个线程工作在同一数据结构的两个实例上的时候，一个向实例添加数据而另一个从实例清除数据  
  
Semaphore允许线程集等待直到被允许继续运行为止  
  
使用：限制访问资源的线程总数。如果许可数是1，常常阻塞线程直到另一个线程给出许可为止  
  
SynchronousQueue允许一个线程把对象交给另一个线程  
使用：在没有显式同步的情况下，当两个线程准备好将一个对象从一个线程传递到另一个时

这些机制具有为线程之间的共用集结点模式（ common rendezvous patterns）提供的“预置功能”（canned functionality）。如果有一个相互合作的线程集满足这些行为模式之一，那么应该直接重用合适的库类而不要试图提供手工的锁与条件的集合。