当前位置：网站首页>线程安全与实现方法

线程安全与实现方法

2022-06-24 06:32:00 【顧棟】

线程安全

文章目录

线程安全的定义

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

线程安全的分类

  • 不可变

    只要一个不可变的对象被正确地构建出来（没有发生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<EventListener> 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<T> {
    
        private final List<T> eventListeners;

        public EventSource() {
    
            eventListeners = new ArrayList<T>();
        }

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

        /** 获取事件监听器*/
        public synchronized List<T> retrieveListeners() throws InterruptedException {
      //获取持有对象引用的数组
            List<T> dest = null;
            if (eventListeners.size() <= 0) {
    
                this.wait();
            }
            dest = new ArrayList<T>(eventListeners.size());  //这里为什么要创建新数组，好处在哪里？防止对象逃逸，同时减少对eventListeners的竞争
            dest.addAll(eventListeners);
            return dest;
        }
    }

    /**新建一个执行线程上*/
    static class ListenerRunnable implements Runnable {
    
        private final EventSource<EventListener> source;

        public ListenerRunnable(EventSource<EventListener> source) {
    
            this.source = source;
        }

        @Override
        public void run() {
    
            List<EventListener> listeners = null;

            try {
    
                // 获取事件监听器
                listeners = this.source.retrieveListeners();
            } catch (InterruptedException e) {
    
                e.printStackTrace();
            }
            assert listeners != null;
            for (EventListener listener : listeners) {
    
                listener.doEvent(new Object());  //执行内部类获取外部类的成员变量的方法
            }
        }
    }
    
    public static void main(String[] args) throws InterruptedException {
    
        EventSource<EventListener> s = new EventSource<>();
        ListenerRunnable runnable = new ListenerRunnable(s);
        Thread thread = new Thread(runnable);
        thread.start();
        EscapeForThis escapeForThis = new EscapeForThis(s);
        System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + escapeForThis.a + ",b=" + escapeForThis.b);
    }

}

执行结果

[15:34:39--Thread-0] a=1,b=null
[15:34:41--main] a=1,b=xx

在外部类中对成员变量进行初始化的时候 休眠了2s，是为了凸显this引用逃逸的效果。

形成this引用逃逸的条件

  • 一个是在构造函数中创建内部类，并且在构造函数中就把这个内部类给发布了出去。可以通过破外条件才避免this引用逃逸。

解决方案

  1. 不要在构造函数中启动线程，可以单独一个方法在外部启动线程。

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

public class EscapeForThis {
      

    int a;
    int b = 0;

