java 线程小结

1, 为什么wait与notify之前必须要加synchronized?

答案其实很简单，也是为了防止等待-通知机制出现race condition

为什么会出现race condition ?
答： 对象在被wait之前已经被另一线程notify , 之后的wait 会永久停止,并导致deadlock(死锁)

理想情况：
1, 第一个线程判断该对象是否要wait
2, 第一个线程将对象wait
3, 第二个线程再将对象notify

实际情况
1, 第一个线程判断该对象是否要wait
2, 第二个线程将对象notify
3, 第一个线程将对象wait

为了防止这些情况，才需要在wait与notify之前加synchronized

java 代码

A a =
 A.getInstance
(
)
;
//单例对象,同一份实例不销毁

synchronized
 (
a)
 {

a.wait
(
)
;

}

-------------------------------
另一线程
A a =
 A.getInstance
(
)
;

synchronized
(
a)
 {

a.notify
(
)
;

}

等待-通知机制必须与sychronized一起用,否则自身也会有 race condition.

2, 静态同步方法与非静态同步方法的区别

有时，我们经常会碰到这样的代码！

业务逻辑的封装类:

public
 class
 Logic {

    private
 static
 final
 Log log =
 LogFactory.getLog
(
Logic.class
)
;

    private
 static
 Logic logic;

 
    private
 Logic(
)
 {
}

 
    public
 static
 Logic getInstance(
)
 {

        if
 (
null
 ==
 logic)
 {

            logic =
 new
 Logic(
)
;

        }

 
        return
 logic;

    }

 
    public
 static
 synchronized
 void
 testStatic(
)
 {

        log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " : static method is running"
)
;

    }

 
    public
 synchronized
 void
 testNonStatic(
)
 {

        log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " : non static method is running"
)
;

    }

}

非静态方法的执行:

public
 class
 ThreadRun1 extends
 Thread
 {

    private
 static
 final
 Log log =
 LogFactory.getLog
(
ThreadRun1.class
)
;

 
    public
 void
 run(
)
 {

        Logic logic =
 Logic.getInstance
(
)
;
 // object reference

 
        try
 {

            Thread
.sleep
(
3000
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
"some exceptions occured :"
, e)
;

        }

 
        logic.testNonStatic
(
)
;

 
        logEnd(
)
;

    }

 
    private
 void
 logEnd(
)
 {

        log.info
(
"thread run1 end"
)
;

    }

}

静态类方法的执行

public
 class
 ThreadRun2 extends
 Thread
 {

    private
 static
 final
 Log log =
 LogFactory.getLog
(
ThreadRun1.class
)
;

 
    public
 void
 run(
)
 {

        Logic.testStatic
(
)
;
 // class static reference

 
        try
 {

            Thread
.sleep
(
5000
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
"some error ocuur :"
, e)
;

        }

 
        logEnd(
)
;

    }

 
    private
 void
 logEnd(
)
 {

        log.info
(
"thread run2 end"
)
;

    }

}

测试类

public
 class
 TestThread {

 
    /**
     * @param args
     */

    public
 static
 void
 main(
String
[
]
 args)
 {

        ThreadRun1 run1 =
 new
 ThreadRun1(
)
;

        run1.start
(
)
;

        ThreadRun2 run2 =
 new
 ThreadRun2(
)
;

        run2.start
(
)
;

    }

}

现在有2根线程，其中一根会调用testStatic() , 而另一根会在testStatic未执行结束前调用testNonStatic！
那么，按照多线程同步原则，该对象会在调用testStatic()方法时被锁定，而该方法未结束前如果调用testNonStatic()方法，则必须要等待第一个线程执行完后，才可以执行继续执行！

但是，实际情况是两线程可同时被调用！

区别在于，前者是静态的，不需要实例化即可调用，那么既然连实例化的对象都没创建，何来锁住对象呢！
大家都知道，静态的方法一般都是直接调用“类.方法”来执行的，因此，调用testStatic锁住的其实是类！（锁住类不等于锁住该类实例的对象！）

总结：每个class只有一个线程可以执行静态同步方法，每个类的对象，只有一个线程可以执行同步方法！当对象实例调用同步方法，而同步方法中又调用了class的静态同步方法，其实此次调用一共锁住了2个不同的对象监视器！

