程序員：最近一直在思考，線程安全與性能的權衡

線程安全

如果多線程情況下使用這個類，無論多線程如何使用和調度這個類，這個類總是表示出正確的行為，這個類就是線程安全的。

類的線程安全表現為：

操作的原子性，類似數據庫事務。

內存的可見性，當前線程修改後其他線程立馬可看到。

不做正確的同步，在多個線程之間共享狀態的時候，就會出現線程不安全。

安全策略

1. 棧封閉

所有的變量都是在方法內部聲明的，這些變量都處於棧封閉狀態。方法調用的時候會有一個棧楨，這是一個獨立的空間。在這個獨立空間創建跟使用則絕對是安全的，但是注意不要返回該變量哦！

<code>\tpublic  void JustTest(){
\t\tInteger a = 12;
\t\tUser user = new User(12);
\t}/<code>

2. 無狀態

說白了就是這個類沒有任何成員變量，只有一堆成員函數，這樣絕對是安全的。

<code>public class StatelessClass {
\t 

\tpublic int service(int a,int b) {
\t\treturn a*b;
\t}
\t//...public void t(){}
}/<code>

3. 類不可變

Java中不管是String對象跟基本類型裝箱後的對象都是不可變的，說白了就是都帶有final。讓狀態不可變，兩種方式：

加final關鍵字，對於一個類，所有的成員變量應該是私有的，同樣的只要有可能，所有的成員變量應該加上final關鍵字，但是加上final，要注意如果成員變量又是一個對象時，這個對象所對應的類也要是不可變，才能保證整個類是不可變的。

根本就不提供任何可供修改成員變量的地方，同時成員變量也不作為方法的返回值,說白了就是不提供任何set方法。

<code>public class ImmutableFinalRef {
\t
\tprivate final int a;
\tprivate final int b;
\tprivate final User user; //這裡，就不能保證線程安全啦
\t
\tpublic ImmutableFinalRef(int a, int b) {
\t\tsuper();
\t\tthis.a = a;
\t\tthis.b = b;
\t\tthis.user = new User(2);
\t} 

\tpublic int getA() {
\t\treturn a;
\t}
\tpublic int getB() {
\t\treturn b;
\t}
\t
\tpublic User getUser() {
\t\treturn user;
\t}
\tpublic static class User{
\t\tprivate int age;
\t\t//private final int age;
\t\tpublic User(int age) {
\t\t\tsuper();
\t\t\tthis.age = age;
\t\t}
\t\tpublic int getAge() {return age;}
\t\tpublic void setAge(int age) {this.age = age;}
\t}
\tpublic static void main(String[] args) {
\t\tImmutableFinalRef ref = new ImmutableFinalRef(12,23);
\t\tUser u = ref.getUser();
\t\tu.setAge(35);// 對象不可變不過對象裡的數據也要不可變才可以！
\t}
}/<code>

volatile

保證類的可見性，最適合一個線程寫，多個線程讀的情景，再重複一遍 只保證變量的可見性！用volatile修飾的變量在get的時候多線程情況下不用加鎖，保證可見性。但是在set的時候要加鎖或者通過CAS操作進行變化。

比如ConcurrentHashMap。

加鎖跟CAS

前面的若干章節都寫過 對於操作變量用syn,lock,CAS。

安全發佈

類中持有的成員變量，特別是對象的引用，如果這個成員對象不是線程安全的，通過get等方法發佈出去(return出去)，會造成這個成員對象本身持有的數據在多線程下不正確的修改，從而造成整個類線程不安全的問題。

解決方法：用concurrentLinkedQueue等線程安全容器或者返回一個副本。

<code>public class UnsafePublish {
\t//要麼用線程的容器替換，要麼發佈出去的時候，提供副本，深度拷貝
\tprivate List<integer> list =  new ArrayList<>(3);
\tpublic UnsafePublish() {
\t\tlist.add(1);
\t\tlist.add(2);
\t\tlist.add(3);
\t}
\t//將list不安全的發佈出去了
\tpublic List<integer> getList() {return list;}
\t//也是安全的,加了鎖
\tpublic synchronized int getList(int index) {
\t\treturn list.get(index);
\t}
\tpublic synchronized void set(int index,int val) {
\t\tlist.set(index, val);
\t}
}/<integer>/<integer>/<code>

TheadLocal

底層類似跟一個HashMap一樣簡單理解，key = 線程，value就是當前線程使用的變量。

死鎖

競爭的資源一定是多於1個，同時小於等於競爭的線程數，資源只有一個，只會產生激烈的競爭。

死鎖的根本成因：獲取鎖的順序不一致導致。

