Java并发编程的艺术

java并发编程笔记

知识补充

• Java多线程
• Java注解
• Java反射

Thead类

/**
 * @author yancy0109
 */
public class TestThreadByExtendThread extends Thread{

    @Override
    public void run() {
        for (int i = 0 ; i < 100; i++){
            System.out.println(Thread.currentThread().getName() + " : " + i);
        }
    }
    
    public static void main(String[] args) {
        new TestThreadByExtendThread().start();
        
        for (int i = 0 ; i < 100; i++){
            System.out.println(Thread.currentThread().getName() + " : " + i);
        } 
    }
}

Thread一些方法

• sleep( ) 字面意思 线程休眠 进入阻塞状态 不会释放锁
• yield( ) 线程礼让 CPU重新选择线程
• join( ) 阻塞其他线程，执行完当前线程后再执行其他线程
• getState() 获得线程状态
• setPriority(int newPriority) 设置优先级()，优先级高低影响的是被CPU选中的概率
• setDaemon(boolean on) 设置是否为守护线程(虚拟机执行结束不必等待守护线程执行完毕..如gc等，但必须等待用户线程执行完毕)

Thread状态

public enum State {
    /**
     * Thread state for a thread which has not yet started.
     */
    NEW,

    /**
     * Thread state for a runnable thread.  A thread in the runnable
     * state is executing in the Java virtual machine but it may
     * be waiting for other resources from the operating system
     * such as processor.
     */
    RUNNABLE,

    /**
     * Thread state for a thread blocked waiting for a monitor lock.
     * A thread in the blocked state is waiting for a monitor lock
     * to enter a synchronized block/method or
     * reenter a synchronized block/method after calling
     * {@link Object#wait() Object.wait}.
     */
    BLOCKED,

    /**
     * Thread state for a waiting thread.
     * A thread is in the waiting state due to calling one of the
     * following methods:
     * <ul>
     *   <li>{@link Object#wait() Object.wait} with no timeout</li>
     *   <li>{@link #join() Thread.join} with no timeout</li>
     *   <li>{@link LockSupport#park() LockSupport.park}</li>
     * </ul>
     *
     * <p>A thread in the waiting state is waiting for another thread to
     * perform a particular action.
     *
     * For example, a thread that has called <tt>Object.wait()</tt>
     * on an object is waiting for another thread to call
     * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on
     * that object. A thread that has called <tt>Thread.join()</tt>
     * is waiting for a specified thread to terminate.
     */
    WAITING,

    /**
     * Thread state for a waiting thread with a specified waiting time.
     * A thread is in the timed waiting state due to calling one of
     * the following methods with a specified positive waiting time:
     * <ul>
     *   <li>{@link #sleep Thread.sleep}</li>
     *   <li>{@link Object#wait(long) Object.wait} with timeout</li>
     *   <li>{@link #join(long) Thread.join} with timeout</li>
     *   <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li>
     *   <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li>
     * </ul>
     */
    TIMED_WAITING,

    /**
     * Thread state for a terminated thread.
     * The thread has completed execution.
     */
    TERMINATED;
}

/**
 * Returns the state of this thread.
 * This method is designed for use in monitoring of the system state,
 * not for synchronization control.
 *
 * @return this thread's state.
 * @since 1.5
 */
public State getState() {
    // get current thread state
    return sun.misc.VM.toThreadState(threadStatus);
}

设计模式

静态代理模式

Thread 实现了 Runable接口，并且内部有runable对象，可以直接调用该对象run方法。

可以通过实现Runable接口，再注入Thread对象中，Runable类对象可以更专注于自己的功能。

//Thread
/**
     * If this thread was constructed using a separate
     * <code>Runnable</code> run object, then that
     * <code>Runnable</code> object's <code>run</code> method is called;
     * otherwise, this method does nothing and returns.
     * <p>
     * Subclasses of <code>Thread</code> should override this method.
     *
     * @see     #start()
     * @see     #stop()
     * @see     #Thread(ThreadGroup, Runnable, String)
     */
@Override
public void run() {
    if (target != null) {
        target.run();
    }
}

Runable接口

由于Java是单继承，所以不推荐使用Thread实现，通过实现Runable接口，可以使同一个对象被多个线程使用。

/**
 * @author yancy0109
 */
public class TestTreadForConcurrent implements Runnable{

    //票数
    private static int ticketNums = 10;
    @Override
    public void run() {
        while (true) {
            if (ticketNums <= 0) {
                break;
            }
            //模拟延时
            try {
                Thread.sleep(200);
            } catch (Exception e) {
                e.printStackTrace();
            }

            System.out.println(Thread.currentThread().getName() + "···拿到了第" + ticketNums-- + "票");
        }
    }

    public static void main(String[] args) {
        TestTreadForConcurrent ticket = new TestTreadForConcurrent();

        new Thread(ticket, "1").start();
        new Thread(ticket, "2").start();
        new Thread(ticket, "3").start();
        new Thread(ticket, "4").start();
     	//发生了并发问题
    }
}

Thread类构造方法示例

/**
 * Allocates a new {@code Thread} object. This constructor has the same
 * effect as {@linkplain #Thread(ThreadGroup,Runnable,String) Thread}
 * {@code (null, target, name)}.
 *
 * @param  target
 *         the object whose {@code run} method is invoked when this thread
 *         is started. If {@code null}, this thread's run method is invoked.
 *
 * @param  name
 *         the name of the new thread
 */
public Thread(Runnable target, String name) {
    init(null, target, name, 0);
}

Callable接口