Class级别的锁与Object级别的锁是不一样的, 两者相互独立

3, thread 的 join 方法与 isAlive 方法的区别.

java 代码

log.info
(
"current thread running"
)
;

thread1.join
(
)
;
 // 当前线程在执行到join方法后, 会被block住 , 直到thread1线程处理结束或死亡

log.info
(
"current thread stopping"
)
;

java 代码

log.info
(
"current thread running"
)
;

thread1.isAlive
(
)
;
 // 直接返回true or false

log.info
(
"current thread stopping"
)
;

join方法是使当前线程阻塞,直到引用的线程结束才激活.

4, wait-notify机制

在一个以上的thread wait住时,调用notify是随机的唤醒某一thread.

而notifyAll则是唤醒所有等待的线程, 但只有一个线程可以在唤醒后lock object monitor,
所以, notifyAll操作也是有利弊的.

wait-notify机制, 单次唤醒是随机的, 全部唤醒则会导致大部分线程阻塞.

8, Lock接口替代synchronized

a, Lock接口可以比sychronized提供更广泛的锁定操作.可以提供多把不同的锁.且锁之间互不干涉.
b, Lock接口提供lock()与unlock()方法, 使用明确调用来完成同步的, OO思想好于前者.
c, Lock可以自由操控同步范围(scope).
d, Lock接口支持nested lock(嵌套锁定).并提供了丰富的api.
e, Lock接口提供了tryLock()方法, 支持尝试取得某个object lock.

5, Condition替代wait与notify

// 生产/消费者模式

public
 class
 Basket {

    Lock lock =
 new
 ReentrantLock(
)
;

 
    //产生Condition对象

    Condition produced =
 lock.newCondition
(
)
;

    Condition consumed =
 lock.newCondition
(
)
;

    boolean
 available =
 false
;

 
    public
 void
 produce(
)
 throws
 InterruptedException
 {

        lock.lock
(
)
;

 
        try
 {

            if
 (
available)
 {

                produced.await
(
)
;
 //放弃lock进入睡眠

            }

 
            System
.out
.println
(
"Apple produced."
)
;

 
            available =
 true
;

 
            consumed.signal
(
)
;
 //发信号唤醒等待这个Condition的线程

        }
 finally
 {

            lock.unlock
(
)
;

        }

    }

 
    public
 void
 consume(
)
 throws
 InterruptedException
 {

        lock.lock
(
)
;

 
        try
 {

            if
 (
!
available)
 {

                consumed.await
(
)
;
 //放弃lock进入睡眠

            }

 
            /*吃苹果*/

            System
.out
.println
(
"Apple consumed."
)
;

 
            available =
 false
;

 
            produced.signal
(
)
;
 //发信号唤醒等待这个Condition的线程

        }
 finally
 {

            lock.unlock
(
)
;

        }

    }

}

 
// 测试用类

public
 class
 ConditionTester {

    public
 static
 void
 main(
String
[
]
 args)
 throws
 InterruptedException
 {

        final
 Basket basket =
 new
 Basket(
)
;

 
        //定义一个producer

        Runnable
 producer =
 new
 Runnable
(
)
 {

                public
 void
 run(
)
 {

                    try
 {

                        basket.produce
(
)
;

                    }
 catch
 (
InterruptedException
 ex)
 {

                        ex.printStackTrace
(
)
;

                    }

                }

            }
;

 
        //定义一个consumer

        Runnable
 consumer =
 new
 Runnable
(
)
 {

                public
 void
 run(
)
 {

                    try
 {

                        basket.consume
(
)
;

                    }
 catch
 (
InterruptedException
 ex)
 {

                        ex.printStackTrace
(
)
;

                    }

                }

            }
;

 
        //各产生10个consumer和producer

        ExecutorService service =
 Executors.newCachedThreadPool
(
)
;

 
        for
 (
int
 i =
 0
;
 i <
 10
;
 i++
)

            service.submit
(
consumer)
;

 
        Thread
.sleep
(
2000
)
;

 
        for
 (
int
 i =
 0
;
 i <
 10
;
 i++
)

            service.submit
(
producer)
;

 
        service.shutdown
(
)
;

    }

}