<code>public class NormalDeadLock {
    private static Object valueFirst = new Object();//第一個鎖
    private static Object valueSecond = new Object();//第二個鎖
    //先拿第一個鎖，再拿第二個鎖
    private static void fisrtToSecond() throws InterruptedException {
        String threadName = Thread.currentThread().getName();
        synchronized (valueFirst) {
            System.out.println(threadName + " 獲得第一個");
            TimeUnit.MILLISECONDS.sleep(100);
            synchronized (valueSecond) {
                System.out.println(threadName + " 獲得第二個");
            }
        }
    }
    //先拿第二個鎖，再拿第一個鎖
    private static void SecondToFisrt() throws InterruptedException {
        String threadName = Thread.currentThread().getName();
        synchronized (valueSecond) {
            System.out.println(threadName + " 獲得第一個");
            TimeUnit.MILLISECONDS.sleep(100);
            synchronized (valueFirst) {
                System.out.println(threadName + " 獲得第二個");
            }
        }
    }
    //執行先拿第二個鎖，再拿第一個鎖 

    private static class TestThread extends Thread {
        private String name;
        public TestThread(String name) {
            this.name = name;
        }
        public void run() {
            Thread.currentThread().setName(name);
            try {
                SecondToFisrt();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        Thread.currentThread().setName("TestDeadLock");
        TestThread testThread = new TestThread("SubTestThread");
        testThread.start();
        try {
            fisrtToSecond();//先拿第一個鎖，再拿第二個鎖
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}/<code>

如果懷疑發送死鎖：

簡單來說就是 甲拿著A鎖，獲取B鎖，乙拿著B鎖獲取A鎖，注意在甲乙獲得第一個鎖的時候休眠會兒，來製造死鎖。

解決方法：保證加鎖的順序性。

設定為鎖定AB的時候，總是先鎖小的再鎖大的。可以通過甲乙的唯一ID活著通過System自帶的獲得ID函數System.identityHashCode();

<code>public class SafeOperate implements ITransfer { 

\tprivate static Object tieLock = new Object();//加時賽鎖
    @Override
    public void transfer(UserAccount from, UserAccount to, int amount)
            throws InterruptedException {
    \t
    \tint fromHash = System.identityHashCode(from);
    \tint toHash = System.identityHashCode(to);
    \t// 或者你可以保證 ID唯一可以用ID實現
    \t//先鎖hash小的那個
    \tif(fromHash<tohash>            synchronized (from){
                System.out.println(Thread.currentThread().getName() +" get"+from.getName());
                Thread.sleep(100);
                synchronized (to){
                    System.out.println(Thread.currentThread().getName() +" get"+to.getName());
                    from.flyMoney(amount);
                    to.addMoney(amount);
                }
            }    \t\t
    \t}else if(toHash<fromhash>            synchronized (to){
                System.out.println(Thread.currentThread().getName() +" get"+to.getName());
                Thread.sleep(100);
                synchronized (from){
                    System.out.println(Thread.currentThread().getName() +" get"+from.getName());
                    from.flyMoney(amount);
                    to.addMoney(amount);
                }
            }    \t\t
    \t}else {//解決hash衝突的方法
    \t\tsynchronized (tieLock) { //那個線程拿到再處理
\t\t\t\tsynchronized (from) {
\t\t\t\t\tsynchronized (to) {
\t                    from.flyMoney(amount);
\t                    to.addMoney(amount);\t\t\t\t\t\t
\t\t\t\t\t}
\t\t\t\t}
\t\t\t}
    \t}
    }
}/<fromhash>/<tohash>/<code>

通過tryLock

核心思路就是while死循環獲得兩個鎖，都獲得才可以進行操作然後break。

<code>    public void transfer(UserAccount from, UserAccount to, int amount)
            throws InterruptedException {
    \tRandom r = new Random();
    \twhile(true) {
    \t\tif(from.getLock().tryLock()) {
    \t\t\ttry {
    \t\t\t\tSystem.out.println(Thread.currentThread().getName() +" get "+from.getName());
    \t\t\t\tif(to.getLock().tryLock()) {
    \t\t\t\t\ttry {
    \t    \t\t\t\tSystem.out.println(Thread.currentThread().getName() +" get "+to.getName());
    \t\t\t\t\t\t//兩把鎖都拿到了
    \t                    from.flyMoney(amount);
    \t                    to.addMoney(amount);
    \t                    break;
    \t\t\t\t\t}finally {
    \t\t\t\t\t\tto.getLock().unlock();
    \t\t\t\t\t}
    \t\t\t\t}
    \t\t\t}finally {
    \t\t\t\tfrom.getLock().unlock();
    \t\t\t}
    \t\t}
    \t\tSleepTools.ms(r.nextInt(10)); // 防止發生活鎖！
    \t}
    }/<code>

活鎖

嘗試拿鎖的機制中，發生多個線程之間互相謙讓，不斷髮生拿鎖，釋放鎖的過程。

比如上面的活鎖代碼while循環如果沒有時間休眠的話，由於JDK線程獲得鎖是謙讓式獲得的，可能出現如下：

甲拿到A嘗試拿B，拿B失敗了再重新嘗試拿A，再重新拿B，這樣週而復始的嘗試。

乙拿到B嘗試拿A，拿A失敗了再重新嘗試拿B，再重新拿A，這樣週而復始的嘗試。

解決辦法：把對象加鎖順序的不確定性變成確定性的順序。

解決：

通過內在排序，保證加鎖的順序性

通過嘗試拿鎖配合休眠若干也可以。

線程飢餓

飢餓：線程因無法訪問所需資源而無法執行下去的情況。

不患寡，而患不均，如果線程優先級“不均”，在CPU繁忙的情況下，優先級低的線程得到執行的機會很小，就可能發生線程飢餓；持有鎖的線程，如果執行的時間過長，也可能導致飢餓問題。

解決方法：

保證資源充足

公平的分配資源

防止持有鎖的線程長時間執行。

性能

多線程是好但是要切記勿裝逼強行使用，裝逼必被打。我們使用多線程的出發點是要了解，是為了提供系統的性能，充分利用系統資源。但是引入多線程後會引入額外的開銷。

衡量應用程序性能一般：服務時間、延遲時間、吞吐量、可伸縮性，深入瞭解性能優化。

做應用的時候：

先保證程序的正確性跟健壯性，確實達不到性能要求再想如何提速。

一定要以測試為基準。

一個程序中串行的部分永遠是有的.

裝逼利器：阿姆達爾定律 S=1/(1-a+a/n)

系統中某一部件因為採用更快的實現後，整個系統性能的提高與該部分的使用頻率或者在總運行時間中比例有關。直觀地，你把一個部件提升了很多，但是這個部件卻不經常使用，因此這種提高看上去是提高其實並沒有。所以Amdahl定律認為我們除了需要關注部件的加速比，還要關注該部件的使用頻率/情況。

影響因素

多線程的上下文切換，不是多線程一定好。比如只有一個核心，讓你做語數外三門作業，如果你順序做是可以的，這樣不會涉及到任何作業場景的佈置環境切換，而如果你非要同時做三門作業那麼就會來回切換了，反正耗時！聯想到多線程的上下文切換同樣如此，CPU切換一個上下文就是幾微妙哦！線程池的設置思想跟這個類似。

內存同步加鎖等操作在編譯代碼後都有疊加指令存在的。

一些線程獲得鎖失敗了還會進行阻塞式的等待。

常用的思路一般也是如下幾點。

縮小鎖的範圍

能用方法塊儘量不要鎖函數

<code>\tprivate Map<string> matchMap = new HashMap<>();
\t
\tpublic synchronized boolean isMatch(String name,String regexp) { // 太大
\t\tString key = "user."+name;
\t\tString job = matchMap.get(key);
\t\tif(job == null) {
\t\t\treturn false;
\t\t}else {
\t\t\treturn Pattern.matches(regexp, job);//很耗費時間
\t\t} 

\t}
\tpublic  boolean isMatchReduce(String name,String regexp) {
\t\tString key = "user."+name;
\t\tString job ;
\t\tsynchronized(this) { // 細緻化 更好
\t\t\tjob = matchMap.get(key);
\t\t}
\t
\t\tif(job == null) {
\t\t\treturn false;
\t\t}else {
\t\t\treturn Pattern.matches(regexp, job);
\t\t}
\t}/<string>/<code>

鎖粗化

synchronized特性：可重入，獨享，悲觀鎖

鎖優化:鎖消除是發生在編譯器級別的一種鎖優化方式，是指虛擬機即時編譯器在運行時，對一些代碼上要求同步，但是被檢測到不可能存在共享數據競爭的鎖進行削除（開啟鎖消除的參數：-xx:+DoEscapeAnalysis -XX:+EliminateLocks）

鎖粗化是指有些情況下我們反而希望把很多次鎖的請求合併成一個請求，以降低短時間內大量鎖請求、同步、釋放帶來的性能損耗。

減少鎖的粒度跟鎖分段

使用鎖的時候，鎖保護對象鎖是多個的，多個之間其實是獨立變化的 ，那就用多個鎖來分別保護。但是要注意發生死鎖。

<code>public class FinenessLock {
    public final Set<string> users = new HashSet<string>();
    public final Set<string> queries = new HashSet<string>();
    public void addUser(String u) {
        synchronized (users) { // 注意鎖的誰
            users.add(u);
        }
    }
    public void addQuery(String q) {
        synchronized (queries) { // 注意鎖的誰
            queries.add(q);
        }
    }
}/<string>/<string>/<string>/<string>/<code>

比如我們的ConcurrentHashMap用的分段鎖來提速。

替換獨佔鎖

• 讀寫鎖的使用，讀頻繁寫很少。
• 用CAS操作來替換重型鎖。
• 儘量用JDK自帶的併發容器。

閱讀更多 程序猿的內心獨白 的文章

關鍵字: 程序員 成員 線程