	private Thread t;
    public EscapeForThis() {
      
        a = 1;
        // 在构造函数中新建一个线程（拥有this引用），访问成员变量
        t= 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);
            }
        });
        b = 1;
    }
    
    public void initStart() {
      
        t.start();
    }
    
    public static void main(String[] args) {
      
        EscapeForThis s = new EscapeForThis();
        s.initStart();
        System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + s.a + ",b=" + s.b);
    }
}

  2. 不要在外部类初始化完成之前，发布内部类。

    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 final EventListener listener;

    private EscapeForThis(){
      
        a = 1;
        listener = new EventListener(){
      
            @Override
            public void doEvent(Object obj) {
      
                System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + a + ",b=" + EscapeForThis.this.b);
            }
        };
        b = "xx";
    }

    public static EscapeForThis getInstance(EventSource<EventListener> source) {
      
        EscapeForThis safe = new EscapeForThis();  //先初始化
        source.registerListener(safe.listener);  //发布内部类
        return safe;
    }

    static class EventSource<T> {
      
        private final List<T> eventListeners;

        public EventSource() {
      
            eventListeners = new ArrayList<T>();
        }

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

        /** 获取事件监听器*/
        public synchronized List<T> retrieveListeners() throws InterruptedException {
        //获取持有对象引用的数组
            List<T> dest = null;
            if (eventListeners.size() <= 0) {
      
                this.wait();
            }
            dest = new ArrayList<T>(eventListeners.size());  //这里为什么要创建新数组，好处在哪里？防止对象逃逸，同时减少对eventListeners的竞争
            dest.addAll(eventListeners);
            return dest;
        }
    }

    /**新建一个执行线程上*/
    static class ListenerRunnable implements Runnable {
      
        private final EventSource<EventListener> source;

        public ListenerRunnable(EventSource<EventListener> source) {
      
            this.source = source;
        }

        @Override
        public void run() {
      
            List<EventListener> listeners = null;

            try {
      
                // 获取事件监听器
                listeners = this.source.retrieveListeners();
            } catch (InterruptedException e) {
      
                e.printStackTrace();
            }
            assert listeners != null;
            for (EventListener listener : listeners) {
      
                listener.doEvent(new Object());  //执行内部类获取外部类的成员变量的方法
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
      
        EventSource<EventListener> s = new EventSource<>();
        ListenerRunnable runnable = new ListenerRunnable(s);
        Thread thread = new Thread(runnable);
        thread.start();
        EscapeForThis escapeForThis = EscapeForThis.getInstance(s);
        System.out.println("[" + new SimpleDateFormat("HH:mm:ss").format(new Date()) + "--" + Thread.currentThread().getName() + "] a=" + escapeForThis.a + ",b=" + escapeForThis.b);
    }

}

线程安全的实现

互斥同步（阻塞同步）

采用悲观并发策略的同步措施，无论是否出现竞争，都按加锁操作。

同步是指在多个线程并发访问数据时，保证共享数据在同一时刻只被一条线程使用。互斥是实现同步的一种手段，临界区、互斥量、信号量都是常见的互斥实现方式。

同步手段就是synchronized关键字、Lock接口。

线程阻塞和唤醒需要额外的性能开销。

实现举例

public class Monitor {
    

    private int a = 0;

    public synchronized void writer(String s) {
         // 1
        System.out.println(s);                      // 2
        a++;                                        // 3
    }                                               // 4

    public synchronized void reader(String s) {
        // 5
        System.out.println(s);                     // 6
        int i = a;                                 // 7
        System.out.println(i);                     // 8
    }                                              // 9
}
public class Client {
    

    public static void main(String[] args) {
    
        Monitor monitor = new Monitor();

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

        new Thread(() -> {
    
            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 < THREAD_COUNT; i++) {
    
            int finalI = i;
            POOL_EXECUTOR.execute(() -> {
    
                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 < 20000; j++) {
    
                        increase();
                    }
                    System.out.println("[" + new SimpleDateFormat("HH:mm:ss:sss").format(new Date()) + "--" + Thread.currentThread().getName() + "] " + finalI + "end ...");
                } catch (InterruptedException e) {
    
                    e.printStackTrace();
                } finally {
    
                    COUNT_DOWN_LATCH.countDown();
                }
            });
        }
        if (COUNT_DOWN_LATCH.await(2, TimeUnit.MINUTES)) {
    
            System.out.println("[" + new SimpleDateFormat("HH:mm:ss:sss").format(new Date()) + "--" + Thread.currentThread().getName() + "] RACE = " + RACE);
        }

        POOL_EXECUTOR.shutdown();
    }
}

执行结果

[17:28:12:012--From WorkerThreadFactory's AtomicTest-Worker-18] 17start ...
[17:28:12:012--From WorkerThreadFactory's AtomicTest-Worker-2] 1start ...
...
[17:28:12:012--From WorkerThreadFactory's AtomicTest-Worker-2] 1end ....
[17:28:12:012--From WorkerThreadFactory's AtomicTest-Worker-18] 17end ...
...
[17:28:12:012--main] RACE=400000

无同步方案

若一个方法不涉及共享数据，就不需要同步措施去保证其正确性。

可重入代码，可以在代码执行的任何时刻中断它，转而去执行另一端代码（包含递归），而在控制权返回后，原来的程序不会出现任何错误，也不对结果有影响。

线程本地存储（Thread Local Storage）：如果一段代码中所需要的数据必须与其他代码共享，那就看看这些共享数据的代码是否保证在同一个线程中执行。如果能保证，我们就可以把共享数据的可见范围限制在同一个线程之内，就无需同步也能保证线程之间不出现数据争用的问题。实现类ThreadLocal。

参考文章

《深入理解JAVA虚拟机》第三版 周志明版

原网站

版权声明
本文为[顧棟]所创，转载请带上原文链接，感谢
https://blog.csdn.net/weixin_43820556/article/details/125431643

边栏推荐

猜你喜欢

随机推荐