Java 多线程详解

class SomeThead extends Thraad { 
    public void run()   { 
     // do something here  
    }  
} 
 
public static void main(String[] args) {
    SomeThread oneThread = new SomeThread();   
    
    // 步骤3：启动线程：   
    oneThread.start(); 
}

class SomeRunnable implements Runnable { 
    public void run() { 
    // do something here  
    }  
}
 
Runnable oneRunnable = new SomeRunnable();   
Thread oneThread = new Thread(oneRunnable);   
oneThread.start();

public interface Callable { 
  V call() throws Exception;  
} 
 
// 步骤1：创建实现 Callable 接口的类 SomeCallable (略);  
 
// 步骤2：创建一个类对象： 
    Callable oneCallable = new SomeCallable(); 
 
// 步骤3：由 Callable 创建一个 FutureTask 对象：   
    FutureTask oneTask = new FutureTask(oneCallable); 
 
// 注释： FutureTask 是一个包装器，它通过接受 Callable 来创建，它同时实现了 Future 和 Runnable 接口。 
// 步骤4：由 FutureTask 创建一个 Thread 对象：   
    Thread oneThread = new Thread(oneTask);   
  
// 步骤5：启动线程：  
    oneThread.start();

public class Test1 {  
 public static void main(String[] args) throws InterruptedException {
     System.out.println(Thread.currentThread().getName());  
     MyThread myThread=new MyThread();  
     myThread.start();  
     myThread.sleep(1000);//这里sleep的就是main线程，而非myThread线程  
     Thread.sleep(10);  
     for (int i=0;i<100;i++) {  
         System.out.println("main"+i);  
     }  
 }  
}

public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        new MyThread("低级", 1).start();  
        new MyThread("中级", 5).start();  
        new MyThread("高级", 10).start();  
    }  
}  
  
class MyThread extends Thread {  
    public MyThread(String name, int pro) {  
        super(name); // 设置线程的名称  
        this.setPriority(pro); // 设置优先级  
    }  
  
    @Override  
    public void run() {  
        for (int i = 0; i < 30; i++) {
            System.out.println(this.getName() + "线程第" + i + "次执行！");  
            if (i % 5 == 0) {
                Thread.yield();
            }  
        }  
    }  
}

// 当前线程等该加入该线程后面，等待该线程终止。
void join()
 
// 当前线程等待该线程终止的时间最长为 millis 毫秒。 如果在 millis 时间内，该线程没有执行完，那么当前线程进入就绪状态，重新等待 CPU 调度  
void join(long millis)
 
// 等待该线程终止的时间最长为 millis 毫秒 + nanos 纳秒。如果在 millis 时间内，该线程没有执行完，那么当前线程进入就绪状态，重新等待 CPU 调度  
void join(long millis,int nanos)

MAX_PRIORITY   = 10

MIN_PRIORITY   = 1

NORM_PRIORITY  = 5

public class Test1 {  
    public static void main(String[] args) throws InterruptedException {  
        new MyThread("高级", 10).start();  
        new MyThread("低级", 1).start();  
    }  
}  

class MyThread extends Thread {  
    public MyThread(String name,int pro) {  
        super(name); // 设置线程的名称  
        setPriority(pro); // 设置线程的优先级  
    }  
        
    @Override  
    public void run() {  
        for (int i = 0; i < 100; i++) {  
            System.out.println(this.getName() + "线程第" + i + "次执行！");  
        }  
    }  
}

// 将该线程标记为守护线程或用户线程。当正在运行的线程都是守护线程时，Java 虚拟机退出。
// 该方法必须在启动线程前调用。 该方法首先调用该线程的 checkAccess 方法，且不带任何参数。这可能抛出 SecurityException（在当前线程中）。
// 参数：on 如果为 true，则将该线程标记为守护线程。
// 如果该线程处于活动状态，抛出 IllegalThreadStateException异常。 如果当前线程无法修改该线程，抛出 SecurityException 异常。
public final void setDaemon(boolean on)

class MyThread extends Thread {  
    int i=0;  
    boolean next=true;  
    @Override  
    public void run() {  
        while (next) {  
            if(i==10) {
                next=false;
            }  
            i++;  
            System.out.println(i);  
        }  
    }  
}

public synchronized void save(){}

public class Bank {  
     
    private int count =0; // 账户余额  
 
    // 存钱  
    public void addMoney(int money) {  
 
        synchronized (this) {  
            count +=money;  
        }
 
        System.out.println(System.currentTimeMillis()+"存进："+money);  
    }  
 
