kotlin线程、线程池、线程安全

kotlin线程、线程池、线程安全

线程

Java中创建线程有2种方式，继承Thread或者实现Runnable并将其传入Thread的构造函数，两种方式都是通过start()来启动线程。同样的，这两种方式在kotlin也适用，并且kotlin提供了一个函数thread()，可以更方便的创建线程，示例如下：

//方式1
class MyThread : Thread() {
    override fun run() {
        super.run()
        println(Thread.currentThread().name)
    }
}
MyThread().start()

//方式1变体
object : Thread() {
    override fun run() {
        println(Thread.currentThread().name)
    }
}.start()


//方式2
val thread = Thread {
    println(Thread.currentThread().name)
}
thread.start()

//方式3，使用kotlin提供的thread函数
thread(name = "kotlin_thread") {
    println(Thread.currentThread().name)
}
复制代码

方式2就是实现Runnable，并将其传入Thread的构造函数 的方式，这里用了lambda表达式，{}内的代码就是Runnable的run()方法。第三种方式使用了kotlin的thread()函数，可以传入name，以供调试使用。这种方式内部其实也是继承了Thread。

线程池

使用线程池同样可以间接创建线程，并且线程池可以对线程进行管理，线程池中的线程可以被重用，从而避免频繁地创建和销毁线程带来的性能消耗。线程池可以有效控制线程的最大并发数量，防止线程数量过多导致占用系统资源过多。

ThreadPoolExecutor

线程池的核心类是ThreadPoolExecutor，ThreadPoolExecutor的构造方法如下

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
}
复制代码

参数解释：

• corePoolSize 核心线程数。
• maximumPoolSize 最大线程数。线程数=核心线程数+非核心线程数。
• keepAliveTime 非核心线程的保活时间，非核心线程空闲时间大于这个保活时间时会被回收。如果调用了allowCoreThreadTimeOut(true)设置允许核心线程超时回收，那么核心线程超时达到保活时间，一样会被回收。
• unit keepAliveTime参数的时间单位。
• workQueue 任务的队列，execute方法提交的Runnable任务在执行之前会保存到这个队里中。
• threadFactory 创建线程的工厂，如果使用重载方法，默认为Executors.defaultThreadFactory()。
• handler 线程池处理不过来时的拒绝回调，所谓的处理不过来就是指队列已满并且达到了最大线程数。这个参数如果使用重载方法，默认为AbortPolicy，也就是直接抛出RejectedExecutionException。

补充说明几点：

1. 核心线程就类似于正式员工，来任务了并且人手不够就招人（未达到corePoolSize时），任务干完了就继续干下一个任务（线程复用），即使闲着也不会把人裁掉（除非设置了allowCoreThreadTimeOut）。非核心线程就类似于临时工，正式员工人数已满，活还是干不完，此时就招临时工，临时工用完了如果闲着没事做就裁掉以节省资金。

2. workQueue，任务的队列会影响到任务的处理速度，workQueue 过大时，workQueue永远不会满，永远不会创建非核心线程，只能依靠核心线程，任务量大时平均响应时间会变长。workQueue 过小时，核心线程满了之后会创建非核心线程，并发线程数高了，单位时间消耗系统资源也增加了，处理速度会变快，但是如果任务量继续增加，由于队列太小无法容纳，就会导致任务被拒。根据任务类型合理的取舍，合理的安排workQueue、corePoolSize、maximumPoolSize可以尽量避免这些问题。jdk提供了一些常用的workQueue，包括：SynchronousQueue、LinkedBlockingQueue、ArrayBlockingQueue。

   • SynchronousQueue 没有容量，是无缓冲等待队列，是一个不存储任务的阻塞队列，会直接尝试创建新的线程执行任务，必须等队列中添加的任务被移除够才能添加新的任务。使用SynchronousQueue一般要求maximumPoolSize 为无限制(Integer.MAX_VALUE)，避免线程池拒绝执行操作。
   • LinkedBlockingQueue 容量为Integer.MAX_VALUE的队列，由于容量超大(可以称之为无限)，队列永远不会满，所以maximumPoolSize 永远不会起作用，也永远不会发生线程池拒绝执行的情况。LinkedBlockingQueue 也可以传入一个值指定他的容量上限。
   • ArrayBlockingQueue，指定上限的队列，公平策略下，按照队列的先进先出处理任务。非公平策略下，不会按照先进先出的规则来处理任务。

