并发读源码——AtomicStampedReference/AtomicMarkableReference

╰+哭是因爲堅強的太久メ 2022-10-28 10:13 150阅读 0赞

### 文章目录 ###

*  1、AtomicStampedReference介绍
 *  2、AtomicStampedReference原理分析
 *  3、AtomicStampedReference源码分析
 *  4、AtomicStampedReference多线程用法
 *  5、AtomicMarkableReference分析

# 1、AtomicStampedReference介绍 #

前面两节介绍的AtomicBoolean、AtomicInteger、AtomicLong都是解决的CAS问题，即拿读取的变量和内存变量比较，如果相同就把要更新的值更新到内存中。但有一种情况：假设一个对象O中一个变量的值为1，线程A读取该变量的值为1，然后线程B把该对象O中的变量值更新为2，然后又从2更新到1，线程A认为拿到的对象O的变量值在内存中没有改变，做CAS操作就会成功，但实际A拿到的值已经被改过了，只是被线程B给“欺骗”了，导致线程A后续处理的逻辑发生变化。该现象称为ABA问题，为解决此问题，在更新值时，不仅要比较值是否相等，还应比较值的版本号是否相等。AtomicStampedReference正是为解决ABA问题而生。

# 2、AtomicStampedReference原理分析 #

AtomicStampedReference是通过下面compareAndSet方法解决ABA问题的

public boolean compareAndSet(V   expectedReference,
                                 V   newReference,
                                 int expectedStamp,
                                 int newStamp) {

newReference为要更新的内容，newStamp为要更新的版本号，expectedReference和expectedStamp原内容和版本号，compareAndSet在更新时是通过 UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val)执行的，cmp和val都是Pair类型的，Pair是封装在AtomicStampedReference的一个内部类。  
为什么要疯转Pair内部类呢，是因为compareAndSwapObject只能比较cmp与内存中的Pair的对象是否相等，而Pair正好封装了reference对象和stamp版本号，compareAndSwapObject只能做两个对象之间的比较，不能在对象的比较时同时做版本号的比较，因此把要比较的对象和对象的版本号封装到Pair中

private static class Pair<T> { 
        final T reference;
        final int stamp;
        private Pair(T reference, int stamp) { 
            this.reference = reference;
            this.stamp = stamp;
        }
        static <T> Pair<T> of(T reference, int stamp) { 
            return new Pair<T>(reference, stamp);
        }
    }

可见，要更新newReference和newStamp需要先比较expectedReference和expectedStamp与Pari中的reference 和stamp 是否相等，因此就转化成cmp与Pari对象中内容比较是否相等，如果相等则把val中的newReference和newStamp更新到内存Pair对象中。

# 3、AtomicStampedReference源码分析 #