    // 取钱  
    public void subMoney(int money) {  
 
        synchronized (this) {  
            if(count-money < 0) {  
                System.out.println("余额不足");  
                return;  
            }  
            count -=money;  
        }  
 
        System.out.println(+System.currentTimeMillis()+"取出："+money);  
    }  
 
    // 查询  
    public void lookMoney() {  
        System.out.println("账户余额："+count);  
    } 
}

public class SynchronizedThread {
 
    class Bank {
 
        private volatile int account = 100;
 
        public int getAccount() {
            return account;
        }
 
        /**
         * 用同步方法实现
         * 
         * @param money
         */
        public synchronized void save(int money) {
            account += money;
        }
 
        /**
         * 用同步代码块实现
         * 
         * @param money
         */
        public void save1(int money) {
            synchronized (this) {
                account += money;
            }
        }
    }
 
    class NewThread implements Runnable {
        private Bank bank;
 
        public NewThread(Bank bank) {
            this.bank = bank;
        }
 
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                // bank.save1(10);
                bank.save(10);
                System.out.println(i + "账户余额为：" +bank.getAccount());
            }
        }
 
    }
 
    /**
     * 建立线程，调用内部类
     */
    public void useThread() {
        Bank bank = new Bank();
        NewThread new_thread = new NewThread(bank);
        System.out.println("线程1");
        Thread thread1 = new Thread(new_thread);
        thread1.start();
        System.out.println("线程2");
        Thread thread2 = new Thread(new_thread);
        thread2.start();
    }
 
    public static void main(String[] args) {
        SynchronizedThread st = new SynchronizedThread();
        st.useThread();
    }
}

ReentrantLock() : 创建一个 ReentrantLock 实例         
 
lock() : 获得锁        
 
unlock() : 释放锁

// 只给出要修改的代码，其余代码与上同
class Bank {
    
    private int account = 100;
 
    // 需要声明这个锁
    private Lock lock = new ReentrantLock();
    public int getAccount() {
        return account;
    }
 
    // 这里不再需要synchronized 
    public void save(int money) {
        lock.lock();
        try {
            account += money;
        } finally {
            lock.unlock();
        }
    }
｝

// 生产者和消费者都要操作的资源 
class Resource {   
    private String name;  
    private int count=1;  
    private boolean flag=false;  
 
    public synchronized void set(String name) {  
        if(flag) {  
            try { 
                wait(); 
            } catch(Exception e) {}
        }
        this.name=name+"---"+count++;  
        System.out.println(Thread.currentThread().getName()+"...生产者..."+this.name);  
        flag=true;  
        this.notify();  
    }  
 
    public synchronized void out() {  
        if(!flag) {
            try {
                wait();
            } catch(Exception e) {}  
        }
        System.out.println(Thread.currentThread().getName()+"...消费者..."+this.name);  
        flag=false;  
        this.notify();  
    }  
} 
 
class Producer implements Runnable {  
    private Resource res;  
 
    Producer(Resource res) {  
        this.res=res;  
    }  
 
    public void run() {  
        while(true) {  
            res.set("商品");  
        }  
    }  
}  
 
class Consumer implements Runnable{  
    private Resource res;  
 
    Consumer(Resource res) {  
        this.res=res;  
    }  
 
    public void run() {  
        while(true) {  
            res.out();  
        }  
    }  
 
}  
 
public class ProducerConsumerDemo {  
    public static void main(String[] args) {  
        Resource r=new Resource();  
        Producer pro=new Producer(r);  
        Consumer con=new Consumer(r);  
        Thread t1=new Thread(pro);  
        Thread t2=new Thread(con);  
        t1.start();  
        t2.start();  
    }
}
// 运行结果正常，生产者生产一个商品，紧接着消费者消费一个商品。

class Resource {  
    private String name;  
    private int count=1;  
    private boolean flag=false;  
 
    public synchronized void set(String name) {  
        // 原先是 if，现在改成 while，这样生产者线程从冻结状态醒来时，还会再判断 flag
        while(flag) {
            try {
                wait();
            } catch(Exception e) {}  
        }
        this.name=name+"---"+count++;  
        System.out.println(Thread.currentThread().getName()+"...生产者..."+this.name);  
        flag=true;  
 
        // 原先是 notity()，现在改成 notifyAll()，这样生产者线程生产完一个商品后可以将等待中的消费者线程唤醒，否则只将上面改成 while 后，可能出现所有生产者和消费者都在 wait() 的情况。
        this.notifyAll();
    }  
 
