【Java】多线程编程

概念

进程具有自己变量的完备集；线程则共享相同的数据。

抢占式调度：直接中断而不需要实现和被中断程序协商

协作式调度：只有在被中断程序同意交出控制权之后才能执行中断

多线程实现

方法一：

class MyRunnable implements Runnable {
    public void run() {
        ...
    }
}
Runnable r = new MyRunnable();
Thread t = new Thread(r);
t.start();

方法二(不建议)：

class MyThread extends Thread {
    public void run() {
        ...
    }
}
Thread t = new MyThread();
t.start();

Thread类

• sleep(t)：static|线程暂停t毫秒，暂停当前线程的活动，会抛出InterruptedException

• void run()

• void start()

• static Thread currentThread()：返回代表当前执行线程的Thread对象

• void interrupt()：发送中断请求给一个线程，中断状态为true，如果线程当前被sleep调用阻塞，则抛出InterruptedException

• boolean isInterrupted(): 检查线程是否被终止

  public boolean isInterrupted() {
     return isInterrupted(false);
  }
  

• static boolean interrupted()

  public static boolean interrupted() {
     return currentThread().isInterrupted(true);
  }
  

• boolean isAlive() ：线程处于Runnable或Blocked状态返回true

• void join() 等待直到指定的线程死亡

• void setPriority(int newPriority)：设置线程优先级

• static void yield()：当前执行线程处于让步状态，会执行其他具有同样优先级的线程

Runnable接口

• void run()：必须重载

线程中断

中断线程就是其他线程给该线程发一个信号，该线程收到信号后结束执行run()方法，使得自身线程能立刻结束运行。

在其他线程中对目标线程调用interrupt()方法，目标线程需要反复检测是否是interrupted状态，做出相应的响应。

运行流程：

1. 其他线程调用目标线程的interrupt()方法，目标线程的中断位置为true
2. 目标线程调用自身的isInterrupted()方法检测自身的中断位是否为true
3. 目标线程检测到中断状态，可以选择做出响应也可以选择忽略，一般默认为终止线程操作

需要响应中断的Runnable:

Runnable r=()->{
    try{
        while(!Thread.currentThread().isInterrupted && ...)
            ...
    }
    catch(InterrupedException e){
        ...
    }
    finally
    {
        ...
    }
}

当对被阻塞的线程执行interrupted()方法则会抛出interruptedException异常。

线程状态

• New

  new Thread(r)之后线程还没有开始运行，处于New状态

• Runnable

  调用start方法后，线程成为Runnable状态(可能在运行，可能没有)

• Blocked(被阻塞)

  发生以下情况时线程进入被阻塞状态：

  • 调用在I/O上被阻塞的操作
  • 线程试图得到一个锁，而该锁正被其他线程持有

  发生以下情况线程由Blocked变为Runnable:

  • I/O操作完成
  • 等待的锁被释放（或者等待超时）

• Waiting

  • 调用join()方法，不指定超时值

  • 调用Object对象的wait()方法

• Timed_waiting

  调用计时等待的方法

  • Thread.sleep
  • Object.wait
  • Thread.join
  • Lock.tryLock
  • Condition.await

• Terminated

  • run方法正常退出
  • 未捕获异常终止了run方法

线程属性

线程优先级

当调度器有机会选择新线程时，首先选择具有较高优先级的线程。

可以使用setPriority()设置线程优先级，设置范围为1-10之间的整数；一般情况下，线程继承父线程的优先级。

• MIN_PRIORITY=1
• MAX_PRIORITY=10
• NORM_PRIORITY=5

当几个高优先级的线程没有进入非活动状态时，低优先级线程永远也不能执行。

守护线程

调用setDaemon(true)将线程转换为守护线程，为其他线程提供服务，比如计时线程。

守护线程应该不去访问文件，数据库等，因为会发生中断

线程同步

当两个线程同时尝试对同一资源进行访问和修改时，会发生竞争冲突，出现错误，所以需要同步机制。

