显示标签为“多线程”的博文。显示所有博文
显示标签为“多线程”的博文。显示所有博文

2014年8月4日星期一

JAVA多线程编程(二)

原子操作

原子操作,不可被中断的一个或一系列的操作;即该操作具有原子性;
例如为什么i++不是一个原子操作?
原因是执行i++是分为三步,read, inc, write。如果i=1,执行两次i++之后,那么i的最终结果可能不会是3,而是2。
加入现在如果有一个多线程的程序对线程的共享变量i进行操作,每次执行一次i++操作,最终i的输出和执行次数是一致的吗?例如下面的代码:
static int i = 0;
public static void loop() throws InterruptedException {
        Thread[] threads = new Thread[1000];
        for (int j = 0; j < 1000; j++) {
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    i++;
                }
            });
            threads[j] = t;
            t.start();
        }
        for (int j = 0; j < 1000; j++) {
            threads[j].join();//等待所有线程执行完毕
        }
        System.out.println(i.get());
    }

程序输出:999

原因分析

在《JVM原理简述》中,在栈的部分提到了下面的部分:
每一个线程被创建的时候,都会得到自己的程序计数器和java栈。java栈以帧为单位保存调用信息。当线程调用一个方法,JVM会向栈中压入一个新的栈帧,当方法调用结束时,栈帧会被弹出。java栈帧是线程私有的,不存在多线程的安全性问题。
        栈帧由3部分构成,局部变量区,操作数栈,帧数据区。
        1. 局部变量区
        编译器将调用方法的局部变量和参数按照声明顺序放入局部变量数组中,从0开始计数。如果是实例方法,this将被存储到数组的第一个位置。java中的对象传递都是按照引用传递的,因此java对象全部存储在堆中,而局部变量区和操作数栈中只存储了对象的引用。
        2. 操作数栈
        JVM把操作数栈作为它的工作区,大多数指令都要从这里弹出数据,执行运算,然后把结果压回操作数栈,然后等相关的指令将结果再次弹出。操作数栈扮演了暂存操作数的角色。
        3. 帧数据区
        栈帧还可以存储常量池的解析,正常方法返回和异常派发机制。这些都存在帧数据区。当方法要访问常量池时,会通过帧数据区的指向常量池的指针来访问。
每个线程在创建时,都会得到自己的计数器和栈,每次调用方法的时候,都以帧为单位保存调用信息;在栈帧中,i的值会从主内存中加载进来,线程会创建一个局部变量对象i,栈中会保存局部变量的引用,当i的操作完成后,再向主内存更新数据;这就造成同一时间两个线程读取到了同样的值,操作之后又会写回同样的数据,造成最终的i的值小于1000。

volatile

现在我们不禁想起了volatile标签,如果i是volatile类型的变量,结果有变化吗?我们将程序第一行的i声明加上volatile,再执行一次;
程序输出:998
从输出结果看,还是没有改进;证明volatile不能将非原子操作的行为限定为原子操作;
volatile的作用是强制线程在操作变量时都从主内存读取数据,并且操作完成后将操作后的值写入主内存。由于i++不是原子操作,分为read,inc,write三步,因此read即使每次都强制从主内存读取,但是依然有可能两个线程读取到相同的值;

CAS

Compare and swap,CAS操作需要输入两个数值,一个旧值(期望操作前的值)和一个新值,在操作期间先比较下旧值有没有发生变化,如果没有发生变化,才交换成新值,发生了变化则不交换。
说到CAS,要先了解一下处理器是如何保证原子操作的。
32位IA-32处理器使用基于对缓存加锁或总线加锁的方式来实现多处理器之间的原子操作。
JAVA中利用处理器对CMPXCHG指令来实现原子操作,CMPXCHG在处理器中被增加了#LOCK前缀,因此它在处理时是原子操作;
如下代码可以实现对i计数器的原子操作:
volatile static AtomicInteger i = new AtomicInteger(0);
    public static void main(String[] args) throws InterruptedException {
        for (int k = 0; k < 100; k++) {
            loop();
        }
    }

    public static void loop() throws InterruptedException {
        Thread[] threads = new Thread[1000];
        for (int j = 0; j < 1000; j++) {
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    safeCount();
                }
            });
            threads[j] = t;
            t.start();
        }
        for (int j = 0; j < 1000; j++) {
            threads[j].join();
        }
        System.out.println(i.get());
    }

    public static void safeCount() {
        for (; ; ) {
            int tmp = i.get();
            if (i.compareAndSet(tmp, ++tmp)) {
                break;
            }
        }
    }