    public synchronized void out() {  
        // 原先是 if，现在改成while，这样消费者线程从冻结状态醒来时，还会再判断flag
        while(!flag) {
            try {
                wait();
            } catch(Exception e) {}  
        } 
           
        System.out.println(Thread.currentThread().getName()+"...消费者..."+this.name);  
        flag=false;  
 
        // 原先是 notity()，现在改成 notifyAll()，这样消费者线程消费完一个商品后可以将等待中的生产者线程唤醒，否则只将上面改成 while 后，可能出现所有生产者和消费者都在 wait() 的情况。
        this.notifyAll(); 
    }  
}  
 
public class ProducerConsumerDemo {  
    public static void main(String[] args) {  
        Resource r=new Resource();  
        Producer pro=new Producer(r);  
        Consumer con=new Consumer(r);  
        Thread t1=new Thread(pro);  
        Thread t2=new Thread(con);  
        Thread t3=new Thread(pro);  
        Thread t4=new Thread(con);  
        t1.start();  
        t2.start();  
        t3.start();  
        t4.start();  
    }  
}

class Resource {  
    private String name;  
    private int count=1;  
    private boolean flag=false;  
    private Lock lock = new ReentrantLock(); // Lock 是一个接口，ReentrantLock 是该接口的一个直接子类
    private Condition condition_pro=lock.newCondition(); // 创建代表生产者方面的 Condition对象  
    private Condition condition_con=lock.newCondition(); // 使用同一个锁，创建代表消费者方面的 Condition 对象
      
    public void set(String name) {  
        lock.lock(); // 锁住此语句与 lock.unlock() 之间的代码  
        try {  
            while(flag) {
                condition_pro.await(); // 生产者线程在 conndition_pro 对象上等待  
            }
            this.name=name+"---"+count++;  
            System.out.println(Thread.currentThread().getName()+"...生产者..."+this.name);  
            flag=true;  
            condition_con.signalAll();  
        }  
        finally {  
            lock.unlock(); // unlock() 要放在 finally 块中
        }  
    }  
 
    public void out() {  
        lock.lock(); // 锁住此语句与 lock.unlock() 之间的代码  
        try {  
            while(!flag) {
                condition_con.await(); // 消费者线程在 conndition_con 对象上等待
            }
            System.out.println(Thread.currentThread().getName()+"...消费者..."+this.name);  
            flag=false;  
            condition_pro.signqlAll(); // 唤醒所有在 condition_pro 对象下等待的线程，也就是唤醒所有生产者线程  
        }  
        finally {  
            lock.unlock();  
        }  
    }  
}

ArrayBlockingQueue ：基于数组实现的 BlockingQueue 队列。
 
LinkedBlockingQueue：基于链表实现的 BlockingQueue 队列。
 
PriorityBlockingQueue：它并不是保准的阻塞队列，该队列调用 remove（）、poll（）、take（）等方法提取出元素时，并不是取出队列中存在时间最长的元素，而是队列中最小的元素。它判断元素的大小即可根据元素（实现 Comparable 接口）的本身大小来自然排序，也可使用 Comparator 进行定制排序。
 
SynchronousQueue：同步队列。对该队列的存、取操作必须交替进行。
 
DelayQueue：它是一个特殊的 BlockingQueue，底层基于 PriorityBlockingQueue 实现，不过，DelayQueue 要求集合元素都实现 Delay 接口（该接口里只有一个 long getDelay（）方法），DelayQueue 根据集合元素的 getDalay（）方法的返回值进行排序。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
 
public class BlockingQueueTest {
    public static void main(String[] args)throws Exception {
        // 创建一个容量为 1 的 BlockingQueue
        
        BlockingQueue<String> b=new ArrayBlockingQueue<>(1);
 
        // 启动 3 个生产者线程
        new Producer(b).start();
        new Producer(b).start();
        new Producer(b).start();
 
        // 启动一个消费者线程
        new Consumer(b).start();
        
    }
}
 
class Producer extends Thread {
    private BlockingQueue<String> b;
    
    public Producer(BlockingQueue<String> b) {
        this.b=b;
    }
    public synchronized void run() {
        String [] str=new String[] {
            "java",
            "struts",
            "Spring"
        };
 
        for(int i=0;i<9999999;i++) {
            System.out.println(getName()+"生产者准备生产集合元素！");
            try {
                b.put(str[i%3]);
                sleep(1000);
                // 尝试放入元素，如果队列已满，则线程被阻塞
            } catch(Exception e) {
                System.out.println(e);
            }
            System.out.println(getName()+"生产完成："+b);
        }
    }
}
 
class Consumer extends Thread {
    private BlockingQueue<String> b;
    public Consumer(BlockingQueue<String> b) {
        this.b=b;
    }
 