package java.util.concurrent.atomic;
    
    public class AtomicStampedReference<V> { 
    
        /* * reference和stamp封装到Pair内部类中 * Pair内部类是私有的，在Pair类外部不能通过new的方式创建，只能通过调用Pair的of方法创建pair对象 * Pair是静态类，可以通过类直接调用of方法 * */
    	private static class Pair<T> { 
            final T reference;
            final int stamp;
            private Pair(T reference, int stamp) { 
                this.reference = reference;
                this.stamp = stamp;
            }
            static <T> Pair<T> of(T reference, int stamp) { 
                return new Pair<T>(reference, stamp);
            }
        }
    
    	/* * 定义pair对象是volatile的，使pair对象修改立即对其它线程可见 * */
        private volatile Pair<V> pair;
    
        /*创建AtomicStampedReference对象，通过initialRef和initialStamp初始化pari对象*/
        public AtomicStampedReference(V initialRef, int initialStamp) { 
            pair = Pair.of(initialRef, initialStamp);
        }
    
        /*获取Pair封装的reference对象*/
        public V getReference() { 
            return pair.reference;
        }
    
        /*获取Pair封装的stamp版本号*/
        public int getStamp() { 
            return pair.stamp;
        }
    
        /* * dskjf同时返回Pair对象中的reference和stamp * reference通过return返回，stamp通过int[]数组返回 * */
        public V get(int[] stampHolder) { 
            Pair<V> pair = this.pair;
            stampHolder[0] = pair.stamp;
            return pair.reference;
        }
    
        /* * weakCompareAndSet与compareAndSet功能类似，但weakCompareAndSet不会做内存保障，即不会保障volatile类型变量pari的原子性 * */
        public boolean weakCompareAndSet(V   expectedReference,
                                         V   newReference,
                                         int expectedStamp,
                                         int newStamp) { 
            return compareAndSet(expectedReference, newReference,
                                 expectedStamp, newStamp);
        }
    
        /* * 比较expectedReference和expectedStamp与Pair对象中的reference和stamp是否相等 * 如果相等，则分别把newReference和newStamp更新到Pair对象中的reference和stamp，又因为Pair对象是volatile的，会立即更新到主内存中，其它线程可以发现新修改的值 * 案例：如果expectedReference和expectedStamp与主内存中Pair对象中的reference和stamp相等，并且newReference和newStamp也与主内存中Pair对象中的reference和stamp相等，此时不会更新Pair对象中的值 * 案例 ：如果expectedReference和expectedStamp与主内存中Pair对象中的reference和stamp相等，并且newReference和newStamp也与主内存中Pair对象中的reference和stamp不相等，此时才会更新Pair对象中的值 * * 在把newReference和newStamp更新到主内存Pair对象中时，时通过casPair方法执行的，该方法把newReference, newStamp封装到一个Pair对象中， * 然后通过UNSAFE.compareAndSwapObject比较先前读取的Pair的current对象与主内存中的Pair对象比较，如果相等则把新封装的Pair对象更新掉主内存中的Pair对象 * */
        public boolean compareAndSet(V   expectedReference,
                                     V   newReference,
                                     int expectedStamp,
                                     int newStamp) { 
            Pair<V> current = pair;
            return
                expectedReference == current.reference &&
                expectedStamp == current.stamp &&
                ((newReference == current.reference &&
                  newStamp == current.stamp) ||
                 casPair(current, Pair.of(newReference, newStamp)));
        }
    
        /*把newReference和newStamp更新到pair对象中，前提条件：newReference和newStamp分别不等于pair对象中的reference和stamp*/
        public void set(V newReference, int newStamp) { 
            Pair<V> current = pair;
            if (newReference != current.reference || newStamp != current.stamp)
                this.pair = Pair.of(newReference, newStamp);
        }
    
        /*该方法用于更新版本号，前提是expectedReference等于pair对象中的reference * 当expectedReference等于pair对象中的reference时，有两种情况：newStamp==stamp，此时什么都不做，直接返回true； * 当newStamp != stamp时，把expectedReference和newStamp构建一个pair对象更新到成员变量pair对象中 * 单次更新有可能会失败，因为多线程并发时修改了reference，可以通过自旋锁的方式，循环执行该方法，直至成功 * */
        public boolean attemptStamp(V expectedReference, int newStamp) { 
            Pair<V> current = pair;
            return
                expectedReference == current.reference &&
                (newStamp == current.stamp ||
                 casPair(current, Pair.of(expectedReference, newStamp)));
        }
    
        private static final sun.misc.Unsafe UNSAFE = sun.misc.Unsafe.getUnsafe();
        /*获取成员变量pair相对AtomicStampedReference的偏移量，后面compareAndSwapObject做CAS时就是通过该偏移量执行的*/
        private static final long pairOffset =
            objectFieldOffset(UNSAFE, "pair", AtomicStampedReference.class);
    
        private boolean casPair(Pair<V> cmp, Pair<V> val) { 
            return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
        }
    
        static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
                                      String field, Class<?> klazz) { 
            try { 
                return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
            } catch (NoSuchFieldException e) { 
                // Convert Exception to corresponding Error
                NoSuchFieldError error = new NoSuchFieldError(field);
                error.initCause(e);
                throw error;
            }
        }
    }

# 4、AtomicStampedReference多线程用法 #

假设两个线程A和B，A负责生产商品，每生产一个商品库存增加1，商品编号加1；B负责消费商品，每消费一个商品，库存减1，商品编号减1。之所以为每个商品加一个编号（即版本号），当多线程并发时，如果库存量与编号不一致则不能更新库存，保证了线程安全性。

