Java多线程学习开发笔记

线程有有序性和可见性

多个线程之间是不能直接传递数据交互的，它们之间的交互只能通过共享变量来实现。

在多个线程之间共享类的一个对象，这个对象是被创建在主内存(堆内存)中，每个线程都有自己的工作内存(线程栈)，工作内存存储了主内存对象的一个副本，当线程操作对象时，首先从主内存复制对象到工作内存中，然后执行代码改变了值，最后用工作内存刷新主内存。

当一个对象在多个内存中都存在副本时，如果一个内存修改了共享变量，其它线程也应该能够看到被修改后的值，此为可见性。

多个线程执行时，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();
    }
    
}