3. handler，线程池处理不过来时的拒绝回调，JDK提供了几种默认实现，AbortPolicy、CallerRunsPolicy、DiscardPolicy、DiscardOldestPolicy。

   • AbortPolicy 直接抛出RejectedExecutionException异常。

   • CallerRunsPolicy 在execute方法的调用线程中运行被拒绝的任务，如果线程池被关闭了，就丢弃任务。

   • DiscardPolicy 什么也不做，默默地丢弃被拒绝的任务。

   • DiscardOldestPolicy 丢弃最早加入队列中的任务。

     也可以自己实现RejectedExecutionHandler，在rejectedExecution中处理拒绝的情况。

4. 使用shutdown()可以等线程池中已有任务执行完成后关闭线程池，调用后不再接受新的任务。使用prestartCoreThread()提前启动一个空闲的核心线程。使用prestartAllCoreThreads()提前启动所有核心线程。

线程池流程

线程池任务提交流程如下图所示：

关于线程池任务提交流程可以参考

mp.weixin.qq.com/s/OKTW_mZnN…

工厂方法

虽然JDK中提供了ThreadPoolExecutor的构造方法，但是官方建议使用如下几种更方便的Executors工厂方法：

• Executors.newCachedThreadPool 无上限线程池，所有线程都是非核心线程，线程数量上限为Integer.MAX_VALUE，线程空闲60秒会被回收，所以长时间没有任务，这种方式创建的线程池将不会耗费系统资源。等待队列是SynchronousQueue，也就是说所有的任务，提交后立即被执行。这种方式适合大量的耗时小的任务。

• Executors.newFixedThreadPool(int nThreads) 固定大小线程池，线程池中全部是核心线程，并且等待队列无上限。

• Executors.newSingleThreadExecutor 单线程模式，只有一个核心线程，等待队列无上限。保证了所有任务按照队列依次执行，不存在线程同步问题。

• Executors.newScheduledThreadPool(int corePoolSize) 有数量固定的核心线程，且有数量无限多的非核心线程，非核心线程空闲保活时间是10秒。这类线程池适合用于执行定时任务和固定周期的重复任务。

线程安全

多线程同时访问一个共享数据，不管运行时线程以何种顺序运行，总是能得到预期的结果，我们就称其为线程安全的。举个反面例子：

var value = 0

@JvmStatic
fun main(args: Array<String>) {
    val pool = Executors.newCachedThreadPool()

    for (index in 1..100) {
        pool.execute {
            value ++
        }
    }

    pool.shutdown()
    while (!pool.isTerminated){}
    println(value)
}
复制代码

100个线程，每个线程中执行++，执行完成后，预期结果是100，然而实际结果并不是100，而且每次结果都不一样。以上现象就是非线程安全的。为什么会出现这种情况呢，因为++操作可以认为是三个操作，也就是先取值，再+1，然后赋值回去。如果线程1，先取到的值为10，还没有进行+1并赋值回去时，另一个线程已经将value值改为了11，按照预期，线程1此时应该在11的基础上做运算，然而线程1取到的值是10，运算后赋值回去的也是11，这就出现调用两次++，最终结果只加了1，如此错误积累下去，最终肯定无法得到预期值。导致这个问题的原因是多个线程以一种会引起冲突的方式访问共享数据，这被称之为竞争状态。如果一个类的对象在多线程中不会导致竞争状态，那就可以称其为线程安全的。