    public synchronized void run() {    
        while(true) {
            System.out.println(getName()+"消费者准备消费集合元素！");
            try {
                sleep(1000);
                // 尝试取出元素，如果队列已空，则线程被阻塞
                b.take();
            } catch(Exception e) { 
                System.out.println(e);
            }
            System.out.println(getName()+"消费完："+b);
        }
    
    }
}

ThreadPoolExecutor
ScheduledThreadPoolExecutor

- execute(Runnable)
- submit(Runnable)
- submit(Callable)
- invokeAny(...)
- invokeAll(...)

ExecutorService executorService = Executors.newSingleThreadExecutor();

executorService.execute(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

executorService.shutdown();

Future future = executorService.submit(new Runnable() {
    public void run() {
        System.out.println("Asynchronous task");
    }
});

future.get();  // returns null if the task has finished correctly.

Future future = executorService.submit(new Callable(){
    public Object call() throws Exception {
        System.out.println("Asynchronous Callable");
        return "Callable Result";
    }
});

System.out.println("future.get() = " + future.get());

ExecutorService executorService = Executors.newSingleThreadExecutor();
 
Set<Callable<String>> callables = new HashSet<Callable<String>>();
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});
 
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();

ExecutorService executorService = Executors.newSingleThreadExecutor();
 
Set<Callable<String>> callables = new HashSet<Callable<String>>();
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 1";
    }
});
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 2";
    }
});
 
callables.add(new Callable<String>() {
    public String call() throws Exception {
        return "Task 3";
    }
});
 
List<Future<String>> futures = executorService.invokeAll(callables);
for(Future<String> future : futures) {
    System.out.println("future.get = " + future.get());
}
executorService.shutdown();

public class ForkJoinTest {
    private double[] d;
    private class ForkJoinTask extends RecursiveTask {
        private int first;
        private int last;
        public ForkJoinTask(int first, int last) {
            this.first = first;
            this.last = last;
        }
 
        protected Integer compute() {
            int subCount;
            if (last - first < 10) {
                subCount = 0;
                for (int i = first; i <= last; i++) {
                    if (d[i] < 0.5){
                        subCount++;
                     }
                }
             }else {
                int mid = (first + last) / 2;
                ForkJoinTask left = new ForkJoinTask(first, mid);
                left.fork();
                ForkJoinTask right = new ForkJoinTask(mid + 1, last);
                right.fork();
                subCount = left.join();
                subCount += right.join();
            }
            return subCount;
        }
    }
 
    public static void main(String[] args) {
        ForkJoinPool pool=new ForkJoinPool();
        pool.submit(new ForkJoinTask(0, 9999999));
        pool.awaitTermination(2,TimeUnit.SECONDS);
        System.out.println("Found " + n + " values");
    }
}

// 创建一个线程本地变量
ThreadLocal()
 
// 返回此线程局部变量的当前线程副本中的值，如果这是线程第一次调用该方法，则创建并初始化此副本
T get()
          
// 返回此线程局部变量的当前线程的初始值
protected  T initialValue()

/**
* Returns the value of this variable for the current thread. If an entry
* doesn't yet exist for this variable on this thread, this method will
* create an entry, populating the value with the result of
* {@link #initialValue()}.
*
* @return the current value of the variable for the calling thread.
*/
@SuppressWarnings("unchecked")
public T get() {
   // Optimized for the fast path.
   Thread currentThread = Thread.currentThread();
   Values values = values(currentThread);
   if (values != null) {
        Object[] table = values.table;
        int index = hash & values.mask;
        if (this.reference == table[index]) {
　　　　　      return (T) table[index + 1];
         }
    } else {
　　　　values = initializeValues(currentThread);
　　}
　　return (T) values.getAfterMiss(this);
}
 
/**
* Sets the value of this variable for the current thread. If set to
* {@code null}, the value will be set to null and the underlying entry will
* still be present.
*
* @param value the new value of the variable for the caller thread.
*/
public void set(T value) {
     Thread currentThread = Thread.currentThread();
　　 Values values = values(currentThread);
　　 if (values == null) {
　　　　  values = initializeValues(currentThread);
　　　}
   values.put(this, value);
}