Callable可以在call()方法中处理任务中异常或者向上抛出，并且此方法可以返回执行结果。

package com.test.callable;

import java.util.concurrent.*;

/**
 * Callable 接口实现
 * @author yancy0109
 */
public class TestCallable implements Callable<Boolean> {


    @Override
    public Boolean call() throws Exception {
        for (int i = 0; i < 5;  i++){
            System.out.println(Thread.currentThread().getName() + " :: " + i);
        }
        return true;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //创建执行线程任务?
        TestCallable callable1 = new TestCallable();
        TestCallable callable2 = new TestCallable();
        TestCallable callable3 = new TestCallable();

        //创建执行服务
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        //提交执行
        Future<Boolean> submit1 = executorService.submit(callable1);
        Future<Boolean> submit2 = executorService.submit(callable2);
        Future<Boolean> submit3 = executorService.submit(callable3);

        //获取结果
        Boolean result1 = submit1.get();
        Boolean result2 = submit2.get();
        Boolean result3 = submit3.get();

        //关闭服务
        executorService.shutdownNow();
    }
}

线程同步

Synchronized

多个线程操作同一个资源 – synchronized

Synchronized影响效率

同步方法

加锁对象为this，当前方法同一时间只有一个线程可以执行。

class Buy implements Runnable {

    private int ticketNums = 10;
    boolean flag = true;
    @Override
    public void run() {
        while (flag){
            try {
                buy();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
    //默认同步监视器位this,当前类只有一个线程可以执行buy()方法
    synchronized void buy() throws InterruptedException {
        if (ticketNums <= 0) {
            flag = false;
            return;
        }
        //买票
        System.out.println(Thread.currentThread().getName()+"拿到"+ ticketNums--);
    }
}

同步块

Synchronized(Object){ }

Object称之为同步监视器，OBJECT可以为任意一个对象，使得一个对象只能有一个线程进行操作。

public class SynchronizedBlock {

    public static void main(String[] args) {
        Account account = new Account(500);
        Runnable runnable = () -> {
        while (true) {
            // 只允许一个线程操作这个对象
            synchronized (account) {
                if (account.getMoney() <= 0){
                    break;
                }
                account.setMoney(account.getMoney() - 100);
                System.out.println(Thread.currentThread().getName()+ "取钱"+ "   当前余额   " + account.getMoney());
            }
            try {
                // 线程礼让
                Thread.sleep(1111);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        };
        new Thread(runnable, "1").start();
        new Thread(runnable, "2").start();
    }
}

class Account {

    public Account(int money) {
        Money = money;
    }

    private int Money;

    public int getMoney() {
        return Money;
    }

    public void setMoney(int money) {
        Money = money;
    }
}

Lock

Lock接口

package java.util.concurrent.locks;

volatile

volatile是轻量级的synchronized,它在多处理器开发中保证了共享变量的”可见性”.

可见性指的是一个线程修改一个共享变量时,另外的一个线程能读到这个修改的值.

我的理解是,synchronized实现的是只能有一个线程去操作一个变量,而volatile实现的是不同线程在操作时如果变量被修改,不同线程内访问到的是相同的。

线程缓存后的数据，如果在另一个线程被修改i，当前线程数据将会被标记为失效。

等待通知机制

涉及到Object方法：

• Object.wait() 将操作对象线程阻塞至当前对象的等待队列，释放锁
• Object.notify() 唤醒当前等待队列中任意线程，在同步代码块执行完成后释放锁

注意：需要配合synchronized关键字使用

优点

使线程处于阻塞状态，而非通过while循环判断是否可以执行方法，节省了CPU资源。

生产者和消费者

/**
 * 生产者和消费者
 * @author yancy0109
 */
public class Customer {

    private final static List<String> list = new ArrayList<>();

    private static int listNum = 0;

    // 等待/通知锁
    private final static Object lock = new Object();

    // List最大存储数量
    private final static int MAX_PRODUCE = 3;

    public static String getList() {
        String s = list.get(0);
        list.remove(0);
        return s;
    }

    public static void main(String[] args) {
        Thread customer = new Thread(() -> {
            int getAll = 5;
            synchronized (Customer.lock) {
                for (int i = 0; i < getAll; i++) {
                    // 如果没有商品，则让当前线程阻塞
                    if (Customer.listNum == 0) {
                        try {
                            Customer.lock.wait();
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    // 消费
                    System.out.println(Thread.currentThread().getName() + ":" + Customer.getList());
                    Customer.listNum--;

                    if (i != getAll-1) {
                        // 唤醒生产者可能由于存储空间大小而产生的阻塞
                        Customer.lock.notify();
                    }
                }
                System.out.println("消费完成");
            }
        }, "消费者");

        Thread producer = new Thread(() -> {
            int produceAll = 5;
            synchronized (Customer.lock) {
                for (int i = 0; i < produceAll; i++) {
                    // 如果存储空间不足
                    if (Customer.listNum == Customer.MAX_PRODUCE) {
                        try {
                            // 阻塞至消费者消费后唤醒
                            Customer.lock.wait();
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                    }
                    // 唤醒阻塞消费者进程
                    System.out.println(Thread.currentThread().getName() + ":" + "生产第" + (i+1) + "号产品");
                    Customer.list.add("第" + (i+1) + "号产品");
                    Customer.listNum++;
                    // 如果还未生产完，通知消费者线程，防止它处于阻塞
                    if (i != produceAll-1) {
                        Customer.lock.notify();
                    }
                }
                System.out.println("生产完成");
            }
        }, "生产者");

        customer.start();
        producer.start();
    }
}