前言

去年年底面試了多多買菜，有圖為證，現整理面經，希望各位不要覺得太遲（這該死的拖延癥😂）。

周日晚上8點視頻面試的拼多多，結果人家全員加班中，辦公室中都是人，所以大家去多多前還是思考下把。🤣

?

問題1.arraylist線程是否安全，具體體現在哪行？

答：線程不安全，具體表現新增元素的賦值操作elementData[size++]=e。

?

我們先來看下什么是線程的安全性：

線程安全就是多線程訪問時對數據進行了加鎖機制(樂觀鎖或悲觀鎖)，只有一個線程能夠正常訪問，其他線程不能進行訪問直到該線程讀取完。不會出現數據不一致或者數據污染。

線程不安全就是多線程訪問時不提供數據訪問保護，有可能出現多個線程先后更改數據，所以得到的數據是臟數據。

?

如圖，AbstractList下面有兩個類，一個是線程不安全ArrayList，另外一個是線程安全vector。

?

從源碼的角度來看，因為Vector的方法前加了synchronized 關鍵字，也就是同步的意思，所以其是線程安全的。

?

所以我們看到這兩個的區別：

Vector：線程安全，但是性能低。

ArrayList:線程不安全，但是性能高，高效。

?

所以自古魚和熊掌不可兼得。

?

我們來看下具體的代碼，從下圖我們可以得出其添加元素時候的兩步走：

1. 在 Items[Size] 的位置存放此元素；

2. 增大 Size 的值。

?

我們來思考下：

在單線程運行的情況下，如果 Size = 0，添加一個元素后，此元素在位置 0，而且 Size=1，一切都是正常的；

而如果是在多線程情況下，比如有兩個線程，線程 A 先將元素存放在位置 0，此時size并沒有加一，但是此時 CPU 調度線程A暫停，線程 B 得到運行的機會。

線程B也向此 ArrayList 添加元素，因為此時 Size 仍然等于 0 （注意哦，我們假設的是添加一個元素是要兩個步驟哦，而線程A僅僅完成了步驟1），

所以線程B也將元素存放在位置0，此時size也沒有加1。然后線程A和線程B都繼續運行，都增加 Size 的值，此時size等于2。

那好，現在我們發現問題了，元素實際上只有一個，存放在位置 0，而 Size 卻等于 2。這就是“線程不安全”了。

?

