线程安全与实现方法

线程安全

线程安全的定义

当多个线程同时访问一个对象时，如果不用考虑这些线程在运行时环境下的调度和交替执行，也不用进行额外的同步，或者在调用方进行任何其他的协调操作，调用这个对象的行为都可以获得正确的结果，那就称这个对象是线程安全的。

线程安全的分类

• 不可变

  只要一个不可变的对象被正确地构建出来（没有发生this引用逃逸），那其外部的可见状态永远不会改变，永远都不会看到它在多个线程之中处于不一致的状态。对于一个基本数据类型，在定义时使用final关键字修饰它就可以保证它是不可变的。

• 绝对线程安全

  不管运行时环境如何，调用者都不需要任何额外的同步措施。

• 相对线程安全

  通常意义上的线程安全，它需要保证这个对象单次的操作是线程安全的，在调用的时候不需要进行额外的保障措施，但是对于一些特定顺序的连续调用，需要在调用端使用额外的同步措施来保证调用的正确性。

• 线程兼容

  指对象本身并不是线程安全的，但是可以通过在调用端正确地使用同步措施来保证在并发环境中可以安全地使用。

• 线程对立

  指不管调用端是否采用了同步措施，都无法在多线程环境中并发使用代码。

补充：this引用逃逸

逃逸分析针对的是内存逃逸，当一个对象在方法体定义后，通过一些方式在其他地方被引用，可能导致GC时，无法立即回收，从而造成内存逃逸。如果被外部方法引用，譬如作为调用参数传递到其他方法中，这种称为方法逃逸；如果被外部线程访问到，例如赋值给可以在其他线程中访问的实例变量，这种称为线程逃逸。

什么是this引用逃逸

在构造器还未完成前（实例初始化完成之前），将自身this引用向外抛出并被其他线程复制使用了，可能会导致其他线程访问到"初始化了一半的对象"（即尚未初始化完成的对象），会造成影响。

逃逸场景 场景一

在构造函数中启动了新的线程（新的线程拥有this引用）

import java.text.SimpleDateFormat;
import java.util.Date;

public class EscapeForThis {

    int a;
    int b = 0;

    public EscapeForThis() {
        a = 1;
        // 在构造函数中新建一个线程（拥有this引用），访问成员变量
        new Thread(new Runnable() {

            @Override
            public void run() {
                System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + a + ",b=" + b);
            }
        }).run();
        b = 1;
    }

    public static void main(String[] args) {
        EscapeForThis s = new EscapeForThis();
        System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + s.a + ",b=" + s.b);
    }
}

执行结果s

[12:04:55--new] a=1,b=0
[12:04:55--main] a=1,b=1

新建的线程在访问b的时候b尚未完成初始化，没有访问到正确的数据。

场景二

在构造器中内部类使用外部类：内部类可以无条件的访问外部类(自动持有外部类的this引用)，当内部类发布出去后，即外部类的this引用也发布出去了，此时无法保证外部类已经初始化完成。

外部类EscapeForThis，内部类EventListener

/**
 * 事件监听器接口，调用事件处理器
 */
public interface EventListener {
    /**
     * 事件处理方法
     */
    void doEvent(Object obj);
}

demo

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class EscapeForThis {

  	private final int a;
    private final String b;

    private EscapeForThis(EventSource source) throws InterruptedException {
        a = 1;
        source.registerListener(o -> System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + a + ",b=" + EscapeForThis.this.b));
        // 为了演示效果
        TimeUnit.SECONDS.sleep(2);
        b = "xx";
    }

    static class EventSource {
        private final List eventListeners;

        public EventSource() {
            eventListeners = new ArrayList();
        }

        /** 注册监听器*/
        public synchronized void registerListener(T eventListener) {  //数组持有传入对象的引用
            this.eventListeners.add(eventListener);
            this.notifyAll();
        }

        /** 获取事件监听器*/
        public synchronized List retrieveListeners() throws InterruptedException {  //获取持有对象引用的数组
            List dest = null;
            if (eventListeners.size()  {
            monitor.writer("Thread 1");
        }).start();

        new Thread(() -> {
            monitor.reader("Thread 2");
        }).start();
    }
}

非阻塞同步

采用乐观并发策略的同步措施，先操作，若遇到冲突则进行补偿操作。

依赖硬件指令集，常见指令：

• 测试并设置(Test-and-Set)
• 获取并增加(Fetch-and-Increment)
• 交换(Swap)
• 比较并交换(Compare-and-Swap)，即常用的CAS
• 加载连接/条件存储(Load-Linked/Store-Conditional)

在JDK中使用Unsafe类中使用了CAS。

实现举例

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class AtomicTest {

    private static final int THREAD_COUNT = 20;

    private static final CountDownLatch COUNT_DOWN_LATCH = new CountDownLatch(THREAD_COUNT);

    public static final AtomicInteger RACE = new AtomicInteger(0);

    private static final ThreadPoolExecutor POOL_EXECUTOR = initThreadPool(THREAD_COUNT, THREAD_COUNT, 1000);

    /**
     * 工作线程
     */
    public static class WorkerThreadFactory implements ThreadFactory {
        private final String namePrefix;
        private final AtomicInteger nextId = new AtomicInteger(1);

        WorkerThreadFactory(String whatFeatureOfGroup) {
            this.namePrefix = "From WorkerThreadFactory's " + whatFeatureOfGroup + "-Worker-";
        }

        @Override
        public Thread newThread(Runnable task) {
            String name = namePrefix + nextId.getAndIncrement();
            return new Thread(null, task, name, 0);
        }
    }

    /**
     * 初始化线程池
     */
    public static ThreadPoolExecutor initThreadPool(int corePoolSize, int maxPoolSize, long keepAliveTime) {
        return new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                keepAliveTime,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue(1000),
                new WorkerThreadFactory("AtomicTest"),
                new ThreadPoolExecutor.AbortPolicy());
    }

    public static void increase() {
        // incrementAndGet内部使用的sun.misc.Unsafe的compareAndSwapInt实现的
        RACE.incrementAndGet();
    }

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

        for (int i = 0; i  {
                try {
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println("[" + new SimpleDateFormat("HH:mm:ss.SSS").format(new Date()) + "--" + Thread.currentThread().getName() + "] " + finalI + "start ...");
                    for (int j = 0; j