那如何实现线程安全呢？或者说如何消除多线程中的竞争状态呢？主要有两种方式：锁、原子操作

锁

经常使用synchronized来加锁，获取到锁的线程才能执行，其他线程只能被阻塞。需要注意的是**synchronized方法锁住的是当前对象;static synchronized方法锁住的是当前类;synchronized(lock)代码块锁住的是括号中的变量;**

示例:

class KotlinThreadTest {
    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            for (index in 1..5) {
                thread {
                    TestSync().test1(index)
                }
            }
        }
	}
}

class TestSync {
    @Synchronized
    fun test1(num: Int) {
        println("开始，$num")
        Thread.sleep(10)
        println("结束，$num")
    }

    private fun test2(num: Int) {
        synchronized(this) {
            println("开始，$num")
            Thread.sleep(10)
            println("结束，$num")
        }
    }
}
复制代码

main方法中生成了5个线程，每个线程都会创建一个TestSync，并调用它的test1()。kotlin中的@Synchronized相当于java中的synchronized方法。最终输出：

开始，4
开始，5
开始，2
开始，1
开始，3
结束，4
结束，2
结束，3
结束，1
结束，5
复制代码

并没有按照预期输出，锁并不起作用。因为Java synchronized方法(对应kotlin中带有@Synchronized的函数)锁住的是当前对象，也就是说test1和test2是相同的。而调用时每个线程都实例化了一个TestSync对象，每个线程要获得的锁都不一样，当然是没法锁住的。改为如下代码：

class KotlinThreadTest {
    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
			TestSync().mainTest()
        }
	}
}

class TestSync {
    fun mainTest() {
        for (index in 1..5) {
            thread {
                test1(index)
            }
        }
    }

    @Synchronized
    fun test1(num: Int) {
        println("开始，$num")
        Thread.sleep(10)
        println("结束，$num")
    }

    private fun test2(num: Int) {
        synchronized(this) {
            println("开始，$num")
            Thread.sleep(10)
            println("结束，$num")
        }
    }
}
复制代码

输出

开始，1
结束，1
开始，3
结束，3
开始，4
结束，4
开始，5
结束，5
开始，2
结束，2
复制代码

可以看到，总是一个线程结束之后，另一个线程才开始，证明锁起作用了。这里是在TestSync内部调用自己的方法，只有一个TestSync对象，多个线程竞争的是同一个对象，所以锁就起作用了。

以上代码证明了：Java synchronized方法(对应kotlin中带有@Synchronized的函数)锁住的是当前对象；synchronized(lock)代码块锁住的是括号中的变量

Java中static synchronized方法，在kotlin中写法是给fun加上@Synchronized和@JvmStatic注解，并且要放在companion object中。例如如下代码：

class TestSync {
    companion object {
        @Synchronized
        @JvmStatic
        fun test(num: Int) {
            println("开始，$num")
            Thread.sleep(10)
            println("结束，$num")
        }
    }
}

class KotlinThreadTest {
    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            for (index in 1..5) {
                thread {
                    TestSync.test(index)
                }
            }
        }
    }
}
复制代码

输出结果不贴了，也是符合预期的。如果将test方法改为如下代码，依旧是符合预期的

class TestSync {
    fun test(num: Int) {
        synchronized(javaClass){
            println("开始，$num")
            Thread.sleep(10)
            println("结束，$num")
        }
    }
}
复制代码

因为static synchronized方法锁住的是当前class，当前class只会有一个，所以上述两种代码效果一样，都可以加锁消除竞争状态。

上方的示例中，虽然锁起作用了，但是打印顺序并不是从1到5，是因为synchronized是非公平锁，如果想按照从1到5的顺序，需要公平锁，所谓公平锁就是多个线程按照申请锁的顺序去获得锁，线程会进⼊队列去排队等待锁，永远都是队列的第⼀位才能获得锁，也就是等待时间最长的线程会获得锁。要实现公平锁，可以使用ReentrantLock，示例如下：