输出:
1000
2000
3000
4000
5000
使用CAS会存在下面几个问题

  • ABA的问题,即A被修改成为B,又被修改为A,此时使用CAS会发现变量没有变化,实际上却变化了;解决的办法是加入版本号,即1A-2B-3A。从Java1.5开始JDK的atomic包里提供了一个类AtomicStampedReference来解决ABA问题。这个类的compareAndSet方法作用是首先检查当前引用是否等于预期引用,并且当前标志是否等于预期标志,如果全部相等,则以原子方式将该引用和该标志的值设置为给定的更新值。
  • CAS的CPU开销问题,由于存在循环,因此某些时候会使得CPU的开销比较大;
  • 只能保证一个共享变量的原子操作,无法对多个变量保证。如果存在多个变量,建议使用锁;

Lock

下面我们用加锁的方式来实现i++的原子操作;
    volatile static int i = 0;
    static final ReentrantLock lock = new ReentrantLock();
    public static void main(String[] args) throws InterruptedException {
        for(int k=0;k<100;k++){
            loop();
        }
    }
    public static void loop() throws InterruptedException {
        for(int j = 0;j<1000;j++){
            Thread t = new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    lock.lock();
                    try{
                        i++;
                    }finally {
                        lock.unlock();
                    }

                }
            });
            t.start();
        }
        Thread.sleep(1001);
        System.out.println(i);
    }
程序输出
1000
2000
3000
4000
5000

线程池(Threadpool)

百度百科关于线程池的定义非常全面。
线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小,以默认的优先级运行,并处于多线程单元中。如果某个线程在托管代码中空闲(如正在等待某个事件),则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙,但队列中包含挂起的工作,则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队,但他们要等到其他线程完成后才启动。
以java中的Excutor组件为实例说明线程池的整体流程;
以Executor组件的ThreadPoolExecutor为例来说明线程池的运行规则;
首先,调用new方法创建一个ThreadPoolExecutor实例,代码:
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
我们看到有几个核心参数,

  • corePoolSize:可以保留运行线程的数量
  • maximumPoolSize:最大线程数量
  • workQueue:阻塞队列,用于存储任务
下面再来看一下主要的方法,execute(),添加任务,
public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
            if (runState == RUNNING && workQueue.offer(command)) {
                if (runState != RUNNING || poolSize == 0)
                    ensureQueuedTaskHandled(command);
            }
            else if (!addIfUnderMaximumPoolSize(command))
                reject(command); // is shutdown or saturated
        }
    }
以上代码可以看出,线程池的运行机制如下:

  1. 当poolsize<corePoolSize时,任务会被直接以创建workerThread的形式直接执行,不会存储到queue中;
  2. 当poolsize达到corePoolSize大小时,会向队列存储任务,workQueue.offer(), 这个在前面的blog中已经介绍过,offer是非阻塞方法,并且如果队列满的话,会直接返回false;插入队列的数据会被现有的workerThreads消费执行;
  3. 当队列无法插入数据的时候,会创建新的workerThread来执行,直到当前的poolSize达到了maximumPoolSize
  4. 当无法向队列插入数据,并且poolSize= maximumPoolSize,会拒绝这个任务;
  5. 如果poolsize>corePoolSize,那么多余的线程会在等待keepAliveTime时间后销毁;
下图解释了整个流程。
线程池说明


参考资料:

2014年8月1日星期五

JAVA多线程编程(一)

Thread.join(), Thread.start()

Java线程类java.lang.Thread中,有两个方法,start和join。start和直接执行run有何区别?调用join方法会导致什么后果?

start()

启动一个线程并调用run()方法;API中的注释说明:调用start()方法会导致线程开始执行;JVM会调用本线程的run()方法;结果是两个线程并行运行:当前线程和另外执行run方法的线程;
如果直接调用run()方法,则当前线程不会启动;

join()

