线程有有序性和可见性
多个线程之间是不能直接传递数据交互的,它们之间的交互只能通过共享变量来实现。
在多个线程之间共享类的一个对象,这个对象是被创建在主内存(堆内存)中,每个线程都有自己的工作内存(线程栈),工作内存存储了主内存对象的一个副本,当线程操作对象时,首先从主内存复制对象到工作内存中,然后执行代码改变了值,最后用工作内存刷新主内存。
当一个对象在多个内存中都存在副本时,如果一个内存修改了共享变量,其它线程也应该能够看到被修改后的值,此为可见性。
多个线程执行时,CPU对线程的调度是随机的,我们不知道当前程序被执行到哪步就切换到了下一个线程。为保证线程有序执行此为有序性。
使用synchronized修饰的方法或者代码块可以看成是一个原子操作。
每个锁对象(JLS中叫monitor)都有两个队列,一个是就绪队列,一个是阻塞队列,就绪队列存储了将要获得锁的线程,阻塞队列存储了被阻塞的线程,当一个线程被唤醒(notify)后,才会进入到就绪队列,等待CPU的调度,反之,当一个线程被wait后,就会进入阻塞队列,等待下一次被唤醒。看我们的例子,当第一个线程执行输出方法时,获得同步锁,执行输出方法,恰好此时第二个线程也要执行输出方法,但发现同步锁没有被释放,第二个线程就会进入就绪队列,等待锁被释放。一个线程执行互斥代码过程如下:
1. 获得同步锁;
2. 清空工作内存;
3. 从主内存拷贝对象副本到工作内存;
4. 执行代码(计算或者输出等);
5. 刷新主内存数据;
6. 释放同步锁。
所以,synchronized既保证了多线程的并发有序性,又保证了多线程的内存可见性。
volatile是第二种Java多线程同步的机制,一个变量可以被volatile修饰,在这种情况下内存模型(主内存和线程工作内存)确保所有线程可以看到一致的变量值。
JAVA线程创建方式
1、继承Thread类创建线程类
继承Thread类并重写该类的run方法,该run方法代表了线程要完成的任务。
2、通过Runnable接口创建线程类
实现runnable接口,重写该接口的run()方法,该run()方法的方法体同样是该线程的线程执行体。将Runnable实现类实例作为Thread的target来创建Thread对象,该Thread对象才是真正的线程对象。
3、通过Callable和Future创建线程
(1)实现Callable接口,重写call()方法,该call()方法将作为线程执行体,并且有返回值。
(2)使用FutureTask类来包装Callable对象。
(3)使用FutureTask对象作为Thread对象的target创建并启动新线程。
(4)调用FutureTask对象的get()方法来获得子线程执行结束后的返回值。
Callable和Future
Callable接口类似于Runnable,但是Runnable不会返回结果,并且无法抛出返回结果的异常,而Callable被线程执行后,可以返回值,这个返回值可以被Future拿到,下面来看一个简单的例子:
public class CallableAndFuture {
public static void main(String[] args) {
Callable<Integer> callable = new Callable<Integer>() {
public Integer call() throws Exception {
return new Random().nextInt(100);
}
};
FutureTask<Integer> future = new FutureTask<Integer>(callable);
new Thread(future).start();
try {
Thread.sleep(5000);// 可能做一些事情
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
FutureTask实现了两个接口,Runnable和Future,所以它既可以作为Runnable被线程执行,又可以作为Future得到Callable的返回值,那么这个组合的使用有什么好处呢?假设有一个很耗时的返回值需要计算,并且这个返回值不是立刻需要的话,那么就可以使用这个组合,用另一个线程去计算返回值,而当前线程在使用这个返回值之前可以做其它的操作,等到需要这个返回值时,再通过Future得到,岂不美哉!
下面来看另一种方式使用Callable和Future,通过ExecutorService的submit方法执行Callable,并返回Future,代码如下:
public class CallableAndFuture {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newSingleThreadExecutor();
Future<Integer> future = threadPool.submit(new Callable<Integer>() {
public Integer call() throws Exception {
return new Random().nextInt(100);
}
});
try {
Thread.sleep(5000);// 可能做一些事情
System.out.println(future.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
执行多个带返回值的任务,并取得多个返回值,代码如下:
public class CallableAndFuture {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newCachedThreadPool();
CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(threadPool);
for(int i = 1; i < 5; i++) {
final int taskID = i;
cs.submit(new Callable<Integer>() {
public Integer call() throws Exception {
return taskID;
}
});
}
// 可能做一些事情
for(int i = 1; i < 5; i++) {
try {
System.out.println(cs.take().get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
}
其实也可以不使用CompletionService,可以先创建一个装Future类型的集合,用Executor提交的任务返回值添加到集合中,最后遍历集合取出数据,代码略。提交到CompletionService中的Future是按照完成的顺序排列的,这种做法中Future是按照添加的顺序排列的。
线程协作-生产者/消费者问题
处理线程协作时,在同步方法中,必须作进入检查,如果不满足进入条件,须放弃锁,进入等待
完成业务处理后,要做*处理,同时完成唤醒其他等待的线程的操作
1. import java.util.ArrayList;
2. import java.util.List;
3. /** 定义一个盘子类,可以放鸡蛋和取鸡蛋 */
4. publicclass Plate {
5. /** 装鸡蛋的盘子 */
6. List<Object> eggs = new ArrayList<Object>();
7. /** 取鸡蛋 */
8. publicsynchronized Object getEgg() {
9. while (eggs.size() == 0) {
10. try {
11. wait();
12. } catch (InterruptedException e) {
13. e.printStackTrace();
14. }
15. }
16. Object egg = eggs.get(0);
17. eggs.clear();// 清空盘子
18. notify();// 唤醒阻塞队列的某线程到就绪队列
19. System.out.println("拿到鸡蛋");
20. return egg;
21. }
22. /** 放鸡蛋 */
23. publicsynchronizedvoid putEgg(Object egg) {
24. while (eggs.size() > 0) {
25. try {
26. wait();
27. } catch (InterruptedException e) {
28. e.printStackTrace();
29. }
30. }
31. eggs.add(egg);// 往盘子里放鸡蛋
32. notify();// 唤醒阻塞队列的某线程到就绪队列
33. System.out.println("放入鸡蛋");
34. }
35. staticclass AddThread implements Runnable {
36. private Plate plate;
37. private Object egg = new Object();
38. public AddThread(Plate plate) {
39. this.plate = plate;
40. }
41. publicvoid run() {
42. plate.putEgg(egg);
43. }
44. }
45. staticclass GetThread implements Runnable {
46. private Plate plate;
47. public GetThread(Plate plate) {
48. this.plate = plate;
49. }
50. publicvoid run() {
51. plate.getEgg();
52. }
53. }
54. publicstaticvoid main(String args[]) {
55. Plate plate = new Plate();
56. for(int i = 0; i < 10; i++) {
57. new Thread(new AddThread(plate)).start();
58. new Thread(new GetThread(plate)).start();
59. }
60. }
61. }
JAVA线程池
1. 创建一个可重用固定线程集合的线程池,以共享的*队列方式来运行这些线程。
ExecutorService threadPool = Executors.newFixedThreadPool(3);// 创建可以容纳3个线程的线程池
2. 创建一个可根据需要创建新线程的线程池,但是在以前构造的线程可用时将重用它们。
ExecutorService threadPool = Executors.newCachedThreadPool();// 线程池的大小会根据执行的任务数动态分配
3. 创建一个使用单个 worker 线程的 Executor,以*队列方式来运行该线程。
ExecutorService threadPool = Executors.newSingleThreadExecutor();// 创建单个线程的线程池,如果当前线程在执行任务时突然中断,则会创建一个新的线程替代它继续执行任务
4. 创建一个可安排在给定延迟后运行命令或者定期地执行的线程池。
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(3);// 效果类似于Timer定时器
JAVA线程池其实也是一个生产者和消费者模式
线程池模拟:
package threadpool; import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit; public class ThreadPool
{
private int threadNum; private LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();; private TaskThread[] taskThread; public ThreadPool(int threadNum)
{
this.threadNum = threadNum; taskThread = new TaskThread[threadNum]; for (int i=0; i<threadNum; i++)
{
taskThread[i] = new TaskThread(i);
taskThread[i].start();
}
} public void execute(Runnable run)
{
queue.offer(run);
} public void destroy()
{
while (!queue.isEmpty())
{
System.out.println(queue.size() + " job(s) not finish!");
try
{
Thread.sleep(1000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
} System.out.println("begin to destroy"); for (int i=0; i<threadNum; i++)
{
taskThread[i].setDone();
} for (int i=0; i<threadNum; i++)
{
try
{
taskThread[i].join();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
} System.out.println(queue.size() + " job(s) not finish!"); queue.clear();
} private final class TaskThread extends Thread
{
private boolean done = false; private int no; public TaskThread(int no)
{
super();
this.no = no;
} public void run()
{
Runnable run = null;
while (!done)
{
System.out.println("task " + no + " ready...");
try
{
run = queue.poll(1, TimeUnit.SECONDS);
if (null == run)
{
continue;
}
System.out.println("task " + no + " get job");
Thread.sleep(500);
run.run();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
System.out.println("task " + no + " done");
} public void setDone()
{
done = true;
}
} public static void main(String[] args)
{
ThreadPool pool = new ThreadPool(5); pool.execute(new Runnable(){
public void run()
{
System.out.println("Runnable 1 added");
}
});
pool.execute(new Runnable(){
public void run()
{
System.out.println("Runnable 2 added");
}
});
pool.execute(new Runnable(){
public void run()
{
System.out.println("Runnable 3 added");
}
});
pool.execute(new Runnable(){
public void run()
{
System.out.println("Runnable 4 added");
}
}); pool.destroy();
} }