class TestSync {
    companion object {
        private val lock = ReentrantLock(true)
        fun testReentrantLock(num: Int) {
            //使用ReentrantLock时最好使用try-catch-finally代码块，并在finally代码块里释放锁，
            // 这样可以避免由于异常导致释放锁代码没执行，其他线程获取不到锁。
            lock.lock()
            try {
                println("${Thread.currentThread().name}开始，$num")
                Thread.sleep(10)
                println("${Thread.currentThread().name}结束，$num")
            } catch (e: Exception) {

            } finally {
                lock.unlock()
            }
        }
    }
}

class KotlinThreadTest {
    companion object {
        @JvmStatic
        fun main(args: Array<String>) {
            for (index in 1..5) {
                thread(name = "thread-$index") {
                    TestSync.testReentrantLock(index)
                }
                Thread.sleep(5)
            }
        }
    }
}
复制代码

打印结果为：

thread-1开始，1
thread-1结束，1
thread-2开始，2
thread-2结束，2
thread-3开始，3
thread-3结束，3
thread-4开始，4
thread-4结束，4
thread-5开始，5
thread-5结束，5
复制代码

测试的main函数中，每隔5毫秒创建一个线程，指定了线程的名字，并在线程中调用testReentrantLock()函数。testReentrantLock中先锁定，然后执行代码，然后释放锁。释放之后，等待时间最长的（也就是最早创建的线程）将会获得锁，所以就会按照创建时间的先后顺序依次执行。main方法的for循环中增加sleep(5)就是为了更好的重现这个现象。需要注意的是，使用ReentrantLock，最好使用try-catch-finally代码块，并在finally代码块里释放锁，这样可以避免由于异常导致释放锁代码没执行，其他线程获取不到锁的问题

原子操作

线程安全一开始的那个例子，由于++是三个操作，所以导致了线程交替执行时会出现线程安全问题。那如果++是一个不可分割的操作，那不就没有问题了？Java中不可分割的操作就是原子操作，原子操作执行时不会被打断，所以可以简单理解为不可分割。使用Atomic开头的类来完成原子操作。示例如下：

var atomicValue = AtomicInteger(0)

@JvmStatic
fun main(args: Array<String>) {
    val pool = Executors.newCachedThreadPool()

    for (index in 1..100) {
        pool.execute {
            atomicValue.getAndAdd(1)
        }
    }

    pool.shutdown()
    while (!pool.isTerminated) {
    }
    println(atomicValue)
}
复制代码

运行后，输出100。AtomicInteger的getAndAdd方法会以原子的方式获取值并做加法运算，这个过程不会被打断，所以是线程安全的。AtomicInteger还提供了getAndUpdate、updateAndGet等方法用于原子的设置、更新值。JDK提供了AtomicInteger、AtomicBoolean、AtomicLong用于原子操作基本类型，提供了AtomicIntegerArray、AtomicLongArray、AtomicReference用于原子操作数组和引用类型，原理想通，此处不再过多延伸。

其他方式

还有其他方式来达到线程安全，例如：将变量作用域限制在对象内部，也就是多个线程不要共享一个变量，这样肯定就没有线程安全问题了。还可以使用将ThreadLocal 数据存放在每个线程自己的作用域内，多个线程互不影响，这样也可以实现线程安全。

需要额外说明的是只使用volatile关键字不能实现线程安全，一个共享变量（类的成员变量、类的静态成员变量）被volatile修饰之后，它具有以下特性：

1. 保证了不同线程对这个变量进行操作时的可见性，即一个线程修改了某个变量的值，这新值对其他线程来说是立即可见的。
2. 禁止进行指令重排序。

volatile不能保证原子性，所以一样会存在线程安全问题。

更多volatile的知识，建议参考www.cnblogs.com/dolphin0520…