Condition配合Lock接口可以轻松实现,比sychronized配合wait,notify更
强大的功能.

Condition接口可以为单个对象锁生成多个类似wait-notify机制的条件变量.

每个条件变量在执行wait-notify时,只会控制自身条件的线程,即触发notify时,只唤醒
自身条件变量上的wait线程,不会唤醒其他条件变量的wait线程.

建议: 同一把锁下, 允许有多个Condition, 且相互不干涉, 但是, 每个Condition都是按顺序执行的.
(java关键字, 如果使用this, 则范围过大, 自己创建object来局部控制, 又不优雅)

注意: Condition的wait操作, 允许出现人为或意外的”虚假唤醒”, 所以, 为了保证Condition的作用域.
当调用wait时, 尝试使用循环结构.其中condition为await-singal的操作标示.

boolean
 condition =
 true
;

while
(
condition)
 {

condition.await
(
)
;

condition =
 false
;

}

...
condition
 =
 true
;

condition.singal
(
)
;

6, 使用java.util.concurrent.atomic包,原子操作及解决volatile变量计算的race condition

private
 static
 AtomicInteger i =
 new
 AtomicInteger(
0
)
;

 
public
 void
 run(
)
 {

    int
 v =
 i.incrementAndGet
(
)
;
  // 相当于++i

    log.info
(
"i = "
 +
 v)
;

}

包的特色:
1, 普通原子数值类型AtomicInteger, AtomicLong提供一些原子操作的加减运算.

2, 解决race condition问题的经典模式-”比对后设定”, 即 查看主存中数据是否与
预期提供的值一致,如果一致,才更新.

// 这边采用无限循环

for
 (
;;
)
 {

            int
 current =
 get(
)
;

            if
 (
compareAndSet(
current, newValue)
)

                return
 current;

}

3, 使用AtomicReference可以实现对所有对象的原子引用及赋值.包括Double与Float,
但不包括对其的计算.浮点的计算,只能依靠同步关键字或Lock接口来实现了.

4, 对数组元素里的对象,符合以上特点的, 也可采用原子操作.包里提供了一些数组原子操作类

建议: 针对非浮点类型的数值计算, 数组元素及对象的引用/赋值, 优先采用原子类型.

优先考虑使用atmoic框架 .

7, 利用java semaphore信号量机制,控制某操作上线程的数量

java信号量的实现逻辑与操作系统解决进程同步问题时采用的PV操作类似.
即 P -> 临界区 -> V
其中P为消费,V生产,临界区是同步区域.

java semaphore提供了acquire()与release()两种操作,类似Lock的lock()与unlock.
区别在于, java semaphore对acquire有数量控制,即利用它的计数器大小,来控制多少线程可执行,其余全部阻塞.
而Lock中的lock()方法,一次只能允许一根线程执行,其余全部阻塞.

semaphore接口的构造函数中还提供了 一个boolean型的fair变量,表示,是否公平.
如果为ture,则每个线程会根据到达的顺序执行,而默认是false.

// 业务逻辑实现类

public
 class
 Logic {

    private
 static
 final
 Log log =
 LogFactory.getLog
(
Logic.class
)
;

    private
 AtomicInteger sum =
 new
 AtomicInteger(
0
)
;

    private
 Semaphore sp =
 new
 Semaphore(
5
)
;
 // 吞吐量为5条线程

 
    public
 void
 test(
)
 {

        try
 {

            sp.acquire
(
)
;

            log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " entered"
)
;

            Thread
.sleep
(
2000
)
;

            log.info
(
sum.getAndIncrement
(
)
)
;

            sp.release
(
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
"sleep error:"
, e)
;

        }

    }

}

 
// 线程测试类

public
 class
 RunThread {

    public
 static
 void
 main(
String
[
]
 args)
 {

        final
 Logic logic =
 new
 Logic(
)
;

 
        //定义一个producer

        Runnable
 test =
 new
 Runnable
(
)
 {

            public
 void
 run(
)
 {

                logic.test
(
)
;

            }

        }
;

 
        ExecutorService service =
 Executors.newCachedThreadPool
(
)
;

 
        for
 (
int
 i =
 0
;
 i <
 10
;
 i++
)
 {

            service.submit
(
test)
;

        }

 
        service.shutdown
(
)
;

    }

}

