java.util.concurrent.atomic
一个小型工具包，支持单个变量上的无锁线程安全编程.
包含的类:这些类的相关操作都是原子性的
java.util.concurrent
线程池
//线程池的创建//创建一个线程池，该线程池重用固定数量的从共享无界队列中运行的线程。 //在任何时候，最多nThreads个线程同时处理任务。 //如果所有线程处于活动状态时又提交其他任务，则它们将在等待队列中直到有一个线程空闲可用。 //如果任何线程由于在关闭之前的执行期间发生故障而终止，如果需要执行后续任务，则新线程将占用它。 //池中的线程将一直存在，直到调用shutdown方法 。//public static ExecutorService newFixedThreadPool(int nThreads)ExecutorService es = Executors.newFixedThreadPool(3);//向线程池中添加任务es.execute(new Runnable(){});//可以添加多个任务//但是最多有三个线程去执行这些任务,即最多只有三个任务同时执行//线程池的关闭es.shutdown();//所有任务执行完后关闭es.shutdownNow(); //立即关闭，不管任务有没有执行完//创建缓存的线程池//创建一个根据需要创建新线程的线程池，但在可用时将重新使用以前构造的线程。 //这些池通常会提高执行许多短暂异步任务的程序的性能。 //调用execute将重用以前构造的线程（如果可用）。 //如果没有可用的线程，将创建一个新的线程并将其添加到该池中。 //未使达六十秒的线程将被终止并从缓存中删除。 //因此，长时间保持闲置的池将不会消耗任何资源public static ExecutorService newCachedThreadPool()//创建只有单个线程的线程池public static ExecutorService newSingleThreadExecutor() //如何实现线程死掉后重新启动，通过单一线程池，这样这个线程死掉后，线程池会自动重新创建一个线程来替代//调度线程池ScheduledExecutorService se = Executors.newScheduledThreadPool(3);se.schedule(new Runnable() //3秒后执行{ @Override public void run() { System.out.println("bombing"); }}, 3, TimeUnit.SECONDS);se.scheduleAtFixedRate(new Runnable()//6秒后第一次执行，以后每隔两秒执行一次{ @Override public void run() { System.out.println("bombing"); }}, 6, 2, TimeUnit.SECONDS);se.scheduleWithFixedDelay(new Runnable(){ @Override public void run() { System.out.println("bombing"); }}, 3, 2, TimeUnit.SECONDS);Callable 与 Future 类似 dot net 的async与await
使用Callable来执行任务，Future获取结果，在使用到结果的时候如果Callable还没结束，程序会暂停并等待结果，就跟await一样
public class CallableAndFuture{/** * @param args */public static void main(String[] args){ ExecutorService es = Executors.newSingleThreadExecutor(); Futurefuture = es.submit(new Callable () { @Override public String call() throws Exception { Thread.sleep(2000); return "Hello"; } }); System.out.println("等待结果"); try { System.out.println("返回结果:"+future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } es.shutdown(); ExecutorService es1 = Executors.newFixedThreadPool(10); CompletionService completionService = new ExecutorCompletionService<>(es1); for(int i=1;i<=10;i++) { final int sequence=i; completionService.submit(new Callable () { @Override public Integer call() throws Exception { Thread.sleep(new Random().nextInt(5000)); return sequence; } }); } for(int i=0;i<10;i++) { try { System.out.println(completionService.take().get()); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } es1.shutdown();}} java.util.concurrent.locks
主要的类与接口
//Lock接口void lock() //获得锁。 void lockInterruptibly() //获取锁定，除非当前线程是 interrupted 。 Condition newCondition() //返回一个新Condition绑定到该实例Lock实例。boolean tryLock() //只有在调用时才可以获得锁。 boolean tryLock(long time, TimeUnit unit) //如果在给定的等待时间内是空闲的，并且当前的线程尚未得到 interrupted，则获取该锁。 void unlock() //释放锁。 //使用规范//当在不同范围内发生锁定和解锁时，//必须注意确保在锁定时执行的所有代码由try-finally或try-catch保护，以确保在必要时释放锁定。Lock l = ...; l.lock(); try { } finally { l.unlock(); } //ReentrantLockstatic class Outputer{ Lock lock = new ReentrantLock(); public void output(String name) { lock.lock(); try { int len=name.length(); for(int i=0;i cache = new HashMap<>(); private ReentrantReadWriteLock rwl = new ReentrantReadWriteLock(); public static void main(String[] args) { } //这种思路不能很好的处理获取数据 //因为所有的获取数据都是互斥的，效率低 public synchronized Object getData1(String key) { Object value = cache.get(key); if(value==null) { value=new Random().nextInt(100); cache.put(key, value); } return value; } //锁降级：从写锁变成读锁； //锁升级：从读锁变成写锁。 //读锁是可以被多线程共享的，写锁是单线程独占的。 //也就是说写锁的并发限制比读锁高。 /** * 这种方法使用读写锁来处理缓存，同时读的时候不互斥， *当有一个去修改的时候，转换为写锁 * @param key * @return value */ public Object getData2(String key) { rwl.readLock().lock(); Object value = null; try { value=cache.get(key); if(value==null)//缓存中没有数据，去获取 { rwl.readLock().unlock();//释放读锁 rwl.writeLock().lock(); //加写锁 try { value=cache.get(key);//为了防止多次写入 if(value==null)//为了防止多次写入 { value=new Random().nextInt(100); cache.put(key, value); } rwl.readLock().lock(); } finally { // TODO: handle finally clause rwl.writeLock().unlock(); } } } finally { rwl.readLock().unlock(); } return value; }}/* 关于读写锁升级和降级的限制 1.因为同一个线程中，在没有释放读锁的情况下，就去申请写锁，这属于锁升级，ReentrantReadWriteLock是不支持的。 2.从写锁降级成读锁，并不会自动释放当前线程获取的写锁，仍然需要显示的释放，否则别的线程永远也获取不到写锁。*/ Condition 线程通信
//利用condition进行线程间的通信 static class Business { private ReentrantLock lock = new ReentrantLock(); private Condition condition1 = lock.newCondition(); private Condition condition2 = lock.newCondition(); private Condition condition3 = lock.newCondition(); private int flag=1; public void sub(int i) { lock.lock(); try { while(flag!=1) { try { //this.wait(); condition1.await(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int j = 1; j <= 10; j++) { System.out.println("sub thread sequence of " + j+",loop of "+i); } flag=2; condition2.signal(); } finally { // TODO: handle finally clause lock.unlock(); } } public void sub2(int i) { lock.lock(); try { while(flag!=2) { try { //this.wait(); condition2.await(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int j = 1; j <= 20; j++) { System.out.println("sub2 thread sequence of " + j+",loop of "+i); } flag=3; condition3.signal(); } finally { // TODO: handle finally clause lock.unlock(); } } public void main(int i) { lock.lock(); try { while(flag!=3) { try { //this.wait(); condition3.await(); } catch (InterruptedException e) { e.printStackTrace(); } } for (int j = 1; j <= 100; j++) { System.out.println("main thread sequence of " + j+",loop of "+i); } flag=1; //this.notify(); condition1.signal(); } finally { // TODO: handle finally clause lock.unlock(); } } }