ReentrantLock

myLock.lock(); // ReentrantLock Object
try{
    ...
}
finally{
    myLock.unlock();
}

当一个线程持有锁，另一个线程执行到lock()语句时会进入阻塞状态。

锁可重入，当锁的持有计数为0时，线程释放锁。

使用lockInterruptibly()可获得可中断锁。

条件对象

当线程要执行一个需要条件的操作时，条件没有达到，需要释放锁让其他线程执行，当条件满足时再回来执行。

使用ReentrantLock对象的newCondition()方法来获得Condition对象。

当条件不满足时调用Condition对象的await()方法，阻塞线程并放弃锁。

当调用了await()方法后，线程进入条件的等待集，只有当另一线程调用了同一条件的signalAll()方将等待集中的所有线程移除，当获得锁后，从await()处继续执行。

在从条件等待集移出后，获得执行权的线程应该再次检测条件，所以await()的调用应该放在循环里

while(!ok to proceed)
    condition.await();

死锁

当一个线程调用await()时，它没有办法重新激活自身，需要其他线程调用signalAll()或者signal()来激活等待线程，当没有线程来重新激活等待线程时，便导致了死锁现象

synchronized关键字

自动提供一个锁以及相关的"条件"，对于大多数需要显式锁的情况都有效。

每个Java对象具有一个内部对象锁instrinsicLock和一个相关条件。

public synchronized void test() {
    ...
}
//相当于
public void test() {
    this.intrinsicLock.lock();
    try{
        ...
    }
    finally {
        this.intrinsicLock.unlock();
    }
}

使用Object对象的wait()方法和notifyAll()来代替Condition对象的await()方法和signalAll()方法。

同步阻塞

synchronized(obj) {
    ...
}

使用obj对象的锁来实现同步阻塞

Volatile

volatile关键字可以对一个变量单一上锁。

线程局部变量

定义ThreadLocal变量：

public static ThreadLocal<S> var = ThreadLocal.widthInitial(()->new S());

使用var.get()来得到线程当前值，首次使用会调用initialize()来得到值。

可以通过set()和remove()方法对该值进行修改

阻塞队列(BlockingQueue)

最简单的阻塞队列，只提供offer()和take()方法(注意在正常的BlockingQueue中的offer()方法的返回值为boolean)

设置队列的最大大小，当队列满时offer()方法阻塞，等待取出；当队列空时take()方法阻塞，等待放入

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BQueue<T> {
    private List<T> list = new ArrayList<>();
    private ReentrantLock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition listFull = lock.newCondition();

    private int maxSize;
    public BQueue(int maxSize) {
        this.maxSize = maxSize;
    }

    public void offer(T item) throws InterruptedException{
        try {
            lock.lockInterruptibly();
            while(list.size() == maxSize) {
                System.out.println("list Full");
                notEmpty.signalAll();
                listFull.await();
            }
            list.add(item);
            notEmpty.signalAll();
        }finally {
            lock.unlock();
        }

    }

    public T take() throws InterruptedException{
        try {
            lock.lockInterruptibly();
            while(list.size() == 0) {
                System.out.println("list empty");
                notEmpty.await();
            }
            T item = list.get(0);
            list.remove(0);
            listFull.signalAll();
            return item;
        }finally {      
            lock.unlock();
        }
    }
}

阻塞队列和生产者/消费者模型

BQueue<Integer> queue = new BQueue<>(3);
Random r = new Random();
Thread producer = new Thread(new Runnable() {
    @Override
    public void run() {
        while (true) {
            try {
                Integer i = r.nextInt();
                queue.offer(i);
                System.out.println("Producer offer: " + i.toString());
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
});
Thread consumer = new Thread(new Runnable(){
    @Override
    public void run() {
        while(true) {
            try {
                Integer i = queue.take();
                System.out.println("Consumer take: " + i.toString());
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
});
producer.start();
consumer.start();