本文共 10874 字，大约阅读时间需要 36 分钟。
本文介绍Java多线程运用知识点.
(1).继承java.lang.Thread,覆盖run().
(2).实现接口java.lang.Runnable,覆盖run().
(3)实现接口java.lang.Callable,覆盖call()此方法可以带返回值和抛出异常.
实现接口的线程有更大的灵活性,更符合面向对象分工的思想,将线程任务独立出来.如果线程任务有返回值,就采用实现Callable接口
1.thread.join() : 调用其他线程的join()方法,当前线程会等待thread执行完成后继续.
2.thread.setDaemon(true) : 设置thread线程为后台线程(必须在线程启动之前设置),当所有前台线程结束后,系统通知后台线程结束
3.Thread.sleep() : 阻塞当前线程一定的时间(ms).
4.Thread.yield() : 将当前线程设置为就绪.也就是让系统重新调度
5.设置获取线程优先级 : setPripority(int),getPripority().可以为1-10,10个等级.默认有三种常量
MAX-PRIORITY=10,MIN_PRIORITY=1,NORM_PRIORITY=5.
1.synchronized关键字.
使用synchronized可以同步语句块和同步方法.同步语句块需要指定一个加锁对象.对于实例方法,会给调用该方法的对象加锁,对于静态方法,会给类加锁.在解锁之前另一个调用那个对象(类)的方法的线程会被阻塞,直到解锁.当同步语句块或者方法结束后即解锁.
2.同步锁Lock.
一种更强大的线程同步机制-通过显示定义同步锁来实现同步.而且可以支持多个Condition对象以实现线程协作.
Condition提供方法:
await(),当前线程等待其他线程唤醒
signal(),唤醒此Lock对象上等待的一个线程.如果有多个,则被唤醒的线程是不确定的.
signalAll().唤醒此Lock对象上等待的所有线程.
在实现中,比较常用的是可重入锁ReentrantLock.
使用锁和条件的格式示例:
1: /** 2: * 同步锁示例 3: * @author 4: * 5: */ 6: public class LockDemo { 7: public static void main(String[] args) { 8: Acount acount=new Acount(); 9: new Thread(new WithdrawTask(acount)).start(); 10: new Thread(new DepositTask(acount)).start(); 11: } 12: } 13: class Acount{ 14: //定义锁 15: private final ReentrantLock lock=new ReentrantLock(); 16: //定义一个条件 17: private final Condition condition1=lock.newCondition(); 18: private int blance=0; 19: public int getBlance() { 20: return blance; 21: } 22: public void withdraw(int amount){ 23: lock.lock(); 24: try{ 25: while(blance
有于加锁机制,程序会发生多个线程互相等待对方释放锁的情况,这就是死锁.
在多线程中要考虑死锁的情况.为避免死锁可以通过设置资源访问优先级,或者调整线程推进顺序等方法解决.
死锁情况示例:
1: import java.util.concurrent.locks.ReentrantLock; 2: 3: public class DeadLockDemo { 4: public static void main(String[] args) { 5: A a=new A(); 6: B b=new B(); 7: a.setB(b); 8: b.setA(a); 9: new Thread(a).start(); 10: new Thread(b).start(); 11: } 12: } 13: class A implements Runnable{ 14: private ReentrantLock lock=new ReentrantLock(); 15: private B b; 16: public B getB() { 17: return b; 18: } 19: public void setB(B b) { 20: this.b = b; 21: } 22: public void fun(){ 23: lock.lock(); 24: System.out.println("A请求执行B的方法"); 25: b.fun(); 26: System.out.println("A方法执行"); 27: lock.unlock(); 28: } 29: public void run(){ 30: while(true){ 31: fun(); 32: } 33: } 34: } 35: class B implements Runnable{ 36: private ReentrantLock lock=new ReentrantLock(); 37: private A a ; 38: 39: public A getA() { 40: return a; 41: } 42: public void setA(A a) { 43: this.a = a; 44: } 45: public void fun(){ 46: lock.lock(); 47: System.out.println("B请求执行A的方法"); 48: a.fun(); 49: System.out.println("B方法执行"); 50: lock.unlock(); 51: } 52: public void run(){ 53: while(true){ 54: fun(); 55: } 56: } 57: }
BlockingQueue 方法以四种形式出现，对于不能立即满足但可能在将来某一时刻可以满足的操作，这四种形式的处理方式不同：第一种是抛出一个异常，第二种是返回一个特殊值（null 或 false，具体取决于操作），第三种是在操作可以成功前，无限期地阻塞当前线程，第四种是在放弃前只在给定的最大时间限制内阻塞。下表中总结了这些方法：
| 抛出异常 | 特殊值 | 阻塞 | 超时 | |
| 插入 | add((e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除 | remove() | poll() | take() | pool(time,unit) |
| 检查 | element() | peek() | _ | _ |
阻塞队列主要用于实现生产者-消费者队列.
生产者-消费者场景示例:
1: 2: /** 3: * 使用阻塞队列实现生产者消费者场景 4: * @author WeiCong 5: * 6: */ 7: public class ProducerAndConsumer { 8: 9: public static void main(String[] args) { 10: Buffer buffer=new MyBuffer(1); 11: ExecutorService executor = Executors.newFixedThreadPool(3); 12: executor.execute(new Producer(buffer)); 13: executor.execute(new Consumer(buffer)); 14: executor.shutdown(); 15: } 16: } 17: 18: class MyBuffer extends Buffer{ 19: public MyBuffer(int size){ 20: super(size); 21: } 22: public MyBuffer(){ 23: super(); 24: } 25: public Object produceCore() { 26: return new Random().nextInt(100)+1; 27: } 28: } 29: 30: class Producer implements Runnable{ 31: private Buffer buffer; 32: public Producer(Buffer buffer){ 33: this.buffer=buffer; 34: } 35: public void run() { 36: try { 37: while(true) { 38: Object obj=buffer.produceCore(); 39: System.out.println("生产:"+obj); 40: buffer.produce(obj); 41: Thread.sleep(new Random().nextInt(1000)+10); 42: } 43: } catch (InterruptedException ex) {ex.printStackTrace();} 44: } 45: public Object produce() { return null; } 46: 47: } 48: class Consumer implements Runnable{ 49: private Buffer buffer; 50: public Consumer(Buffer buffer){ 51: this.buffer=buffer; 52: } 53: public void run() { 54: try { 55: while(true) { 56: System.out.println("\t\t消费:"+buffer.consume()); 57: Thread.sleep(new Random().nextInt(1000)+10); 58: } 59: } catch (InterruptedException ex) {ex.printStackTrace();} 60: } 61: 62: } 63: abstract class Buffer { 64: private BlockingQueue queue=null; 65: public Buffer(Collection c){ 66: queue=new LinkedBlockingQueue (c); 67: } 68: public Buffer(int size){ 69: queue=new LinkedBlockingQueue (size); 70: } 71: public Buffer(){ 72: queue=new LinkedBlockingQueue (); 73: } 74: public T consume() throws InterruptedException { 75: T t=queue.take(); 76: return t; 77: } 78: public void produce(T obj) throws InterruptedException { 79: queue.put(obj); 80: } 81: public abstract T produceCore(); 82: } 83: 信号量用来限制同时访问资源的线程数.在访问资源前,必须从信号量获取许可.访问完成后将许可返回给信号量.允许设置公平策略(具体查看API)
java.util.concurrent.Semaphore.
+Semaphore(numberOfPermits: int) 创建指定数目许可的信号量.公平策略为false
+Semaphore(numberOfPermits: int,fair : boolean ) 创建带制定数目的许可和公平策略的信号量
+acquire(): void 获得信号量的许可,一直阻塞直到获取.
+release() : void 释放信号量许可
说明:只有一个许可的信号量可以实现线程互斥.
示例1:修改前面Acount的内部实现为使用信号量:
定义信号量:
1: //信号量 2: private final Semaphore semaphore =new Semaphore(1); 3: 将deposit设置为只允许一个线程访问:
1: public void deposit(int amount){ 2: // lock.lock(); 3: try{ 4: semaphore.acquire(); 5: blance+=amount; 6: System.out.println("存钱"+amount+",余额"+getBlance()+""); 7: // condition1.signalAll(); 8: } catch (InterruptedException e) { 9: e.printStackTrace(); 10: }finally{ 11: // lock.unlock(); 12: semaphore.release(); 13: } 14: } 使用线程组可以对一批线程进行分类管理.
java使用ThreadGroup表示线程组.创建线程组时可以指定父线程组和此线程组的名称.
通过Thread的构造方法可以为任务指定线程组.
如果线程执行过程中抛出了一个未处理的异常,JVM在结束该线程之前自动查找是否有对应的Thread.UncaughtExceptionHandler对象,如果有,则调用该对象的uncaughtException(Thread t,Throwable e)处理异常.
Thread类提供了静态方法可设置异常处理器.
ThreadGroup类实现了Thread.UncaughtExceptionHandler.所以每一个线程所属的线程组将会作为默认的处理器.
从JDK5开始java内建支持线程池.
通过Executors工厂可以差un关键不同的线程池:
| ExecutorService newCachedThreadPool() | 创建具有缓存功能的线程池 |
| ExecutorService newFixedThreadPool() | 创建一个可重用的,具有固定线程数的线程池 |
| ExecutorService newSingleThreadExecutor() | 创建一个单线程线程池 |
| ScheduledExecutorService newScheduledThreadPool() | 创建具有指定数目的线程池,可以在指定延迟后执行线程任务 |
| ScheduledExecutorService newSingleThreadScheduledExecutor() | 单个线程池,可延迟执行任务 |
| ExecutorService newWorkStealingPool(int parallelism) | (jdk8)创建持有足够多的线程的线程池来支持给定的并行级别,该方法能使用多个队列减少竞争. |
| ExecutorService newWorkStealingPool() | (jdk8)上一个线程池的简化版,自动根据CPU数目设置并行级别. |
ExecutorService代表尽快执行线程的线程池.
ScheduledExecutorService 代表可在指定延迟后或周期性地执行线程任务的线程池.通过相应的方法设置延迟.
线程池用完后应该调用shutdown()方法,调用此方法的线程池不在接收任务,并且启动关闭序列,但会将所有提交的任务执行完成.
使用shutdownNow()方法将会立即强制结束线程.
为从分利用多核CPU,多CPU的性能优势.我们可以将一个大任务分解成多个小任务放在多个CPU上执行,之后再合并执行结果.
ForkJoinPool就支持这种并行计算.
ForkJoinPool是ExecutorService的实现类,因此是一种特殊的线程池.
示例:
1: 2: /** 3: * java 8 增强的ForkJoinPool ,充分利用多CPU,多核CPU的优势,将一个任务分解成多个小任务并行执行来提高效率 4: */ 5: public class ForkJoinPoolTest { 6: public static void main(String[] args) throws Exception { 7: test2(); 8: test1(); 9: } 10: public static void test2() throws Exception { 11: int[] arr = new int[10000]; 12: for (int i = 0; i < arr.length; i++) { 13: arr[i] = i + 1; 14: } 15: long start = System.nanoTime();// 获取系统累加开始时间点 16: ForkJoinPool pool = new ForkJoinPool(); 17: Future future = pool.submit(new SumTask(arr, 0, arr.length-1)); 18: System.out.println(future.get()); 19: pool.shutdown(); 20: long end = System.nanoTime(); 21: long ms = TimeUnit.NANOSECONDS.toMillis(end - start);// 得到累加所用的时间 22: System.out.println("用时:" + ms + " ms"); 23: } 24: 25: public static void test1() throws InterruptedException { 26: ForkJoinPool pool = new ForkJoinPool(); 27: pool.submit(new PrintTask(0, 300)); 28: pool.awaitTermination(2, TimeUnit.SECONDS); 29: pool.shutdown(); 30: } 31: 32: } 33: /**有返回值的任务 34: * 计算数组1-10000的和分解为多个求相隔1000的小任务 */ 35: class SumTask extends RecursiveTask { 36: private static final int e = 1000; 37: private int[] arr; 38: private int start; 39: private int end; 40: public SumTask(int[] arr, int start, int end) { 41: this.start = start; 42: this.end = end; 43: this.arr = arr; 44: } 45: protected Integer compute() { 46: int sum = 0; 47: if (end - start < e) { 48: for (int i = start; i <=end; i++) { 49: sum += arr[i]; 50: } 51: } else { 52: int middle = (start + end) / 2; 53: SumTask left = new SumTask(arr, start, middle); 54: SumTask right = new SumTask(arr, middle+1, end); 55: left.fork(); 56: right.fork(); 57: return left.join() + right.join(); 58: } 59: return sum; 60: } 61: } 62: /**无返回值的任务 63: * 将输出0-300的任务分解为输出相隔为50的多个小任务 */ 64: class PrintTask extends RecursiveAction { 65: // 每个任务最多处理50个 66: private static final int THRESHOLD = 50; 67: private int start; 68: private int end; 69: 70: public PrintTask(int start, int end) { 71: this.start = start; 72: this.end = end; 73: } 74: protected void compute() { 75: if (end - start < THRESHOLD) { 76: for (int i = start; i <= end; i++) { 77: System.out.println(Thread.currentThread().getName() + "::" + i 78: + " "); 79: } 80: } else { 81: // 当数量多余50个时,任务分解 82: int middle = (start + end) / 2; 83: PrintTask left = new PrintTask(start, middle); 84: PrintTask right = new PrintTask(middle, end); 85: left.fork(); 86: right.fork(); 87: } 88: } 89: } JDK1.5之后,在java.util.concurrent包下供了大量支持高效并发访问的集合接口和实现类.
几乎对所有的传统集合都进行了包装,大致分为两类:
1.以Concurrent开头的集合类.代表了支持并发访问的集合.
2.以CopyOnWrite开头的集合类,采用底层赋值数组的方式来实现写操作.
读写锁,Java提供的同步互斥工具
Lock readLock() 获取一个可以被多个线程读的锁,排斥所有写操作.
Lock writeLock()获取一个写锁,排斥所有其他读/写操作.
转载地址：http://lvvqj.baihongyu.com/