Skip to content

Commit 8058141

Browse files
CodePrometheusCodePrometheus
committed
feat
1 parent 83c3f02 commit 8058141

File tree

13 files changed

+775
-12
lines changed

13 files changed

+775
-12
lines changed

‎Future/Future.iml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,15 @@
77
</content>
88
<orderEntry type="inheritedJdk" />
99
<orderEntry type="sourceFolder" forTests="false" />
10+
<orderEntry type="module-library">
11+
<library name="JUnit4">
12+
<CLASSES>
13+
<root url="jar://$MAVEN_REPOSITORY$/junit/junit/4.13.1/junit-4.13.1.jar!/" />
14+
<root url="jar://$MAVEN_REPOSITORY$/org/hamcrest/hamcrest-core/1.3/hamcrest-core-1.3.jar!/" />
15+
</CLASSES>
16+
<JAVADOC />
17+
<SOURCES />
18+
</library>
19+
</orderEntry>
1020
</component>
1121
</module>

‎Future/src/future/MultiFutures.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ public static void main(String[] args) {
2323

2424
//获取结果,将会每两个打印一次
2525
for (int i = 0; i < 20; i++) {
26-
var future = futures.get(i);
26+
Future future = futures.get(i);
2727
try {
2828
System.out.println(future.get());
2929
} catch (InterruptedException | ExecutionException e) {

‎Interview/Interview.iml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,6 @@
77
</content>
88
<orderEntry type="inheritedJdk" />
99
<orderEntry type="sourceFolder" forTests="false" />
10+
<orderEntry type="library" name="Java EE 6-Java EE 6" level="project" />
1011
</component>
1112
</module>

‎Interview/src/Con_Pro_1.java‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
import java.util.PriorityQueue;
2+
import java.util.Queue;
3+
import java.util.concurrent.locks.Condition;
4+
import java.util.concurrent.locks.Lock;
5+
import java.util.concurrent.locks.ReentrantLock;
6+
7+
/**
8+
* 生产者消费者
9+
*
10+
* @Author: Starry
11+
* @Date: 08-23-2022 12:06
12+
*/
13+
public class Con_Pro_1 {
14+
private int size = 10;
15+
private Queue<Integer> queue = new PriorityQueue<>(size);
16+
private Lock lock = new ReentrantLock();
17+
private Condition consumer = lock.newCondition();
18+
private Condition produce = lock.newCondition();
19+
20+
class Consumer extends Thread {
21+
@Override
22+
public void run() {
23+
while (true) {
24+
lock.lock();
25+
try {
26+
while (queue.size() == 0) {
27+
System.out.println("队列空，等待数据");
28+
try {
29+
consumer.await(); // 消费者等待
30+
} catch (InterruptedException e) {
31+
throw new RuntimeException(e);
32+
}
33+
}
34+
queue.poll();
35+
produce.signalAll(); // 唤醒生产者
36+
System.out.println("队列剩余: " + queue.size() + " 个元素");
37+
} finally {
38+
lock.unlock();
39+
}
40+
}
41+
}
42+
}
43+
44+
class produce extends Thread {
45+
@Override
46+
public void run() {
47+
while (true) {
48+
lock.lock();
49+
try {
50+
while (queue.size() == size) {
51+
System.out.println("队列已满");
52+
try {
53+
produce.await(); // 生产者等待
54+
} catch (InterruptedException e) {
55+
throw new RuntimeException(e);
56+
}
57+
}
58+
queue.offer(1);
59+
consumer.signalAll(); // 唤醒消费者
60+
System.out.println("队列现存: " + queue.size() + " 个元素");
61+
} finally {
62+
lock.unlock();
63+
}
64+
}
65+
}
66+
}
67+
68+
public static void main(String[] args) {
69+
Con_Pro_1 test = new Con_Pro_1();
70+
Consumer con = test.new Consumer();
71+
Con_Pro_1.produce pro = test.new produce();
72+
pro.start();
73+
con.start();
74+
}
75+
}

‎Interview/src/ConcurrencyValue1.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
import java.util.ArrayList;
2+
import java.util.Collections;
3+
import java.util.List;
4+
import java.util.concurrent.CountDownLatch;
5+
import java.util.concurrent.ExecutorService;
6+
import java.util.concurrent.Executors;
7+
import java.util.concurrent.Semaphore;
8+
9+
public class ConcurrencyValue1 {
10+
11+
public static void main(String[] args) {
12+
int count = 10000000;
13+
List<Long> listKey = new ArrayList<>();
14+
for (int i = 0; i < count; i++) {
15+
listKey.add(114560315500000000L + i);
16+
}
17+
ConcurrencyValue1 concurrencyTest = new ConcurrencyValue1();
18+
List<Long> valueList1 = concurrencyTest.getValueList1(listKey);
19+
System.out.println("====>> getValueList1 valueList.size: " + valueList1.size());
20+
}
21+
22+
/**
23+
* 模拟一千万请求数据，创建一个线程池10个线程控制最大并发数，每个线程一次处理10万条，开启100个线程。（用于控制每批次线程的数据量业务场景）
24+
*
25+
* @param listKey 请求处理的总数据量
26+
* @return
27+
*/
28+
public List<Long> getValueList1(List<Long> listKey) {
29+
30+
/**
31+
（1）newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
32+
（2）newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
33+
（3）newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
34+
（4）newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
35+
*/
36+
// 创建一个线程池
37+
final ExecutorService executorService = Executors.newFixedThreadPool(10); //用于控制同时并发执行的线程数，使用线程池创建资源提高效率
38+
List<Long> list_val = Collections.synchronizedList(new ArrayList<>()); //保证多线程操作的是同一个List
39+
try {
40+
long t1 = System.currentTimeMillis();
41+
int max_one_batch = 100000; // 一批次最多处理100000条数据
42+
List<List<Long>> newList = ListUtils.splitList(listKey, max_one_batch);
43+
int runSize = newList.size(); // 开启的线程数
44+
45+
/**
46+
* CountDownLanch 只需要在子线程执行之前, 赋予初始化countDownLanch, 并赋予线程数量为初始值。
47+
* 每个线程执行完毕的时候, 就countDown一下。主线程只需要调用await方法, 可以等待所有子线程执行结束。
48+
*/
49+
final CountDownLatch countDownLatch = new CountDownLatch(runSize); //计数器
50+
51+
/**
52+
* Semaphore（信号量）是用来控制同时访问特定资源的线程数量，拿到信号量的线程可以进入，否则就等待。
53+
* 通过acquire()和release()获取和释放访问许可。
54+
*/
55+
final Semaphore semaphore = new Semaphore(runSize); //信号量
56+
// 循环创建线程
57+
for (int j = 0; j < runSize; j++) {
58+
final int i = j;
59+
executorService.execute(() -> {
60+
try {
61+
semaphore.acquire();
62+
// 执行程序
63+
List<Long> subList = newList.get(i);
64+
List<Long> sub_ret = getValue(subList);
65+
list_val.addAll(sub_ret);
66+
System.out.println(Thread.currentThread().getName() + ": 当前线程/总线程: [" + i + "/" + runSize + "]"
67+
+ ", 处理的数据条数：" + subList.size());
68+
semaphore.release();
69+
} catch (Exception e) {
70+
System.out.println(e.getMessage());
71+
} finally {
72+
// 计数器减一
73+
countDownLatch.countDown();
74+
}
75+
});
76+
}
77+
78+
// 阻塞主线程等待所有线程执行完成
79+
countDownLatch.await();
80+
// 所有线程执行完，之后才能执行的部分
81+
long t2 = System.currentTimeMillis();
82+
System.out.printf("Call getValueList1 success... ret: {} size, threadCount: {}, costs time: {} ms" + list_val.size() + runSize + (t2 - t1));
83+
return list_val;
84+
} catch (Exception e) {
85+
return null;
86+
} finally {
87+
// 关闭线程池
88+
executorService.shutdown();
89+
}
90+
}
91+
92+
private List<Long> getValue(List<Long> listKey) {
93+
//具体操作省略
94+
return listKey;
95+
}
96+
97+
}

‎Interview/src/ConcurrencyValue2.java‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
import java.util.ArrayList;
2+
import java.util.Collections;
3+
import java.util.List;
4+
import java.util.concurrent.CountDownLatch;
5+
import java.util.concurrent.ExecutorService;
6+
import java.util.concurrent.Executors;
7+
import java.util.concurrent.Semaphore;
8+
9+
public class ConcurrencyValue2 {
10+
11+
public static void main(String[] args) {
12+
int count = 10000000;
13+
List<Long> listKey = new ArrayList<>();
14+
for (int i = 0; i < count; i++) {
15+
listKey.add(114560315500000000L + i);
16+
}
17+
ConcurrencyValue2 concurrencyTest = new ConcurrencyValue2();
18+
List<Long> valueList2 = concurrencyTest.getValueList2(listKey);
19+
System.out.println("====>> getValueList2 valueList.size: " + valueList2.size());
20+
}
21+
22+
23+
/**
24+
* 模拟一千万请求数据，创建一个线程池10个线程控制最大并发数，设置请求的线程数为100，平均处理这一千万请求数据。（用于控制请求线程数的业务场景）
25+
*
26+
* @param listKey 请求处理的总数据量
27+
* @return
28+
*/
29+
public List<Long> getValueList2(List<Long> listKey) {
30+
31+
/**
32+
（1）newCachedThreadPool 创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
33+
（2）newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
34+
（3）newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
35+
（4）newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
36+
*/
37+
// 创建一个线程池
38+
final ExecutorService executorService = Executors.newFixedThreadPool(10); //用于控制同时并发执行的线程数，使用线程池创建资源提高效率
39+
List<Long> list_val = Collections.synchronizedList(new ArrayList<>()); //保证多线程操作的是同一个List
40+
try {
41+
long t1 = System.currentTimeMillis();
42+
int threadCount = 100; // 请求的线程数
43+
List<List<Long>> newList = ListUtils.avgList(listKey, threadCount);
44+
int runSize = threadCount; // 请求开启的线程数
45+
46+
/**
47+
* CountDownLanch 只需要在子线程执行之前, 赋予初始化countDownLanch, 并赋予线程数量为初始值。
48+
* 每个线程执行完毕的时候, 就countDown一下。主线程只需要调用await方法, 可以等待所有子线程执行结束。
49+
*/
50+
final CountDownLatch countDownLatch = new CountDownLatch(runSize); // 计数器
51+
52+
/**
53+
* Semaphore（信号量）是用来控制同时访问特定资源的线程数量，拿到信号量的线程可以进入，否则就等待。
54+
* 通过acquire()和release()获取和释放访问许可。
55+
*/
56+
final Semaphore semaphore = new Semaphore(runSize); // 信号量
57+
// 循环创建线程
58+
for (int j = 0; j < runSize; j++) {
59+
final int i = j;
60+
executorService.execute(() -> {
61+
try {
62+
semaphore.acquire();
63+
// 执行程序
64+
List<Long> subList = newList.get(i);
65+
List<Long> sub_ret = getValue(subList);
66+
list_val.addAll(sub_ret);
67+
System.out.println(Thread.currentThread().getName() + ": 当前线程/总线程: [" + i + "/" + runSize + "]"
68+
+ ", 处理的数据条数：" + subList.size());
69+
semaphore.release();
70+
} catch (Exception e) {
71+
e.printStackTrace();
72+
} finally {
73+
// 计数器减一
74+
countDownLatch.countDown();
75+
}
76+
});
77+
}
78+
79+
// 阻塞主线程等待所有线程执行完成
80+
countDownLatch.await();
81+
// 所有线程执行完，之后才能执行的部分
82+
long t2 = System.currentTimeMillis();
83+
System.out.printf("Call getValueList2 success... ret: {} size, threadCount: {}, costs time: {} ms" + list_val.size() + runSize + (t2 - t1));
84+
return list_val;
85+
} catch (Exception e) {
86+
return null;
87+
} finally {
88+
// 关闭线程池
89+
executorService.shutdown();
90+
}
91+
}
92+
93+
private List<Long> getValue(List<Long> listKey) {
94+
//具体操作省略
95+
return listKey;
96+
}
97+
}

0 commit comments

Comments
 (0)