注意; semaphore可以控制某个资源上读取操作的线程数量, 但是, semaphore本身是线程不安全的,
如果资源涉及到写入操作, 那么在操作中加上同步后, 信号量的作用也就跟Lock接口一样了.(一次只能执行一根线程)

8, 利用CyclicBarrier屏障接口实现,线程集合/解散功能

java有好多种的屏障实现, 简单的几种如下:

a, 利用条件变量Condition实现wait-notify机制,等待所有的线程都wait在某一个
集合点时,notifyAll一下. 缺点是需要一根监控线程

b, 利用join方法,开一个监视线程, 每次调用这个线程取被block住的线程数量.
当达到指定数量后, 监视线程自动死亡,以放开所有的被block threads.

c, 利用CyclicBarrier提供的功能,只需要在集合点处调用await()方法,即可.

// 试验屏障功能的类

public
 class
 Logic {

    private
 static
 final
 Log log =
 LogFactory.getLog
(
Logic.class
)
;

    private
 int
 value =
 21
;

    private
 CyclicBarrier cyclic =
 new
 CyclicBarrier(
3
)
;

 
    public
 int
 getValue(
)
 {

        return
 value;

    }

 
    public
 void
 setValue(
int
 value)
 {

        this
.value
 =
 value;

    }

 
    public
 void
 expression1(
)
 {

        try
 {

            Thread
.sleep
(
1000
)
;

            log.info
(
value/
2
)
;

            cyclic.await
(
)
;

            log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " end."
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
e)
;

        }
 catch
 (
BrokenBarrierException e)
 {

            log.error
(
e)
;

        }

    }

 
    public
 void
 expression2(
)
 {

        try
 {

            Thread
.sleep
(
2000
)
;

            log.info
(
value*
2
)
;

            cyclic.await
(
)
;

            log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " end."
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
e)
;

        }
 catch
 (
BrokenBarrierException e)
 {

            log.error
(
e)
;

        }

    }

 
    public
 void
 expression3(
)
 {

        try
 {

            Thread
.sleep
(
3000
)
;

            log.info
(
value+
2
)
;

            cyclic.await
(
)
;

            log.info
(
Thread
.currentThread
(
)
.getName
(
)
 +
 " end."
)
;

        }
 catch
 (
InterruptedException
 e)
 {

            log.error
(
e)
;

        }
 catch
 (
BrokenBarrierException e)
 {

            log.error
(
e)
;

        }

    }

}

 
// 线程测试类

public
 class
 RunThread {

    public
 static
 void
 main(
String
[
]
 args)
 {

        final
 Logic logic =
 new
 Logic(
)
;

        Runnable
 run1 =
 new
 Runnable
(
)
 {

            public
 void
 run(
)
 {

                logic.expression1
(
)
;

            }

        }
;

 
        Runnable
 run2 =
 new
 Runnable
(
)
 {

            public
 void
 run(
)
 {

                logic.expression2
(
)
;

            }

        }
;

 
        Runnable
 run3 =
 new
 Runnable
(
)
 {

            public
 void
 run(
)
 {

                logic.expression3
(
)
;

            }

        }
;

 
        //各产生10个consumer和producer

        ExecutorService service =
 Executors.newCachedThreadPool
(
)
;

 
        service.submit
(
run1)
;

        service.submit
(
run2)
;

        service.submit
(
run3)
;

 
        service.shutdown
(
)
;

    }

}

注意: 使用屏障的时候, 小心异常的放生,当发生异常,所有线程都会被释放
等待中的线程将被中断. 且发生异常的屏障将不可用,需要屏障的实例reset一下.

9, 利用CountDownLatch接口实现线程集合/解散功能,类似CyclicBarrier,区别是倒数且只跑一次

接口方法与CyclicBarrier基本相同,不同在于构造函数需要传入一数量,表示
倒数的开始数量.以后会递减这个值