package com.lzj.atomic.atomicstampedreferencetest;
    
    import java.util.concurrent.atomic.AtomicStampedReference;
    
    public class AtomicStampedReferenceDemo { 
    
    	/*本案例处理的是Integer对象 */
    	public AtomicStampedReference<Integer> stock = new AtomicStampedReference<Integer>(0, 0);
    	public Integer num = 10;  //当库存量达到10时，停止生产和消费
    	
    	public void producter() { 
    		Integer initialRef = stock.getReference();
    		while(initialRef < num) { 
    			Integer newReference = initialRef + 1;
    			int stamped = stock.getStamp();
    			int newStamped = stamped + 1;
    			if(stock.compareAndSet(initialRef, newReference, stamped, newStamped)) { 
    				System.out.println(Thread.currentThread() + "生产商品 " + newReference + "个；商品编号为 " + newStamped);
    			}
    			initialRef = stock.getReference();
    		}
    		
    	}
    	
    	public void consumer() { 
    		Integer initialRef = stock.getReference();
    		while(initialRef < num) { 
    			Integer newReference = initialRef - 1;
    			int stamped = stock.getStamp();
    			int newStamped = stamped - 1;
    			if(stock.compareAndSet(initialRef, newReference, stamped, newStamped)) { 
    				System.out.println(Thread.currentThread() + "生产商品 " + newReference + "个；商品编号为 " + newStamped);
    			}
    			initialRef = stock.getReference();
    		}
    	}
    		
    	public static void main(String[] args) throws InterruptedException { 
    		AtomicStampedReferenceDemo demo = new AtomicStampedReferenceDemo();
    		Runnable run1 = new Runnable() { 
    			
    			@Override
    			public void run() { 
    				demo.producter();
    				
    			}
    		};
    		Runnable run2 = new Runnable() { 
    			
    			@Override
    			public void run() { 
    				demo.consumer();
    				
    			}
    		};
    		
    		Thread thread1 = new Thread(run1);
    		Thread thread2 = new Thread(run2);
    		thread1.start();
    		thread2.start();
    	}
    }

执行结果如下：

Thread[Thread-0,5,main]生产商品 0个；商品编号为 0
    Thread[Thread-0,5,main]生产商品 1个；商品编号为 1
    Thread[Thread-1,5,main]生产商品 0个；商品编号为 0
    Thread[Thread-0,5,main]生产商品 1个；商品编号为 1
    Thread[Thread-0,5,main]生产商品 1个；商品编号为 1
    Thread[Thread-0,5,main]生产商品 2个；商品编号为 2
    Thread[Thread-0,5,main]生产商品 3个；商品编号为 3
    Thread[Thread-0,5,main]生产商品 4个；商品编号为 4
    Thread[Thread-1,5,main]生产商品 0个；商品编号为 0
    Thread[Thread-0,5,main]生产商品 5个；商品编号为 5
    Thread[Thread-0,5,main]生产商品 5个；商品编号为 5
    Thread[Thread-0,5,main]生产商品 6个；商品编号为 6
    Thread[Thread-1,5,main]生产商品 4个；商品编号为 4
    Thread[Thread-0,5,main]生产商品 7个；商品编号为 7
    Thread[Thread-1,5,main]生产商品 6个；商品编号为 6
    Thread[Thread-1,5,main]生产商品 6个；商品编号为 6
    Thread[Thread-1,5,main]生产商品 5个；商品编号为 5
    Thread[Thread-0,5,main]生产商品 7个；商品编号为 7
    Thread[Thread-0,5,main]生产商品 5个；商品编号为 5
    Thread[Thread-0,5,main]生产商品 6个；商品编号为 6
    Thread[Thread-0,5,main]生产商品 7个；商品编号为 7
    Thread[Thread-0,5,main]生产商品 8个；商品编号为 8
    Thread[Thread-0,5,main]生产商品 9个；商品编号为 9
    Thread[Thread-0,5,main]生产商品 10个；商品编号为 10
    Thread[Thread-1,5,main]生产商品 4个；商品编号为 4

# 5、AtomicMarkableReference分析 #

AtomicMarkableReference用法、源码与AtomicStampedReference基本一致，除了，AtomicStampedReference的版本号是int类型的，可以有2^31-1个版本号可以设置，而AtomicMarkableReference的版本号是boolean类型的，那么AtomicMarkableReference的版本号只能是true或者false两种。如果A和B两个线程，A把版本号由true修改为false，然后又由false修改为true，那么最开始B线程拿到的版本号是true，认为其他线程并没有修改资源，导致后续逻辑处理有问题。所以AtomicMarkableReference只是降低了ABA发生的概率，但不能完全避免ABA问题，优先建议选择AtomicStampedReference。

public class AtomicMarkableReference<V> { 
    
    	/*Pair封装了对象和boolean类型的版本号，其他代码与AtomicStampedReference一致，不做解释*/
        private static class Pair<T> { 
            final T reference;
            final boolean mark;
            private Pair(T reference, boolean mark) { 
                this.reference = reference;
                this.mark = mark;
            }
            static <T> Pair<T> of(T reference, boolean mark) { 
                return new Pair<T>(reference, mark);
            }
        }
        ........
    }