Java之 java.util.concurrent 包之概述

程序员文章站 2022-05-12 11:30:13

...

[size=medium][b]一、概述：java.util.concurrent （Java 并发编程工具包）[/b][/size] [url=http://tutorials.jenkov.com/java-util-concurrent/index.html]Ref- jenkov[/url]

从 JDK1.5 开始新增了一个包：[color=green][b]java.util.concurrent[/b][/color]，这个包包含了一组Java类，它们使多线程开发变得容易。在新增这个包之前，你需要自己写这些工具类。

在本文中，我（用JDK1.6）将与你介绍一下这个包（[color=green][b]java.util.concurrent[/b][/color]）里面的类，挨个介绍。
本文[b]不会[/b]涉及并讨论【Java并发编程】的原理。如果你对此感兴趣，请看这里：[url=http://tutorials.jenkov.com/java-concurrency/index.html]Ref- jenkov[/url]

[size=medium][b]1、 java.util.concurrent.atomic.AtomicInteger[/b][/size]

在Java语言中，++i 和 i++ 操作并不是线程安全的，在多线程并发情况下，不可避免的会用到synchronized关键字。而AtomicInteger则提供了一种线程安全的加减操作。


/**
来看看AtomicInteger提供的接口。

- public final int get()                    //获取当前的值
- public final int getAndIncrement()        //获取当前的值，并自增
- public final int getAndDecrement()        //获取当前的值，并自减
- public final int getAndAdd(int delta)     //获取当前的值，并加上一个（正负）值
- public final int getAndSet(int newValue)  //取当前的值，并设置新的值

*/



import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockedQueueTest {  

    public static void main(String[] args) throws Exception { 
        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024);  

        Producer producer = new Producer(queue);  
        Consumer consumer = new Consumer(queue);  

        new Thread(producer).start();  
        new Thread(consumer).start();  

        System.out.println("done!");

    }  
}  

class Producer implements Runnable{
    protected BlockingQueue<String> queue ;
    public Producer(BlockingQueue<String> queue) {  
        this.queue = queue;  
    }
    public void run() {  
        try {  
            queue.put("1");  
            Thread.sleep(1000);  
            queue.put("2");  
            Thread.sleep(1000);  
            queue.put("3");  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}  


class Consumer implements Runnable{
    protected BlockingQueue<String> queue;
    public Consumer(BlockingQueue<String> queue) {  
        this.queue = queue;  
    }
    public void run() {  
        try {  
            System.out.println(queue.take());  
            System.out.println(queue.take());  
            System.out.println(queue.take());  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

[size=medium][b]3、java.util.concurrent.ConcurrentMap 接口[/b][/size] [url=http://tutorials.jenkov.com/java-util-concurrent/concurrentmap.html]Ref- jenkov[/url]

[color=green]java.util.concurrent.ConcurrentMap[/color] 接口 extends [color=green]java.util.Map[/color] 接口，具有处理并发 put 和 get 的能力。

[b]实现类：java.util.concurrent.ConcurrentHashMap[/b]
ConcurrentHashMap 与 java.util.HashTable 比较相似，但性能更高：读取数据时不锁Map；写数据时不会锁整个Map，只锁写的那部分。

另一个不同是：ConcurrentHashMap 不会抛 ConcurrentModificationException（当使用iterator 且 Map 数据发生改变时）。因为 Iterator 被设计为只允许一个线程使用。

例子：


      ConcurrentMap concurrentMap = new ConcurrentHashMap();
      concurrentMap.put("key", "value");
      Object value = concurrentMap.get("key");

[size=medium][b]4、Executors 框架 [/b][/size] [url=http://www.journaldev.com/1162/java-multithreading-concurrency-interview-questions-answers]Ref- journaldev[/url]

JDK1.5以前，创建过多的线程可能会引起内存耗尽。所以使用线程池（ThreadPool）是一个很好的解决方案。

Executors 框架，可以创建线程池线程池（ThreadPool），并提供了对异步线程的调用（invocation），调度（scheduling），执行（execution）的功能。
[color=gray]
问题：如何重复使用一个线程？
解决：线程对象一旦运行结束，不可以再被从新执行。但使用阻塞消息队列，可以不断接收和处理 Runnable 或 Callable 对象。
[/color]

[url=http://tutorials.jenkov.com/java-util-concurrent/executorservice.html][b]java.util.concurrent.ExecutorService[/b][/url]

java.util.concurrent.ExecutorService 接口是一个线程池（Thread Pool），可以异步执行线程。



import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.junit.Test;


public class TestExecutors {

	static class MyRunnable implements Runnable{
		@Override
		public void run() {
			System.out.println("Run, run, run!");
		}

	}

	static class MyCallable implements Callable<String>{
		@Override
		public String call() throws Exception {
			System.out.println("Call, call, call!");
			return null;
		}

	}

	static class MyBlockedRunnale implements Runnable{
		@Override
		public void run() {
			try {
				for(int i = 1; i <= 5; i++){
					System.out.println("Sleeping " + i + "s.");
					Thread.sleep(1000);

				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}




    /**
     *  test01_createServices
     *============================================================
     *  This ExecutorService contains the follow 5 methods 
     *  for passing it task for execution:
     *  
     * 
     *      - execute(Runnable)
     *      
     *      - submit(Runnable)
     *      - submit(Callable)
     *      
     *      - invokeAny(...)
     *      - invokeAll(...)
     *      
     *      
     */
	public void test01_createServices() throws Exception{

	    ExecutorService singlService    = Executors.newSingleThreadExecutor();
	    ExecutorService fixedService    = Executors.newFixedThreadPool(10);
	    ExecutorService schduService    = Executors.newScheduledThreadPool(10);
	    ExecutorService cacheService    = Executors.newCachedThreadPool();

	    singlService.shutdown();
	    fixedService.shutdownNow();
	    schduService.awaitTermination(10, TimeUnit.DAYS);
	    cacheService.awaitTermination(10, TimeUnit.SECONDS);
	}


    /**
     *  test02_Runnable_ExeuteSubmit
     *  ===========================================================
     */
    public void test02_Runnable_ExeuteSubmit() throws Exception{

    	ExecutorService singleService    = Executors.newSingleThreadExecutor();

    	/*
	     * service.execute() - void
	     */
	    singleService.execute(new MyRunnable());

	    /*
	     * service.submit(Runnable) - Future
	     *  - future.get(): this method is blocked, until result is returned.
	     *  - returns null if the task has finished correctly - for Runnable.
	     */
	    Future<?> future =  singleService.submit(new MyRunnable());
	    future.get(); 

    }


    /**
     *  test03_Callable_Submit
     *  ===========================================================
     */
	public void test03_Callable_Submit() throws Exception{  

		ExecutorService singleService    = Executors.newSingleThreadExecutor();

		/*
		 * service.submit(Callable) - Future
		 * - future.get(): this method is blocked, until result is returned.
		 * - returns result if successful completion.
		 */
		Future<String> future = singleService.submit(new MyCallable());
		System.out.println("future.get() = " + future.get());

	}



    /**
     * test04_invokeAny()
     * ================================================================
        The invokeAny() method takes a collection of Callable objects

        Invoking this method does not return a Future, 
        but returns the result of one of the Callable objects. 

        You have no guarantee about which of the Callable's results you get. 
        Just one of the ones that finish.

        If one of the tasks complete (or throws an exception), 
        the rest of the Callable's are cancelled.
     */
	public void test04_invokeAny() throws Exception{  

		ExecutorService cacheService    = Executors.newCachedThreadPool();

		Set<Callable<String>> callables = new HashSet<Callable<String>>();
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 1";
	        }
	    });
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 2";
	        }
	    });
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 3";
	        }
	    });

	    String result = cacheService.invokeAny(callables);
	    System.out.println("result = " + result);

	}


    /**
     * test05_invokeAll
     ===========================================================================
     The invokeAll() method invokes all of the Callable objects you pass to it 
     in the collection passed as parameter. 

     The invokeAll() returns a list of Future objects via which you can obtain 
     the results of the executions of each Callable.


     Keep in mind that a task might finish due to an exception, 
     so it may not have "succeeded". 

     There is no way on a Future to tell the difference.

     */
	public void test05_invokeAll() throws Exception{
		ExecutorService cacheService    = Executors.newCachedThreadPool();

		Set<Callable<String>> callables = new HashSet<Callable<String>>();
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 1";
	        }
	    });
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 2";
	        }
	    });
	    callables.add(new Callable<String>() {
	        public String call() throws Exception {
	            return "Task 3";
	        }
	    });

	    List<Future<String>> futures = cacheService.invokeAll(callables);
	    for(Future<String> future : futures){
	        System.out.println("future.get = " + future.get());
	    }

	}



	public void test06_Shutdown_AwaitTermination() throws Exception{

		ExecutorService cacheService    = Executors.newCachedThreadPool();

		/*
	     * This method does not wait for previously 
	     * submitted tasks to complete execution,
	     * and it will no longer accept new tasks.
	     * 
	     * The ExecutorService will shut down immediately.
		 */
		cacheService.shutdown();


		/*
		 * Blocks until: 
		 * 
		 * - all tasks have completed execution after a shutdown request, or 
		 * - the timeout occurs, or 
		 * - the current thread is interrupted, 
		 * 
		 * whichever happens first.
		 */
		cacheService.awaitTermination(50, TimeUnit.MINUTES);

    }





	@Test
	public void testShutdown(){
		ExecutorService cacheService    = Executors.newCachedThreadPool();
		cacheService.execute(new MyBlockedRunnale());
		cacheService.shutdown();
		System.out.println("shut down...");
	}

	@Test
	public void testAwaitTermination(){
		ExecutorService cacheService    = Executors.newCachedThreadPool();
		cacheService.execute(new MyBlockedRunnale());
		try {
			cacheService.awaitTermination(50, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("shut down...");

	}

	/**
	 * Best Practice
	 */
	@Test
	public void testShutdownAndAwaitTermination(){
		ExecutorService cacheService    = Executors.newCachedThreadPool();
		cacheService.execute(new MyBlockedRunnale());
		try {
			cacheService.shutdown();
			cacheService.awaitTermination(50, TimeUnit.MINUTES);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("shut down...");

	}

}

[size=medium][b]5、java.util.concurrent.locks 包 [/b][/size]

java.util.concurrent.locks 包提供了线性同步机制，跟 synchronized 关键字一样，但是功能更多。

[b]5.1 java.util.concurrent.locks.Lock 接口[/b] [url=http://tutorials.jenkov.com/java-util-concurrent/lock.html]Ref- jenkov[/url]
方法概述：

- lock()
- lockInterruptibly()
- tryLock()
- tryLock(long timeout, TimeUnit timeUnit)
- unlock()

实现类：
java.util.concurrent.locks.ReentrantLock

[b]5.2 java.util.concurrent.locks.ReadWriteLock 接口[/b] [url=http://tutorials.jenkov.com/java-util-concurrent/readwritelock.html]Ref- jenkov[/url]

提供了更高级的锁机制：可以有多个线程同时读，但只能有一个线程写。

Read Lock
取得读锁的条件：
- 没有线程在（或已经）获取写锁。

Write Lock
取得写锁的条件：
- 没有线程在读
- 没有线程在写

实现类：
java.util.concurrent.locks.ReentrantReadWriteLock

例子：




import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class TestConcurrent_Lock {

    public void test(){

        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();


        readWriteLock.readLock().lock();

            // multiple readers can enter this section,
            // if not locked for writing, and no writers waiting
            // to lock for writing.

        readWriteLock.readLock().unlock();


        readWriteLock.writeLock().lock();

            // only one writer can enter this section,
            // and only if no threads are currently reading.

        readWriteLock.writeLock().unlock();
    }

}

/*


readLock.lock();

This means that if any other thread is writing (i.e. holds a write lock) 
then stop here until no other thread is writing.


writeLock.lock();

This means that if any other thread is reading or writing, 
then stop here and wait until no other thread is reading or writing.


*/

java.util.concurrent包之Execuotor系列文章

[url=http://lixh1986.iteye.com/blog/2341898]00_Java之 java.util.concurrent 包之概述[/url]

[url=http://lixh1986.iteye.com/blog/2360304]01_Java之java.util.concurrent包之Executor与ExecutorService[/url]

[url=http://lixh1986.iteye.com/blog/2360306]02_Java之 java.util.concurrent 包之ExecutorService之submit () 之 Future[/url]

[url=http://lixh1986.iteye.com/blog/2351367]03_Java之多线程之Callable与Future[/url]

[url=http://lixh1986.iteye.com/blog/2351294]04_Java之多线程之Lock[/url]

转载请注明，
原文出处：http://lixh1986.iteye.com/blog/2341898

-

上一篇： Hashtable源码探讨（基于JDK1.8）

下一篇：这么回答面试官之--ConcurrentHashMap如何put？

Java之 java.util.concurrent 包之概述

Java缩略图生成库之Thumbnailator应用说明

hadoop入门之通过java代码实现将本地文件上传到hadoop的文件系统

Java之注解详解

java实战技巧之if-else代码优化技巧大全

Java面试题解析之判断以及防止SQL注入

死磕 java同步系列之Phaser源码解析

Java 基础篇之异常

java基础(一)-----java的三大特性之封装

Python标准库之随机数 (math包、random包)介绍

深入探索Java设计模式（三）之装饰器模式