总结:
1:对于map/set的选择使用
不加锁
HashMap
TreeMap 红黑树 本身是排好顺序的
LinkedHashMap
加锁 并发性不是很高
Hashtable
Collections.sychronizedXXX //不加锁非同步的容器转变为加锁同步的容器
ConcurrentHashMap 并发性要求很高
ConcurrentSkipListMap 并发性要求很高且排序
2:队列
ArrayList 不需要加锁,同步
LinkedList 不需要加锁,同步
Collections.synchronizedXXX Vector 并发量比较低的情况下
CopyOnWriteList 适合写少读多的环境 用于监听器 队列
高并发
Queue
CocurrentLinkedQueue //concurrentArrayQueue没有
BlockingQueue //阻塞队列
LinkedBlockingQueue //无界队列,什么时候队列满了,什么时候出错
ArrayBlockingQueue //有界队列,容器里能装的元素个数是固定的
TransferQueue //主要适用于:实时消息处理( 我扔的东西你必须处理掉,否则我就不走了) 游戏服务器转发消息用的多
SynchronusQueue //没有容量的队列,来的任何东西消费者必须里面消费掉,不然就出问题了
DelayQueue执行定时任务
demo1
/*
http://blog.csdn.net/sunxianghuang/article/details/52221913(ConcurrentSkipListMap)
http://www.educity.cn/java/498061.html
阅读concurrentskiplistmap
100个线程,每个线程装10000个数据往map里,每个线程指向完门闩-1,主线程等着,所有线程执行完之后,计算时间
并发容器
Map<String, String> map = new Hashtable<>(); 1147 //锁定整个容器
Map<String, String> map = new ConcurrentHashMap<>();//高并发 1039 (多线程情况下效率比Hashtable高一些)
//容器分别16段,只锁定1段
Map<String, String> map = new ConcurrentSkipListMap<>(); //高并发并且排序 调表数据结构
Map<String, String> map = new HashMap<>(); //Collections.synchronizedXXX 带锁的实现
TreeMap //默认排好顺序
*/
package bingfaDemo.bingfa.T25;
import java.util.Arrays;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
public class T01_ConcurrentMap {
public static void main(String[] args) {
Map<String, String> map = new ConcurrentHashMap<>();//高并发
Random r = new Random();
Thread[] ths = new Thread[100];
CountDownLatch latch = new CountDownLatch(ths.length);//门闩 数值为100
long start = System.currentTimeMillis();
for(int i=0; i<ths.length; i++) {
ths[i] = new Thread(()->{
for(int j=0; j<10000; j++)
map.put("a" + r.nextInt(100000), "a" + r.nextInt(100000));
latch.countDown();
});
}
Arrays.asList(ths).forEach(t->t.start());
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
long end = System.currentTimeMillis();
System.out.println(end - start);
}
}
demo2
/*
写时复制容器 copy on write
多线程环境下,写时效率低,读时效率高
适合写少读多的环境
*/
package bingfaDemo.bingfa.T25;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
public class T02_CopyOnWriteList {
public static void main(String[] args) {
List<String> lists =
//new ArrayList<>(); //这个会出并发问题!
//new Vector();
new CopyOnWriteArrayList<>();
Random r = new Random();
Thread[] ths = new Thread[100];
for(int i=0; i<ths.length; i++) {
Runnable task = new Runnable() {
@Override
public void run() {
for(int i=0; i<1000; i++) lists.add("a" + r.nextInt(10000));
}
};
ths[i] = new Thread(task);
}
runAndComputeTime(ths);
System.out.println(lists.size());
}
static void runAndComputeTime(Thread[] ths) {
long s1 = System.currentTimeMillis();
Arrays.asList(ths).forEach(t->t.start());
Arrays.asList(ths).forEach(t->{
try {
t.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
long s2 = System.currentTimeMillis();
System.out.println(s2 - s1);
}
}
demo3
package bingfaDemo.bingfa.T25;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
public class T03_SynchronizedList {
public static void main(String[] args) {
List<String> strs = new ArrayList<>();//没有锁
List<String> strsSync = Collections.synchronizedList(strs);//加了锁的list
}
}
demo4
package bingfaDemo.bingfa.T25;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
public class T04_ConcurrentQueue {
public static void main(String[] args) {
Queue<String> strs = new ConcurrentLinkedQueue<>();
//链表实现的队列,无界队列,什么时候队列满了,什么时候出错
for(int i=0; i<10; i++) {
strs.offer("a" + i); //类似于add
//但是与add不同的是,返回一个boolean值,可以判断是否加入
}
System.out.println(strs);//[a0, a1, a2, a3, a4, a5, a6, a7, a8, a9]
System.out.println(strs.size());//10
System.out.println(strs.poll());//a0 拿走第一个,并删除
System.out.println(strs.size());//9
System.out.println(strs.peek());//a1 拿走第一个用一下,不删除
System.out.println(strs.size());//9
System.out.println(strs);
//双端队列Deque
}
}
demo5
package bingfaDemo.bingfa.T25;
/*
无界队列,什么时候队列满了,什么时候出错
链表实现同步阻塞式的容器
使用LinkedBlockingQueue可以自动实现阻塞式的生产者消费者模式
put()如果满了就会等待,take()如果空了就会等待
*/
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public class T05_LinkedBlockingQueue {
static BlockingQueue<String> strs = new LinkedBlockingQueue<>();
static Random r = new Random();
public static void main(String[] args) {
new Thread(() -> {
for (int i = 0; i < 100; i++) {
try {
strs.put("a" + i); //如果满了,就会等待
TimeUnit.MILLISECONDS.sleep(r.nextInt(1000));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "p1").start();
for (int i = 0; i < 5; i++) {
new Thread(() -> {
for (;;) {
try {
System.out.println(Thread.currentThread().getName() + " take -" + strs.take()); //如果空了,就会等待
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "c" + i).start();
}
}
}
demo6
package bingfaDemo.bingfa.T25;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/*
有界队列,容器里能装的元素个数是固定的
*/
public class T06_ArrayBlockingQueue {
static BlockingQueue<String> strs = new ArrayBlockingQueue<>(10);
static Random r = new Random();
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 10; i++) {
strs.put("a" + i);
}
strs.put("aaa"); //满了就会等待,程序阻塞
//strs.add("aaa"); //容器已经满了再加就会报异常
//strs.offer("aaa");//容器已经满了再加不会报异常,但是加不进去,得到的boolean是false
//strs.offer("aaa", 1, TimeUnit.SECONDS);
System.out.println(strs);
}
}
demo7
package bingfaDemo.bingfa.T25;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
/*
也是一个无界队列
但是每个元素都记在这个一个时间,还有多久才能被消费者拿走
可以用来 定时执行任务
*/
public class T07_DelayQueue {
static BlockingQueue<MyTask> tasks = new DelayQueue<>();
static Random r = new Random();
static class MyTask implements Delayed {
long runningTime;
MyTask(long rt) {
this.runningTime = rt;
}
@Override
public int compareTo(Delayed o) {
if(this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS))
return -1;
else if(this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS))
return 1;
else
return 0;
}
@Override
public long getDelay(TimeUnit unit) {//还要多长的时间就可以往外拿了
return unit.convert(runningTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
}
@Override
public String toString() {
return "" + runningTime;
}
}
public static void main(String[] args) throws InterruptedException {
long now = System.currentTimeMillis();
MyTask t1 = new MyTask(now + 1000);
MyTask t2 = new MyTask(now + 2000);
MyTask t3 = new MyTask(now + 1500);
MyTask t4 = new MyTask(now + 2500);
MyTask t5 = new MyTask(now + 500);
tasks.put(t1);
tasks.put(t2);
tasks.put(t3);
tasks.put(t4);
tasks.put(t5);
System.out.println(tasks);
for(int i=0; i<5; i++) {
System.out.println(tasks.take());
}
}
}
demo8
package bingfaDemo.bingfa.T25;
/*
transfer :消费者先启动,生产者生成东西直接看有没有消费者,有消费者的话就不往队列里放,
直接给消费者,经常用于更高并发的情况下,效率更高
若没有消费者线程,调用transfer的话就会阻塞
*/
import java.util.concurrent.LinkedTransferQueue;
public class T08_01 {
public static void main(String[] args) throws InterruptedException {
LinkedTransferQueue<String> strs = new LinkedTransferQueue<>();
new Thread(() -> {
try {
System.out.println(strs.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
strs.transfer("aaa");
}
}
package bingfaDemo.bingfa.T25;
import java.util.concurrent.LinkedTransferQueue;
/*
transfer :消费者先启动,生产者生成东西直接看有没有消费者,有消费者的话就不往队列里放,
直接给消费者,经常用于更高并发的情况下,效率更高
若没有消费者线程,调用transfer的话就会阻塞
我扔的东西你必须处理掉,否则我就不走了
主要适用于:实时消息处理
*/
public class T08_TransferQueue {
public static void main(String[] args) throws InterruptedException {
LinkedTransferQueue<String> strs = new LinkedTransferQueue<>();
//没有消费者线程,调用transfer的话就会阻塞
strs.transfer("aaa"); // 用put(),add(),offer()不会阻塞,只有用transer()会阻塞
new Thread(() -> {
try {
System.out.println(strs.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}
demo9
package bingfaDemo.bingfa.T25;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
/*
没有容量的队列,来的任何东西消费者必须里面消费掉,不然就出问题了
*/
public class T09_SynchronusQueue { //容量为0
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> strs = new SynchronousQueue<>();
new Thread(()->{
try {
System.out.println(strs.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
strs.put("aaa"); //阻塞等待消费者消费 put内部用的是transfer
//strs.add("aaa");
System.out.println(strs.size());
}
}