join()方法在API的注释是:等待直到当前线程死去;这个方法可能会抛出InterruptedException,当其他任何线程打断当前线程时;
它有两个重载方法:join(long millis)和join(long millis, int nanos);
join(long millis):在millis(单位毫秒)时间内等待线程死去;第二个是再多等nanos纳秒的时间;
join可以用来使线程顺序执行,例如:
Thread t2 = new Thread(consumer);
Thread t3 = new Thread(consumer);
Thread t4 = new Thread(consumer);
try {
   t2.start();
   t2.join();
   t3.start();
   t3.join();
   t4.start();
   t4.join();
} catch (InterruptedException e) {
 e.printStackTrace();
}

Lock接口和synchronized

@since 1.5,在JDK5中添加的锁接口;它相比synchronized关键字,提供了更多的锁操作;除了可以实现和synchronized关键字一样的效果之外,Lock接口还提供了读写锁,读操作和写操作的加锁是分开的。提高了锁的效率;也可以满足类似ConcurrentHashMap这种数据结构的高性能的有条件锁;
如图,ConcurrentHashMap的锁结构,

ConcurrentHashMap通过分段锁的机制来达到提高并发处理能力的目的,每个Segment是一个重入锁(ReentrantLock,每次写数据的时候,会根据key的Hash值来使用不同的Segment加锁;

wait和sleep方法的不同

wait()方法是Object的一个方法,调用这个方法会导致当前线程处于等待的状态,直到另外一个线程调用它的notify()或notifyAll()方法,或者一个指定的超时时间已经结束;这个方法会导致当前线程将自己放在一个等待集合中,然后放弃所有在这个对象的同步锁;sleep()方法会导致线程休眠指定的时间,而休眠期间,所有关于当前线程的锁都不会被释放;

volatile关键字

volatile关键字修饰的共享变量在线程并发读取时,JAVA内存模型保证每个线程在读取的时刻的值永远是最新的数据;
解释volatile要说到JMM(JAVA内存模型);
我们看下面的代码,执行一个volatile类型的变量的new操作会导致:
0x01a3de1d: movb $0x0,0x1104800(%esi);
0x01a3de24: lock addl $0x0,(%esp);
volatile变量会比普通变量多了第二个汇编操作,这个操作在多核处理器会引起:

  • 将当前处理器缓存的数据保存到系统内存
  • 其他cpu的缓存数据失效
操作系统中,处理器为了处理数据速度更快,通常会将数据缓存到L1,L2级缓存中;但是处理器不知道何时将数据回写到主存中,当volatile类型的变量被创建时,JVM会向处理器发送一个lock指令,此时处理器会在更新数据之后,将缓存行数据回写到主存中。其他的处理器的数据还是旧的,为了实现缓存一致性,各个处理器会实现一个缓存一致性协议,每个处理器通过嗅探在总线上传播的数据来判断自己的数据是否过期了。当处理器发现自己的缓存行数据地址被修改,当前处理器会将这个数据标记为无效,当需要操作这个数据时,处理器会强制从系统内存中重新读取这个数据;

JAVA中阻塞队列

java.util.concurrent.BlockingQueue是java中的阻塞队列接口; 他的核心方法包括:
抛出异常 特殊值 阻塞 超时
插入 add(e) offer(e) put(e) offer(e, time, unit)
移除 remove() poll() take() poll(time, unit)
检查 element() peek() 不可用 不可用


  • add()向队列写入一条数据,如果对象写入失败,则抛出异常,否则返回true
  • remove() 从队列删除数据,如果队列不包括这个对象,则抛出异常,否则返回true
  • offer()想队列写入一条数据,如果写入成功返回true,否则返回false
  • pool()从队列头获取一条数据,在指定超时时间内未能成功获取,则返回null
  • put() 向队列写入一条数据,如果队列没有空间写入,则阻塞知道写入成功为止;
  • take() 从队列头部获取一条数据,如果队列没有数据可以获取,则阻塞到有数据获取为止;

BlockingQueue特点

  • BlockingQueue可以限定容量;没有容量限制的队列,总会返回Integer.MAX_VALUE的剩余容量;
  • BlockingQueue主要用于生产者-消费者队列;关于生产者-消费者问题;
  • BlockingQueue的实现是线程安全的;
  • BlockingQueue不支持使用一种类似close或shutdown的操作来指定不再添加对象;

CountdownLatch和CyclicBarriar

CountdownLatch

    可以看做是一个同步计数器;任何调用该对象的await()方法的函数都会被阻塞;调用该对象的countdown方法可以使计数器减一,这是一个原子操作,同时只有一个线程可以执行这个方法,直到计数器=0之后,await方法的阻塞会被取消;这个典型的应用场景就是需要等待众多线程工作完才能继续执行的场合,可以利用它来阻塞等待线程执行完毕;

CyclicBarriar

    正好相反,CyclicBarriar也是一个同步计数器,当计数器达到指定数目之后,会执行CyclicBarriar中自定义的run()方法,然后被await()方法阻塞的线程会被唤醒;这个典型应用场景是多个线程在执行前,要等待一个公用的方法执行完毕,可以利用CyclicBarriar计数器来完成;

总结

    上面两个计数器中的CountDownLatch. countdown()是通过CAS实现的原子性;而CyclicBarriar.await()是通过ReentrantLock实现的原子性;
    有关原子性会在下一篇文章中详细讨论;

不可变对象(immutable objects)

不可变对象:即被创建出来之后状态不可改变的对象,例如String,Integer等各种基础类型的包装类,都是不可变对象;每次对它的改变都会创建新的不可变对象;
StringBuffer是可变对象,因为每次修改StringBuffer是修改该对象本身;
图.String's Immutablity
如图,我们创建一个String类型的变量s,
String s = "abcd";
s实际上存储了一个引用地址,该地址指向heap中的一个String对象,"abcd";
s = s.concat(s, "ef");
当对String执行concat操作之后,实际上会创建一个新的String对象:"abcdef", 并且s的引用地址修改指向到新的String对象;
不可变对象在多线程中可以不使用锁机制就可以被多个线程共享;它在多线程编程中有以下特点:
  • 线程安全
  • 不需要锁同步,可以直接使用
  • 提高了性能,避免使用锁或synchronized关键字
  • 可以重复使用;
上图也显示了,不可变对象存在的问题,会产生大量垃圾,给垃圾回收带来压力;对String的concat操作会使之前创建的对象"abcd"变成没有引用的垃圾对象,会逐步被垃圾回收机制回收。

线程调度

分时操作系统

CPU将时间切割为时间片,然后将时间片分配给程序,一个程序的时间片运行结束后,下一个程序的时间片继续执行,多个程序轮流执行;由于CPU的高速处理特性,给人感觉是同时处理一样;

进程和线程

    我们使用的操作系统(Linux,Windows)都有进程的概念,进程就是我们上面所说的程序,它有自己独立的内存空间和系统资源,每个进程的内部的数据和状态都是独立的。创建并执行一个进程的开销比较大,因此线程出现了;
    线程是程序执行流的最小单元,它属于某一个进程,并能和其他线程共享该进程内部的所有资源;线程通常只有寄存器和堆栈,用于存储该线程运行的数据;它比进程更加轻量级,系统在线程间切换比进程消耗更小的资源;一个进程包含多个线程;

线程状态

图. 线程运行机制

  • 创建状态
    • 线程刚刚被创建
  • 可运行状态
    • 线程准备好,等待调度,获取CPU时间片运行前的状态
  • 运行状态
    • 获得CPU资源,处于运行状态
  • 阻塞状态
    • 调用了sleep使得线程处于阻塞状态,此时不会释放锁;
    • 调用await方法,使得线程处于等待状态,此时线程会释放锁
    • 等待获得锁
    • 被阻塞事件阻塞,例如等待数据输入等
  • 死亡状态
    • 线程运行结束

Java线程调度器

java的线程调度器负责调度线程,线程被调度器分为10个级别,1-10,默认是NORMAL_PRIORITY=5.
java的线程调度器负责按照优先级调度线程,一旦时间片有空余,会先让优先级高的线程运行,直到线程dead或sleep或wait,才会执行低优先级的线程;
当有多个线程处于可运行状态,并且优先级相同时,JVM会随机选取一个线程运行。
调度算法有两种:分时和独占式。分时是让所有线程轮流使用CPU时间片;独占则是让一个线程一直执行直到执行结束;JVM采用独占式调度算法;