java.util.concurrent
- Java's low-level concurrency primitives
synchronized/volatile/wait/notify/notifyall
常见问题:deadlock 死锁,thread starvation 线程饥饿, race conditiond 竞争条件等。
synchronized:performance issue.
- java.util.concurrent.atomic subpackage contains low-level utility classes that support lock-free thread-safe programming on single variables.
- java.util.concurrent.locks subpackage contains low-level utility types for locking and waiting for conditions.
compare-and-swap (CAS) - cas更轻量 (java.util.concurrent.locks.reentrantLock) ,reentrantLock提供了更多锁的选择。
Executor Framework
executor的优点,在不改变代码的前提下,改变程序对task的调度。
executor的缺点,不能取消正在运行中的task,不能关闭executor或决定一个异步程序是否结束。
常用方法:
ExecutorService newCachedThreadPool()
创建线程池时创建所需的thread,在之前构建的thread可用时会重用,60秒内没有使用的线程会被终止并从缓存中移除。可以改善short-lived异步程序的性能。
ExecutorService newSingleThreadExecutor()
使用single worker thread和 unbounded queue创建executor —— tasks加入queue中依次执行 (依次执行的task不超过一个)。如果thread如果在关闭前因为执行失败终止,新的thread会被创建并替代它的位置。
ExecutorService newFixedThreadPool(int nThreads)
在共享的unbounded queue中重用固定数量的线程,最多可以用nThreads共享task。如果额外的task提交时,所有的线程都被占用,当前task会在queue内等待,知道有可使用的线程。如果任一thread在关闭前执行失败,如果有task需要被执行,新的线程会被创建并取代失败线程的位置。线程池在executor随着shut down而结束。
public class Server { static Executor pool = Executors.newFixedThreadPool(5); public static void main(String[] args) throws IOException { ServerSocket socket = new ServerSocket(9000); while(true){ final Socket s = socket.accept(); Runnable r = new Runnable(){ @Override public void run(){ doWork(s); } }; pool.execute(r); } } public static void doWork(Socket s){ } }
当前callable无法在5秒内完成时,get()将返回TimeoutException。
public class ReadWebPage { public static void main(String[] args) { if(args.length != 1){ System.out.println("usage: java ReadWebPage url"); return; } ExecutorService executor = Executors.newSingleThreadExecutor(); Callable<List<String>> callable; callable = new Callable<List<String>>(){ @Override public List<String> call() throws Exception { List<String> lines = new ArrayList<>(); URL url = new URL(args[0]); HttpURLConnection con; con = (HttpURLConnection) url.openConnection(); InputStreamReader isr; isr = new InputStreamReader(con.getInputStream()); BufferedReader br; br = new BufferedReader(isr); String line; while((line = br.readLine()) != null){ lines.add(line); } return lines; } }; Future<List<String>> future = executor.submit(callable); try{ List<String> lines = future.get(5, TimeUnit.SECONDS); for(String line: lines){ System.out.println(line); } }catch (Exception ee){ System.err.println("Callable through exception: " + ee.getMessage()); } executor.shutdown(); } }