前言
本章节继上章节继续梳理：线程相关的基础理论和工具、多线程程序下的性能调优和电商场景下多线程的使用。
多线程J·U·C
ThreadLocal
概念
ThreadLocal类并不是用来解决多线程环境下的共享变量问题，而是用来提供线程内部的共享变量。在多线程环境下，可以保证各个线程之间的变量互相隔离、相互独立。
使用
ThreadLocal实例一般定义为private static类型的，在一个线程内，该变量共享一份，类似上下文作用，可以用来上下传递信息。
public class ThreadLocalDemo implements Runnable{
private static ThreadLocal<Integer> threadLocal = new ThreadLocal<>();
public void run(){
for (int i = 0; i < 3; i++) {
threadLocal.set(i);
System.out.println(Thread.currentThread().getName()+",value="+threadLocal.get()
);
}
}
public static void main(String[] args) {
ThreadLocalDemo demo = new ThreadLocalDemo();
new Thread(demo).start();
new Thread(demo).start();
}
}
结果分析：
同一个demo实例，不同的thread嵌套
结果打印了各自的变量值，线程内上下文被传递，不同线程间被隔离
应用场景
数据库连接，session管理
下面的基于日志平台的访问链路追踪中，会用到
使用中遇到的坑
参与过一个项目，电商商铺详情页凌晨调度生成。需要上下传递shopid，为每个商铺重新生成一下。在商铺详情页里因为是按面包屑分片生成，比如商铺信息、热卖商品、最多好评、店主推荐、最新上架等。
其他信息全部生成ok，唯独商品列表多个列表出现问题。经查，在商品部分的查询中用到了ThreadLocal，造成当前商铺id丢失。
源码解析
ThreadLocalMap是ThreadLocal内部类，由ThreadLocal创建，每个Thread里维护一个
ThreadLocal. ThreadLocalMap类型的属性threadLocals。所有的value值其实是存储在
ThreadLocalMap中。
这个存储结构的思路是反转的...
set方法
public void set(T value) {
//取到当前线程
Thread t = Thread.currentThread();
//从当前线程中拿出Map
ThreadLocalMap map = getMap(t);
if (map != null)
//如果非空，说明之前创建过了
//以当前创建的ThreadLocal对象为key，需要存储的值为value，写入Map
//因为每个线程Thread里有自己独自的Map，所以起到了隔离作用
map.set(this, value);
else
//如果没有，那就创建
createMap(t, value);
}
get方法
public T get() {
Thread t = Thread.currentThread();
//获取到当前线程下的Map
ThreadLocalMap map = getMap(t);
if (map != null) {
//如果非空，根据当前ThreadLocal为key，取出对应的value即可
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
//如果map是空的，往往返回一个初始值，这是一个protect方法
//这就是为什么创建ThreadLocal的时候往往要求实现这个方法
return setInitialValue();
}
remove方法
public void remove() {
ThreadLocalMap m = getMap(Thread.currentThread());
//很简单，获取到map后，调用remove移除掉
if (m != null)
m.remove(this);
}
ThreadLocal是如何避免内存泄漏的
在上述的get方法中，Entry类继承了WeakReference，即每个Entry对象都有一个ThreadLocal的弱引用，GC对于弱引用的对象采取积极的内存回收策略，避免无人搭理时发生内存泄露。
Fork/Join
概念
ForkJoin是由JDK1.7后提供多线并发处理框架。ForkJoinPool由Java大师Doug Lea主持编写，处理逻辑大概分为两步。
1.任务分割：Fork（分岔），先把大的任务分割成足够小的子任务，如果子任务比较大的话还要对子任务进行继续分割。
2.合并结果：join，分割后的子任务被多个线程执行后，再合并结果，得到最终的完整输出。
组成
- ForkJoinT ask：主要提供fork和join两个方法用于任务拆分与合并；多数使用。RecursiveAction（无返回值的任务）和RecursiveT ask（需要返回值）来实现compute方法。
- ForkJoinPool：调度ForkJoinT ask的线程池；
- ForkJoinWorkerThread：Thread的子类，存放于线程池中的工作线程（Worker）；
- WorkQueue：任务队列，用于保存任务；
基本使用
一个典型的例子：计算1-1000的和
public class SumTask {
private static final Integer MAX = 100;
static class SubTask extends RecursiveTask<Integer> {
// 子任务开始计算的值
private Integer start;
// 子任务结束计算的值
private Integer end;
public SubTask(Integer start , Integer end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if(end - start < MAX) {
//小于边界，开始计算
System.out.println("start = " + start + ";end = " + end);
Integer totalValue = 0;
for(int index = this.start ; index <= this.end ; index++) {
totalValue += index;
}
return totalValue;
}else {
//否则，中间劈开继续拆分
SubTask subTask1 = new SubTask(start, (start + end) / 2);
subTask1.fork();
SubTask subTask2 = new SubTask((start + end) / 2 + 1 , end);
subTask2.fork();
return subTask1.join() + subTask2.join();
}
}
}
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool();
Future<Integer> taskFuture = pool.submit(new SubTask(1,1000));
try {
Integer result = taskFuture.get();
System.out.println("result = " + result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace(System.out);
}
}
}
设计思想
- 普通线程池内部有两个重要集合：工作线程集合，和任务队列。
- ForkJoinPool也类似，工作集合里放的是特殊线程ForkJoinWorkerThread，任务队列里放的是特殊任务ForkJoinTask
- 不同之处在于，普通线程池只有一个队列。而ForkJoinPool的工作线程ForkJoinWorkerThread每个线程内都绑定一个双端队列。
- 在fork的时候，也就是任务拆分，将拆分的task会被当前线程放到自己的队列中。
- 队列中的任务被线程执行时，有两种模式，默认是同步模式（asyncMode==false）从队尾取任务（LIFO）
- 窃取：当自己队列中执行完后，工作线程会到其他队列的队首获取任务（FIFO），取到后如果任务再次fork，拆分会被放入当前线程的队列，依次扩张
注意
使用ForkJoin将相同的计算任务通过多线程执行。但是在使用中需要注意：
注意任务切分的粒度，也就是fork的界限。并非越小越好
判断要不要使用ForkJoin。任务量不是太大的话，串行可能优于并行。因为多线程会涉及到上下文的切换
Volatile
概念
回顾Java 内存模型中的可见性、原子性和有序性：
- 可见性，是指线程之间的可见性，一个线程修改的状态对另一个线程是可见的
- 原子性，指的是这个操作是原子不可拆分的，不允许别的线程中间插队操作
- 有序性指的是你写的代码的顺序要和最终执行的指令保持一致。因为在Java内存模型中，允许编译器和处理器对指令进行重排序，重排序过程不会影响到单线程程序的执行，却会影响到多线程并发执行的正确性。
volatile要解决的就是可见性和有序性问题。
原理
Java内存模型分为主内存和线程工作内存两大类。
主内存：多个线程共享的内存。方法区和堆属于主内存区域。
线程工作内存：每个线程独享的内存。虚拟机栈、本地方法栈、程序计数器属于线程独享的工作内存
Java内存模型规定，所有变量都需要存储在主内存中，线程需要时，在自己的工作内存保存变量的副本，线程对变量的所有操作都在工作内存中进行，执行结束后再同步到主内存中去。这里必然会存在时间差，在这个时间差内，该线程对副本的操作，对于其他线程是不见的，从而造成了可见性问题。
但是，当对volatile变量进行写操作的时候，JVM会向处理器发送一条lock前缀的指令，将这个缓存中的变量回写到系统主存中。
同时，在多处理器下，为了保证各个处理器的缓存是一致的，就会实现缓存一致性协议。每个处理器通过嗅探在总线上传播的数据来检查自己缓存的值是不是过期，一旦发现过期就会将当前处理器的缓存行设置成无效状态，强制从主内存读取，这就保障了可见性。
而volatile变量，通过内存屏障可以禁止指令重排。从而实现指令的有序性。
注意
volatile不能保证锁的原子性。
案例：给前面的计数器案例里加上volatile试试
public class BadVolatile {
private static volatile int i=0;
public int get(){
return i;
}
public void inc(){
int j=get();
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
i=j+1;
}
public static void main(String[] args) throws InterruptedException {
final BadVolatile counter = new BadVolatile();
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
public void run() {
counter.inc();
}
}).start();
}
Thread.sleep(3000);
//理论上10才对。可是....
System.out.println(counter.i);
}
}
达不到目的。说明原子性无法保障。
ConcurrentHashMap
基本使用
public static void main(String[] args) throws InterruptedException {
//定义ConcurrentHashMap
Map map = new ConcurrentHashMap();
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
@Override
public void run() {
//多线程下的put可以放心使用
map.put(UUID.randomUUID().toString(), "1");
}
}).start();
}
Thread.sleep(3000);
System.out.println(map);
}
原理
jdk1.7是分段锁,1.8使用的cas+sychronized操作,具体看代码
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//计算hash
int binCount = 0;
for (Node<K,V>[] tab = table;;) {//自旋，确保插入成功
Node<K,V> f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)
tab = initTable();//表为空的话，初始化表
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//否则，插入元素，看下面的 casTabAt 方法
//cas 在这里！
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break;
}
//...
else {
V oldVal = null;
//其他情况下，加锁保持
//synchronized 在这里！
synchronized (f) {
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);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
//compareAndSetObject，比较并插入，典型CAS操作
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSetObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
//get取值
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
//判断table是不是空的，当前桶上是不是空的
//如果为空，返回null
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//找到对应hash槽的第一个node，如果key相等，返回value
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//如果正在扩容，不影响，继续顺着node找即可
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//其他情况，逐个便利，比对key，找到后返回value
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
总结：
put过程：
1.根据key的hash值定位到桶位置
2.如果table为空if，先初始化table。
3.如果table当前桶里没有node，cas添加元素。成功则跳出循环，失败则进入下一轮for循环。
4.判断是否有其他线程在扩容，有则帮忙扩容，扩容完成再添加元素。
5.如果桶的位置不为空，遍历该桶的链表或者红黑树，若key已存在，则覆盖，不存在则将key插入到链表或红黑树的尾部。
get过程：
1.根据key的hash值定位到桶位置。
2.map是否初始化，没有初始化则返回null
3.定位的桶是否有头结点，没有返回null
4.是否有其他线程在扩容，有的话调用find方法沿node指针往后查找。扩容与find可以并行，因为node的next指针不会变
5.若没有其他线程在扩容，则遍历桶对应的链表或者红黑树，使用equals方法进行比较。key相同则返回value,不存在则返回null
并发容器
1.ConcurrentHashMap
对应：HashMap
目标：代替Hashtable、synchronizedMap，使用最多，前面详细介绍过
原理：JDK7中采用Segment分段锁，JDK8中采用CAS+synchronized
2.CopyOnWriteArrayList
对应：ArrayList
目标：代替Vector、synchronizedList
原理：高并发往往是读多写少的特性，读操作不加锁，而对写操作加Lock独享锁，先复制一份新的集
合，在新的集合上面修改，然后将新集合赋值给旧的引用，并通过volatile 保证其可见性。
查看源码：volatile array，lock加锁，数组复制
3.CopyOnWriteArraySet
对应：HashSet
目标：代替synchronizedSet
原理：与CopyOnWriteArrayList实现原理类似。
4.ConcurrentSkipListMap
对应：TreeMap
目标：代替synchronizedSortedMap(TreeMap)
原理：基于Skip list（跳表）来代替平衡树，按照分层key上下链接指针来实现。
5.ConcurrentSkipListSet
对应：TreeSet
目标：代替synchronizedSortedSet(TreeSet)
原理：内部基于ConcurrentSkipListMap实现，原理一致
6.ConcurrentLinkedQueue
对应：LinkedList
对应：*线程安全队列
原理：通过队首队尾指针，以及Node类元素的next实现FIFO队列
7.BlockingQueue
对应：Queue
特点：拓展了Queue，增加了可阻塞的插入和获取等操作
原理：通过ReentrantLock实现线程安全，通过Condition实现阻塞和唤醒
实现类：
LinkedBlockingQueue：基于链表实现的可阻塞的FIFO队列
ArrayBlockingQueue：基于数组实现的可阻塞的FIFO队列
PriorityBlockingQueue：按优先级排序的队列
性能调优
锁优化
Synchronized
使用synchronized时注意锁粒度
public synchronized void test(){
// TODO
}
public void test(){
synchronized (this) {
// TODO
}
}
public static synchronized void test(){
// TODO
}
public static void test(){
synchronized (TestSynchronized.class) {
// TODO
}
}
Lock锁优化
举个例子：电商系统中记录首页被用户浏览的次数，以及最后一次操作的时间（含读或写）。
public class TotalLock {
//类创建的时间
final long start = System.currentTimeMillis();
//总耗时
AtomicLong totalTime = new AtomicLong(0);
//缓存变量
private Map<String,Long> map = new HashMap(){{put("count",0L);}};
ReentrantLock lock = new ReentrantLock();
//查看map被写入了多少次
public Map read(){
lock.lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
//最后操作完成的时间
map.put("time",end);
lock.unlock();
System.out.println(Thread.currentThread().getName()+",read="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
//写入
public Map write(){
lock.lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//写入计数
map.put("count",map.get("count")+1);
long end = System.currentTimeMillis();
map.put("time",end);
lock.unlock();
System.out.println(Thread.currentThread().getName()+",write="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
public static void main(String[] args) throws InterruptedException {
TotalLock count = new TotalLock();
//读
for (int i = 0; i < 4; i++) {
new Thread(()->{
count.read();
}).start();
}
//写
for (int i = 0; i < 1; i++) {
new Thread(()->{
count.write();
}).start();
}
Thread.sleep(3000);
System.out.println(count.map);
System.out.println("读写总共耗时："+count.totalTime.get());
}
}
查看后,我们发现查看次数这里其实是可以并行读取的，我们关注的业务是写入次数，也就是count，至于读取发生的时间time的写入操作，只是一个put，不需要原子性保障，对这个加互斥锁没有必要。改成读写锁试试……
public class ReadAndWrite {
//类创建的时间
final long start = System.currentTimeMillis();
//总耗时
AtomicLong totalTime = new AtomicLong(0);
//缓存变量，注意！因为read并发，这里换成ConcurrentHashMap
private Map<String,Long> map = new ConcurrentHashMap(){{put("count",0L);}};
ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
//查看map被写入了多少次
public Map read(){
lock.readLock().lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
//最后操作完成的时间
map.put("time",end);
lock.readLock().unlock();
System.out.println(Thread.currentThread().getName()+",read="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
//写入
public Map write(){
lock.writeLock().lock();
try {
Thread.currentThread().sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//写入计数
map.put("count",map.get("count")+1);
long end = System.currentTimeMillis();
map.put("time",end);
lock.writeLock().unlock();
System.out.println(Thread.currentThread().getName()+",write="+(endstart));
totalTime.addAndGet(end - start);
return map;
}
public static void main(String[] args) throws InterruptedException {
ReadAndWrite rw = new ReadAndWrite();
//读
for (int i = 0; i < 4; i++) {
new Thread(()->{
rw.read();
}).start();
}
//写
for (int i = 0; i < 1; i++) {
new Thread(()->{
rw.write();
}).start();
}
Thread.sleep(3000);
System.out.println(rw.map);
System.out.println("读写总共耗时："+rw.totalTime.get());
}
}
再来看读的时间变化和总执行时间。当read远大于write时，这个差距会更明显
CAS乐观锁优化
举例,直接用synchronized
public class NormalSync implements Runnable{
Long start = System.currentTimeMillis();
int i=0;
public synchronized void run() {
int j = i;
//实际业务中可能会有一堆的耗时操作，这里等待100ms模拟
try {
//做一系列操作
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//业务结束后，增加计数
i = j+1;
System.out.println(Thread.currentThread().getId()+
" ok,time="+(System.currentTimeMillis() - start));
}
public static void main(String[] args) throws InterruptedException {
NormalSync test = new NormalSync();
new Thread(test).start();
new Thread(test).start();
Thread.currentThread().sleep(1000);
System.out.println("last value="+test.i);
}
}
//线程二最终耗时会在200ms+，总耗时300ms，原因是悲观锁卡在了read后的耗时操作上，但是保证了
//最终结果是2
使用cas
public class CasSync implements Runnable{
long start = System.currentTimeMillis();
int i=0;
public void run() {
int j = i;
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//CAS处理，在这里理解思想，实际中不推荐大家使用！
try {
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
Unsafe unsafe = (Unsafe) f.get(null);
long offset =
unsafe.objectFieldOffset(CasSync.class.getDeclaredField("i"));
while (!unsafe.compareAndSwapInt(this,offset,j,j+1)){
j = i;
}
} catch (Exception e) {
e.printStackTrace();
}
//实际开发中，要用atomic包，或者while+synchronized自旋
// synchronized (this){
// //注意这里！
// while (j != i){
// j = i;
// }
// i = j+1;
// }
System.out.println(Thread.currentThread().getName()+
" ok,time="+(System.currentTimeMillis() - start));
}
public static void main(String[] args) throws InterruptedException {
CasSync test = new CasSync();
new Thread(test).start();
new Thread(test).start();
Thread.currentThread().sleep(1000);
System.out.println("last value="+test.i);
}
}
//线程一、二均在100ms+，总耗时200ms，最终结果还是2
总结
减少锁的时间 不需要同步执行的代码，能不放在同步快里面执行就不要放在同步快内，可以让锁尽快释放
减少锁的粒度 将物理上的一个锁，拆成逻辑上的多个锁，增加并行度，从而降低锁竞争，典型如分段锁
锁的粒度 拆锁的粒度不能无限拆，最多可以将一个锁拆为当前cup数量相等
减少加减锁的次数 假如有一个循环，循环内的操作需要加锁，我们应该把锁放到循环外面，否则每次进出循环，都要加锁
使用读写锁 业务细分，读操作加读锁，可以并发读，写操作使用写锁，只能单线程写，参考计数器案例
善用volatile volatile的控制比synchronized更轻量化，在某些变量上可以加以运用，如单例模式中
线程池参数优化
一些经验
1）corePoolSize
核心线程数，一旦有任务进来，在core范围内会立刻创建线程进入工作。所以这个值应该参考业务并发量在绝大多数时间内的并发情况。同时分析任务的特性。
高并发，执行时间短的，要尽可能小的线程数，如配置CPU个数+1，减少线程上下文的切换。因为它不怎么占时间，让少量线程快跑干活。
并发不高、任务执行时间长的要分开看：如果时间都花在了IO上，那就调大，如配置两倍CPU个数+1。不能让CPU闲下来，线程多了并行处理更快。如果时间都花在了运算上，运算的任务还很重，本身就很占cpu，那尽量减少cpu，减少切换时间。参考第一条
如果高并发，执行时间还很长……
2）workQueue
任务队列，用于传输和保存等待执行任务的阻塞队列。这个需要根据你的业务可接受的等待时间。是一个需要权衡时间还是空间的地方，如果你的机器cpu资源紧张，jvm内存够大，同时任务又不是那么紧迫，减少coresize，加大这里。如果你的cpu不是问题，对内存比较敏感比较害怕内存溢出，同时任务又要求快点响应。那么减少这里。
3）maximumPoolSize
线程池最大数量，这个值和队列要搭配使用，如果你采用了*队列，那很大程度上，这个参数没有意义。同时要注意，队列盛满，同时达到max的时候，再来的任务可能会丢失（下面的handler会讲）。
如果你的任务波动较大，同时对任务波峰来的时候，实时性要求比较高。也就是来的很突然并且都是着急的。那么调小队列，加大这里。如果你的任务不那么着急，可以慢慢做，那就扔队列吧。
队列与max是一个权衡。队列空间换时间，多花内存少占cpu，轻视任务紧迫度。max舍得cpu线程开销，少占内存，给任务最快的响应。
4）keepaliveTime
线程存活保持时间，超出该时间后，线程会从max下降到core，很明显，这个决定了你养闲人所花的代价。如果你不缺cpu，同时任务来的时间没法琢磨，波峰波谷的间隔比较短。经常性的来一波。那么实当的延长销毁时间，避免频繁创建和销毁线程带来的开销。如果你的任务波峰出现后，很长一段时间不再出现，间隔比较久，那么要适当调小该值，让闲着不干活的线程尽快销毁，不要占据资源。
5）threadFactory（自定义展示实例）
线程工厂，用于创建新线程。threadFactory创建的线程也是采用new Thread()方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）。如果需要自己定义线程的某些属性，如个性化的线程名，可以在这里动手。一般不需要折腾它。
6）handler
线程饱和策略，当线程池和队列都满了，再加入线程会执行此策略。默认不处理的话会扔出异常，打进日志。这个与任务处理的数据重要程度有关。如果数据是可丢弃的，那不需要额外处理。如果数据极其重要，那需要在这里采取措施防止数据丢失，如扔消息队列或者至少详细打入日志文件可追踪。
并发容器选择
案例一：电商网站中记录一次活动下各个商品售卖的数量。
场景分析：需要频繁按商品id做get和set，但是商品id（key）的数量相对稳定不会频繁增删
初级方案：选用HashMap，key为商品id，value为商品购买的次数。每次下单取出次数，增加后再写入
问题：HashMap线程不安全！在多次商品id写入后，如果发生扩容，在JDK1.7 之前，在并发场景下HashMap 会出现死循环，从而导致CPU 使用率居高不下。JDK1.8 中修复了HashMap 扩容导致的死循环问题，但在高并发场景下，依然会有数据丢失以及不准确的情况出现。
选型：Hashtable 不推荐，锁太重，选ConcurrentHashMap 确保高并发下多线程的安全性
案例二：在一次活动下，为每个用户记录浏览商品的历史和次数。
场景分析：每个用户各自浏览的商品量级非常大，并且每次访问都要更新次数，频繁读写
初级方案：为确保线程安全，采用上面的思路，ConcurrentHashMap
问题：ConcurrentHashMap 内部机制在数据量大时，会把链表转换为红黑树。而红黑树在高并发情况下，删除和插入过程中有个平衡的过程，会牵涉到大量节点，因此竞争锁资源的代价相对比较高
选型：用跳表，ConcurrentSkipListMap将key值分层，逐个切段，增删效率高于ConcurrentHashMap
结论：如果对数据有强一致要求，则需使用Hashtable；在大部分场景通常都是弱一致性的情况下，使用ConcurrentHashMap 即可；如果数据量级很高，且存在大量增删改操作，则可以考虑使用
ConcurrentSkipListMap。
案例三：在活动中，创建一个用户列表，记录冻结的用户。一旦冻结，不允许再下单抢购，但是可
以浏览。
场景分析：违规被冻结的用户不会太多，但是绝大多数非冻结用户每次抢单都要去查一下这个列表。低频写，高频读。
初级方案：ArrayList记录要冻结的用户id
问题：ArrayList对冻结用户id的插入和读取操作在高并发时，线程不安全。Vector可以做到线程安全，但并发性能差，锁太重。
选型：综合业务场景，选CopyOnWriteArrayList，会占空间，但是也仅仅发生在添加新冻结用户的时候。绝大多数的访问在非冻结用户的读取和比对上，不会阻塞。