看下面的代碼，在多線程并發情況下，提示了報錯信息，程序報了并發修改異常ConcurrentModificationException，我們也可以通過看下ArrayList底層中add(）方法，是沒有加鎖的操作，當多個線程共享一份資源時，可能發生線程問題；arrayList的add()方法是沒有加鎖的。

public static void main(String[] args) throws InterruptedException {List<Integer> list = new ArrayList<>();ExecutorService threadPool = Executors.newFixedThreadPool(30);for (int i = 1; i <= 30; i++) {int finalI = i;threadPool.execute(() -> {list.add(finalI);System.out.println(list);});}threadPool.shutdown();}

?

如果我們想要不報錯，可以將list轉化為線程安全的集合，使用Collections工具類的synchronizedList方法，來將其轉化線程安全的。

public static void main(String[] args) throws InterruptedException {List<Integer> list = Collections.synchronizedList(new ArrayList<>());//此處只是案例demo,真實使用線程池不建議用這種方式創建ExecutorService threadPool = Executors.newFixedThreadPool(30);for (int i = 1; i <= 30; i++) {int finalI = i;threadPool.execute(() -> {list.add(finalI);System.out.println(list);});}threadPool.shutdown();}

?

問題2.hashmap為什么用紅黑樹，不要AVL樹，或B+樹？

答：因為AVL樹比紅黑樹保持更加嚴格的平衡，是以更多旋轉操作導致更慢的插入和刪除為代價的樹，B+樹所有的節點擠在一起，當數據量不多的時候會退化成鏈表。

?

AVL樹

平衡二叉搜索樹（Self-balancing binary search tree）又被稱為AVL樹，且具有以下性質：它是一 棵空樹或它的左右兩個子樹的高度差的絕對值不超過1，并且左右兩個子樹都是一棵平衡二叉樹。

某在AVL樹中查找通常更快，但這是以更多旋轉操作導致更慢的插入和刪除為代價的。因此，如果希望查找次數主導樹的更新次數，請使用AVL樹。

下面兩張圖都是平衡二叉搜索樹。

? ? ? ? ? ? ? ? ? ? ? ? ? ?

?

那我們來看下不是平衡二叉搜索樹長什么樣子？從下圖可以看到C節點的左子樹有F，L，M，即為兩層，但是C節點的右子樹并沒有，他們相差了2，所以并不是平衡二叉搜索樹。

左子樹的左子樹插入結點 (左左)

?

右子樹的右子樹插入節點 (右右)

?

左子樹的右子樹插入節點 (左右)

?

右子樹的左子樹插入節點 (右左)

B+樹

B+樹的非葉子結點不存儲數據，所以每個結點能存儲的關鍵字更多。所以B+樹更能應對大量數據的情況。jdk1.7中的HashMap本來是數組+鏈表的形式，鏈表由于其查找慢的特點，所以需要被查找效率更高的樹結構來替換。
如果用B+樹的話，在數據量不是很多的情況下，數據都會“擠在”一個結點里面。這個時候遍歷效率就退化成了鏈表

?

一個m階的B樹具有如下幾個特征：

1.根結點至少有兩個子女。

2.每個中間節點都至少包含ceil(m / 2)個孩子，最多有m個孩子。

3.每一個葉子節點都包含k-1個元素，其中 m/2 <= k <= m。

4.所有的葉子結點都位于同一層。

5.每個節點中的元素從小到大排列，節點當中k-1個元素正好是k個孩子包含的元素的值域分劃。

?

?

問題3.ConcurrentHashMap為什么線程安全？哪些點保證了？

答：數組初始化的時候自旋來保證一定可以初始化成功，然后通過 CAS 設置 SIZECTL 變量的值，來保證同一時刻只能有一個線程對數組進行初始化，CAS 成功之后，還會再次判斷當前數組是否已經初始化完成，如果已經初始化完成，就不會再次初始化；

新增槽點時通過自旋保證一定新增成功，然后通過CAS來新增，如果遇到槽點有值，通過鎖住當前槽點或紅黑樹的根節點；

擴容時通過鎖住原數組的槽點，設置轉移節點，以及自旋等操作來保證線程安全。

?

ConcurrentHashMap 在 put 方法上的整體思路：
1. 如果數組為空，初始化，初始化完成之后，走 2；
2. 計算當前槽點有沒有值，沒有值的話，cas 創建，失敗繼續自旋（for 死循環），直到成功，槽點有值的話，走 3；
3. 如果槽點是轉移節點(正在擴容)，就會一直自旋等待擴容完成之后再新增，不是轉移節點走4；
4. 槽點有值的，先鎖定當前槽點，保證其余線程不能操作，如果是鏈表，新增值到鏈表的尾部，如果是紅黑樹，使用紅黑樹新增的方法新增；
5. 新增完成之后 check 需不需要擴容，需要的話去擴容。

具體源碼如下：

final V putVal (K key, V value,boolean onlyIfAbsent){if (key == null || value == null) throw new NullPointerException();//通過hashcode計算 hashint hash = spread(key.hashCode());int binCount = 0;for (Node<K, V>[] tab = table; ; ) {Node<K, V> f;int n, i, fh;//table為空，進行初始化工作，調用initTableif (tab == null || (n = tab.length) == 0)tab = initTable();//如果當前索引位置沒有值，直接創建else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//cas 在 i 位置創建新的元素，當 i 位置是空時，即能創建成功，結束 for 自循，否則繼續自旋if (casTabAt(tab, i, null,new Node<K, V>(hash, key, value, null)))break; // no lock when adding to empty bin}//如果當前槽點是轉移節點，表示該槽點正在擴容，就會一直等待擴容完成//轉移節點的 hash 值是固定的，都是 MOVEDelse if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);//槽點上有值的else {V oldVal = null;//鎖定當前槽點，其余線程不能操作，保證了安全synchronized (f) {//這里再次判斷 i 索引位置的數據沒有被修改//binCount 被賦值的話，說明走到了修改表的過程里面if (tabAt(tab, i) == f) {//鏈表if (fh >= 0) {binCount = 1;for (Node<K, V> e = f; ; ++binCount) {K ek;//值有的話，直接返回if (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {oldVal = e.val;if (!onlyIfAbsent)e.val = value;break;}Node<K, V> pred = e;//把新增的元素賦值到鏈表的最后，退出自旋if ((e = e.next) == null) {pred.next = new Node<K, V>(hash, key,value, null);break;}}}//紅黑樹，這里沒有使用 TreeNode,使用的是 TreeBin，TreeNode 只是紅黑樹的一個節點//TreeBin 持有紅黑樹的引用，并且會對其加鎖，保證其操作的線程安全else if (f instanceof TreeBin) {Node<K, V> p;binCount = 2;//滿足 if 的話，把老的值給 oldVal//在 putTreeVal 方法里面，在給紅黑樹重新著色旋轉的時候//會鎖住紅黑樹的根節點if ((p = ((TreeBin<K, V>) f).putTreeVal(hash, key,value)) != null) {oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}}//binCount 不為空，并且 oldVal 有值的情況，說明已經新增成功了if (binCount != 0) {// 鏈表是否需要轉化成紅黑樹if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;//這一步幾乎走不到。槽點已經上鎖，只有在紅黑樹或者鏈表新增失敗的時候//才會走到這里，這兩者新增都是自旋的，幾乎不會失敗break;}}}//check 容器是否需要擴容，如果需要去擴容，調用 transfer 方法去擴容//如果已經在擴容中了，check 有無完成addCount(1L, binCount);return null;}

數組初始化時的線程安全

數組初始化時，首先通過自旋來保證一定可以初始化成功，然后通過 CAS 設置 SIZECTL 變量的值，來保證同一時刻只能有一個線程對數組進行初始化，CAS 成功之后，還會再次判斷當前數組是否已經初始化完成，如果已經初始化完成，就不會再次初始化，通過自旋 + CAS + 雙重 check等手段保證了數組初始化時的線程安全，源碼如下：
? ? ? ? ? ? ? ? ? ?

//初始化 table，通過對 sizeCtl 的變量賦值來保證數組只能被初始化一次private final Node<K, V>[] initTable () {Node<K, V>[] tab;int sc;//通過自旋保證初始化成功while ((tab = table) == null || tab.length == 0) {// 小于 0 代表有線程正在初始化，釋放當前 CPU 的調度權，重新發起鎖的競爭if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spin// CAS 賦值保證當前只有一個線程在初始化，-1 代表當前只有一個線程能初始化// 保證了數組的初始化的安全性else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {// 很有可能執行到這里的時候，table 已經不為空了，這里是雙重 checkif ((tab = table) == null || tab.length == 0) {// 進行初始化int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];table = tab = nt;sc = n - (n >>> 2);}} finally {sizeCtl = sc;}break;}}return tab;}}

?新增槽點值時的線程安全

?此時為了保證線程安全，做了四處優化：

• ?通過自旋死循環保證一定可以新增成功。
• ?當前槽點為空時，通過 CAS 新增。
• ?當前槽點有值，鎖住當前槽點。
• ?紅黑樹旋轉時，鎖住紅黑樹的根節點，保證同一時刻，當前紅黑樹只能被一個線程旋轉

擴容中時的線程安全

• 拷貝槽點時，會把原數組的槽點鎖住；
• 拷貝成功之后，會把原數組的槽點設置成轉移節點，這樣如果有數據需要 put 到該節點時，發現該槽點是轉移節點，會一直等待，直到擴容成功之后，才能繼續 put，可以參考 put 方 法中的 helpTransfer 方法；
• 從尾到頭進行拷貝，拷貝成功就把原數組的槽點設置成轉移節點。
• 等擴容拷貝都完成之后，直接把新數組的值賦值給數組容器，之前等待 put 的數據才能繼續?put。

? ? ? ? ? ? ? ? ? ??

// 擴容主要分 2 步，第一新建新的空數組，第二移動拷貝每個元素到新數組中去// tab：原數組，nextTab：新數組private final void transfer (Node < K, V >[]tab, Node < K, V >[]nextTab){// 老數組的長度int n = tab.length, stride;if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide range// 如果新數組為空，初始化，大小為原數組的兩倍，n << 1if (nextTab == null) { // initiatingtry {@SuppressWarnings("unchecked")Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n << 1];nextTab = nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;transferIndex = n;}// 新數組的長度int nextn = nextTab.length;// 代表轉移節點，如果原數組上是轉移節點，說明該節點正在被擴容ForwardingNode<K, V> fwd = new ForwardingNode<K, V>(nextTab);boolean advance = true;boolean finishing = false; // to ensure sweep before committing nextTab// 無限自旋，i 的值會從原數組的最大值開始，慢慢遞減到 0for (int i = 0, bound = 0; ; ) {Node<K, V> f;int fh;while (advance) {int nextIndex, nextBound;// 結束循環的標志if (--i >= bound || finishing)advance = false;// 已經拷貝完成else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}// 每次減少 i 的值else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {bound = nextBound;i = nextIndex - 1;advance = false;}}// if 任意條件滿足說明拷貝結束了if (i < 0 || i >= n || i + n >= nextn) {int sc;// 拷貝結束，直接賦值，因為每次拷貝完一個節點，都在原數組上放轉移節點，所以拷貝完成的節點的數據一定不會再發生變化。// 原數組發現是轉移節點，是不會操作的，會一直等待轉移節點消失之后在進行操作。// 也就是說數組節點一旦被標記為轉移節點，是不會再發生任何變動的，所以不會有任何線程安全的問題// 所以此處直接賦值，沒有任何問題。if (finishing) {nextTable = null;table = nextTab;sizeCtl = (n << 1) - (n >>> 1);return;}if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;finishing = advance = true;i = n; // recheck before commit}} else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {synchronized (f) {// 進行節點的拷貝if (tabAt(tab, i) == f) {Node<K, V> ln, hn;if (fh >= 0) {int runBit = fh & n;Node<K, V> lastRun = f;for (Node<K, V> p = f.next; p != null; p = p.next) {int b = p.hash & n;if (b != runBit) {runBit = b;lastRun = p;}}if (runBit == 0) {ln = lastRun;hn = null;} else {hn = lastRun;ln = null;}// 如果節點只有單個數據，直接拷貝，如果是鏈表，循環多次組成鏈表拷貝for (Node<K, V> p = f; p != lastRun; p = p.next) {int ph = p.hash;K pk = p.key;V pv = p.val;if ((ph & n) == 0)ln = new Node<K, V>(ph, pk, pv, ln);elsehn = new Node<K, V>(ph, pk, pv, hn);}// 在新數組位置上放置拷貝的值setTabAt(nextTab, i, ln);setTabAt(nextTab, i + n, hn);// 在老數組位置上放上 ForwardingNode 節點// put 時，發現是 ForwardingNode 節點，就不會再動這個節點的數據了setTabAt(tab, i, fwd);advance = true;}// 紅黑樹的拷貝else if (f instanceof TreeBin) {// 紅黑樹的拷貝工作，同 HashMap 的內容，代碼忽略// 在老數組位置上放上 ForwardingNode 節點setTabAt(tab, i, fwd);advance = true;}}}}}}

?

問題4.剛才說到分布式鎖，談談設計思路和方案

答：主要根據具體的業務場景展開描述（這邊各個項目不一樣，就不展開說了），主要是引入redis實現的分布式鎖，應該保證互斥性（在任何時候只有一個客戶端持有鎖，使用setnx），不能死鎖（設置過期時間），保證上鎖和解鎖是同一個客戶端（設置不同的value值），業務時間太長，導致鎖過期（設置看門狗，自動續鎖），鎖的重入性（使用redis的hset）。

?

如果在一個分布式系統中，我們從數據庫中讀取一個數據，然后修改保存，這種情況很容易遇到并發問題。因為讀取和更新保存不是一個原子操作，在并發時就會導致數據的不正確。如果是單機應用，直接使用本地鎖就可以避免。如果是分布式應用，應用部署在多個JVM中，沒有辦法控制，所以引入分布式鎖來解決。

?

互斥性

127.0.0.1:6379> setnx lock value1 #在鍵lock不存在的情況下，將鍵key的值設置為value1 (integer) 1 127.0.0.1:6379> setnx lock value2 #試圖覆蓋lock的值，返回0表示失敗 (integer) 0 127.0.0.1:6379> get lock #獲取lock的值，驗證沒有被覆蓋 "value1" 127.0.0.1:6379> del lock #刪除lock的值，刪除成功 (integer) 1 127.0.0.1:6379> setnx lock value2 #再使用setnx命令設置，返回0表示成功 (integer) 1 127.0.0.1:6379> get lock #獲取lock的值，驗證設置成功

加鎖：使用setnx key value命令，如果key不存在，設置value(加鎖成功)。如果已經存在lock(也就是有客戶端持有鎖了)，則設置失敗(加鎖失敗)。

解鎖：使用del命令，通過刪除鍵值釋放鎖。

?

不能死鎖

設置過期時間，到點數據刪除，避免導致如果一個客戶端持有鎖的期間突然崩潰了，就會導致無法解鎖，則其他人將無法拿到該鎖，鎖會一直存在，最終導致出現死鎖的現象。

?

鎖過期

有效時間設置多長，假如我的業務操作比有效時間長，我的業務代碼還沒執行完就自動給我解鎖了，不就完蛋了嗎。

這個問題就有點棘手了，在網上也有很多討論，第一種解決方法就是靠程序員自己去把握，預估一下業務代碼需要執行的時間，然后設置有效期時間比執行時間長一些，保證不會因為自動解鎖影響到客戶端業務代碼的執行。但是這并不是萬全之策，比如網絡抖動這種情況是無法預測的，也有可能導致業務代碼執行的時間變長，所以并不安全。有一種方法比較靠譜一點，

就是給鎖續期。在Redisson框架實現分布式鎖的思路，就使用watchDog機制實現鎖的續期。當加鎖成功后，同時開啟守護線程，默認有效期是30秒，每隔10秒就會給鎖續期到30秒，只要持有鎖的客戶端沒有宕機，就能保證一直持有鎖，直到業務代碼執行完畢由客戶端自己解鎖，如果宕機了自然就在有效期失效后自動解鎖。

?

保證上鎖和解鎖都是同一個客戶端

key的值可以根據業務設置，比如是用戶中心使用的，可以命令為USER_REDIS_LOCK，value可以使用uuid保證唯一，用于標識加鎖的客戶端，保證加鎖和解鎖都是同一個客戶端。

每次解鎖可以先判斷鎖的value是不是當前用戶，如果是，說明可以解鎖，如果不是，則不能解鎖，會導致誤解了其他人的鎖。

?

鎖的重入性

可重入鎖意思是在外層使用鎖之后，內層仍然可以使用，那么可重入鎖的實現思路又是怎么樣的呢？在Redisson實現可重入鎖的思路，使用Redis的哈希表存儲可重入次數，當加鎖成功后，使用hset命令，value(重入次數)則是1。

解鎖時，先判斷可重復次數是否大于0，大于0則減一，否則刪除鍵值，釋放鎖資源。

?

問題5.算法題

看到這算法題，我笑了，這不是力扣的第一題嗎，哈哈哈，幸好刷過。簡單方法大家都會寫，暴力操作，但是性能有影響，但是評論區有為大神寫的很巧妙，我就直接搬過來了。

題目：

給定一個整數數組 nums?和一個整數目標值 target，請你在該數組中找出 和為目標值 的那?兩個?整數，并返回它們的數組下標。

你可以假設每種輸入只會對應一個答案。但是，數組中同一個元素在答案里不能重復出現。

你可以按任意順序返回答案。

解法：

這個解法并不是從題目計算答案，而是從答案出發，看需要什么數字。

public int[] twoSum(int[] nums, int target) {int[] indexs = new int[2];// 建立k-v ，一一對應的哈希表HashMap<Integer,Integer> hash = new HashMap<Integer,Integer>();for(int i = 0; i < nums.length; i++){if(hash.containsKey(nums[i])){indexs[0] = i;indexs[1] = hash.get(nums[i]);return indexs;}// 將數據存入 key為補數 ，value為下標hash.put(target-nums[i],i);}// // 雙重循環 循環極限為(n^2-n)/2 // for(int i = 0; i < nums.length; i++){// for(int j = nums.length - 1; j > i; j --){// if(nums[i]+nums[j] == target){// indexs[0] = i;// indexs[1] = j; // return indexs;// }// }// }return indexs;}

?

總結

以上是生活随笔為你收集整理的《关于我横扫一线厂的那些面经》拼多多Java岗（附答案）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。