Skip to content

J
java-concurrent-action
Commit 8e94c09c authored by 刘凯's avatar 刘凯
Browse files
Options

demo

parents
Show whitespace changes
Inline Side-by-side
Showing with 964 additions and 0 deletions
build.gradle 0 → 100644
View file @ 8e94c09c
buildscript {
ext {
springBootVersion = '2.0.6.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}
apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'org.springframework.boot'
apply plugin: 'io.spring.dependency-management'
group = 'com.example'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8
repositories {
mavenCentral()
}
dependencies {
implementation('org.springframework.boot:spring-boot-starter-web')
testImplementation('org.springframework.boot:spring-boot-starter-test')
compile("com.google.guava:guava:18.0")
}
src/main/java/com/example/demo/threadDesignMode/Observable.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode;
public interface Observable {
}
src/main/java/com/example/demo/threadDesignMode/ObservableThread.java 0 → 100644
View file @ 8e94c09c
/**
* @author liukai
* @Description: TODO
* @date 2018/10/293:03 PM
*/
public class ObservableThread {
}
src/main/java/com/example/demo/threadDesignMode/Task.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode;
public interface Task {
}
src/main/java/com/example/demo/threadDesignMode/TaskLifecycle.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode;
public interface TaskLifecycle {
}
src/main/java/com/example/demo/threadDesignMode/threadObservable/threadObservabledemo/Observable.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode.threadObservable.threadObservabledemo;
public interface Observable {
enum Cycle{
STARTED,RUNNING,DONE,ERROR
}
Cycle getCycle();
void start();
void interrupt();
}
src/main/java/com/example/demo/threadDesignMode/threadObservable/threadObservabledemo/ObservableThread.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode.threadObservable.threadObservabledemo;
import java.util.concurrent.TimeUnit;
/**
* @author liukai
* @Description: TODO
* @date 2018/10/293:03 PM
*/
public class ObservableThread<T> extends Thread implements Observable{
private final TaskLifecycle<T> lifecycle;
private final Task<T> task;
private Cycle cycle;
public ObservableThread(TaskLifecycle<T> taskLifecycle,Task<T> task){
super();
if(task==null){
throw new IllegalArgumentException("the task is required");
}
this.lifecycle = taskLifecycle;
this.task =task;
}
@Override
public final void run(){
this.update(Cycle.STARTED,null,null);
try {
this.update(Cycle.RUNNING,null,null);
T result = this.task.call();
this.update(Cycle.DONE,result,null);
} catch (Exception e) {
this.update(Cycle.ERROR,null,e);
}
}
private void update(Cycle cycle,T result,Exception e){
this.cycle = cycle;
if(lifecycle==null){
return ;
}
try {
switch (cycle){
case DONE:
this.lifecycle.onFinish(currentThread(),result);
case ERROR:
this.lifecycle.onError(currentThread(),e);
case RUNNING:
this.lifecycle.onRunning(currentThread());
case STARTED:
this.lifecycle.onStart(currentThread());
}
} catch (Exception e1) {
if(cycle ==Cycle.ERROR){
throw e1;
}
}
}
@Override
public Cycle getCycle() {
return this.cycle;
}
public static void main(String[] args) {
final TaskLifecycle<String> lifecycle = new TaskLifecycle.EmptyLifecycle<String>(){
@Override
public void onFinish(Thread thread, String result) {
System.out.println("the result is "+ result);
}
};
Observable observableThread = new ObservableThread<String>(lifecycle,()->{
try {
TimeUnit.SECONDS.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("finish done");
return "Hello 1";
});
observableThread.start();
}
}
src/main/java/com/example/demo/threadDesignMode/threadObservable/threadObservabledemo/Task.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode.threadObservable.threadObservabledemo;
public interface Task<T> {
T call();
}
src/main/java/com/example/demo/threadDesignMode/threadObservable/threadObservabledemo/TaskLifecycle.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.threadDesignMode.threadObservable.threadObservabledemo;
public interface TaskLifecycle<T> {
void onStart(Thread thread);
void onRunning(Thread thread);
void onFinish(Thread thread,T result);
void onError(Thread thead, Exception e);
class EmptyLifecycle<T> implements TaskLifecycle<T>{
@Override
public void onStart(Thread thread) {
}
@Override
public void onRunning(Thread thread) {
}
@Override
public void onFinish(Thread thread, T result) {
}
@Override
public void onError(Thread thead, Exception e) {
}
}
}
src/test/java/com/example/demo/TestThreadPoolExecutor/Task.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.TestThreadPoolExecutor;
public class Task implements Runnable{
private int id ;
private String name ;
public Task(int id, String name){
this.id = id;
this.name = name;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public void run() {
//
System.out.println("当前线程id和名称为:" + this.id +", " + this.name);
try {
Thread.sleep(5*1000);
} catch (Exception e) {
e.printStackTrace();
}
}
public String toString(){
return "{ id : " + this.id + ", name : " + this.name + "}";
}
}
src/test/java/com/example/demo/TestThreadPoolExecutor/TestThreadPoolExecutor.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.TestThreadPoolExecutor;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy;
import java.util.concurrent.ThreadPoolExecutor.DiscardPolicy;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
public class TestThreadPoolExecutor {
public static void main(String[] args) {
ThreadPoolExecutor pool = new ThreadPoolExecutor(
2, //核心的线程数量
100, //最大的线程数量
10, //空闲数值
TimeUnit.SECONDS, //空闲时间单位
//new ArrayBlockingQueue<Runnable>(5)
new LinkedBlockingQueue<Runnable>()
//new x()
); //队列容器
Task t1 = new Task(1, "任务" + 1);
Task t2 = new Task(2, "任务" + 2);
Task t3 = new Task(3, "任务" + 3);
Task t4 = new Task(4, "任务" + 4);
Task t5 = new Task(5, "任务" + 5);
//Task t1 = new Task(1, "任务" + 1);
pool.execute(t1);
pool.execute(t2);
pool.execute(t3);
pool.execute(t4);
pool.execute(t5);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(pool.getQueue().size());
try {
Thread.sleep(6000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(pool.getQueue().size());
pool.shutdown();
}
}
src/test/java/com/example/demo/concurrent/UseCountDownLatch.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.concurrent;
import java.util.concurrent.CountDownLatch;
public class UseCountDownLatch {
public static void main(String[] args) {
final CountDownLatch countDown = new CountDownLatch(2);
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println("进入线程t1" + "等待其他线程处理完成...");
countDown.await();
System.out.println("t1线程继续执行...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
},"t1");
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println("t2线程进行初始化操作...");
Thread.sleep(3000);
System.out.println("t2线程初始化完毕，通知t1线程继续...");
countDown.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
try {
System.out.println("t3线程进行初始化操作...");
Thread.sleep(4000);
System.out.println("t3线程初始化完毕，通知t1线程继续...");
countDown.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t1.start();
t2.start();
t3.start();
}
}
src/test/java/com/example/demo/concurrent/UseCyclicBarrier.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.concurrent;
import java.io.IOException;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class UseCyclicBarrier {
static class Runner implements Runnable {
private CyclicBarrier barrier;
private String name;
public Runner(CyclicBarrier barrier, String name) {
this.barrier = barrier;
this.name = name;
}
@Override
public void run() {
try {
Thread.sleep(1000 * (new Random()).nextInt(5));
System.out.println(name + " 准备OK.");
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(name + " Go!!");
}
}
public static void main(String[] args) throws IOException, InterruptedException {
CyclicBarrier barrier = new CyclicBarrier(3); // 3
ExecutorService executor = Executors.newFixedThreadPool(3);
executor.submit(new Thread(new Runner(barrier, "zhangsan")));
executor.submit(new Thread(new Runner(barrier, "lisi")));
executor.submit(new Thread(new Runner(barrier, "wangwu")));
executor.shutdown();
}
}
\ No newline at end of file
src/test/java/com/example/demo/concurrent/UseFuture.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.concurrent;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
public class UseFuture implements Callable<String>{
private String para;
public UseFuture(String para){
this.para = para;
}
/**
* 这里是真实的业务逻辑，其执行可能很慢
*/
@Override
public String call() throws Exception {
//模拟执行耗时
Thread.sleep(5000);
String result = this.para + "处理完成";
return result;
}
//主控制函数
public static void main(String[] args) throws Exception {
String queryStr = "query";
//构造FutureTask，并且传入需要真正进行业务逻辑处理的类,该类一定是实现了Callable接口的类
FutureTask<String> future = new FutureTask<String>(new UseFuture(queryStr));
FutureTask<String> future2 = new FutureTask<String>(new UseFuture(queryStr));
//创建一个固定线程的线程池且线程数为1,
ExecutorService executor = Executors.newFixedThreadPool(2);
//这里提交任务future,则开启线程执行RealData的call()方法执行
//submit和execute的区别： 第一点是submit可以传入实现Callable接口的实例对象， 第二点是submit方法有返回值
executor.submit(future); //单独启动一个线程去执行的
executor.submit(future2);
System.out.println("请求完毕");
try {
//这里可以做额外的数据操作，也就是主程序执行其他业务逻辑
System.out.println("处理实际的业务逻辑...");
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
//调用获取数据方法,如果call()方法没有执行完成,则依然会进行等待
System.out.println("数据：" + future.get());
System.out.println("数据：" + future2.get());
executor.shutdown();
}
}
src/test/java/com/example/demo/concurrent/UseSemaphore.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.concurrent;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
public class UseSemaphore {
public static void main(String[] args) {
// 线程池
ExecutorService exec = Executors.newCachedThreadPool();
// 只能5个线程同时访问
final Semaphore semp = new Semaphore(5);
// 模拟20个客户端访问
for (int index = 0; index < 20; index++) {
final int NO = index;
Runnable run = new Runnable() {
public void run() {
try {
// 获取许可
semp.acquire();
System.out.println("Accessing: " + NO);
//模拟实际业务逻辑
Thread.sleep((long) (Math.random() * 10000));
System.out.println("Accessing: " + NO + " release");
} catch (InterruptedException e) {
}finally {
// 访问完后，释放
semp.release();
}
}
};
exec.execute(run);
}
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
//System.out.println(semp.getQueueLength());
// 退出线程池
exec.shutdown();
}
}
\ No newline at end of file
src/test/java/com/example/demo/condition/ConditionSingle.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.condition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ConditionSingle {
private static final Logger logger = LoggerFactory.getLogger(ConditionSingle.class);
static final Lock lock = new ReentrantLock();
static final Condition condition = lock.newCondition();
public static void main(String[] args) throws Exception {
final Thread thread1 = new Thread("Thread 1 ") {
@Override
public void run() {
lock.lock(); // 线程 1获取 lock
logger.info(Thread.currentThread().getName() + " 正在运行 .....");
try {
Thread.sleep(2 * 1000);
logger.info(Thread.currentThread().getName() + " 停止运行, 等待一个 signal ");
condition.await(); // 调用 condition.await 进行释放锁, 将当前节点封装成一个 Node 放入 Condition Queue 里面, 等待唤醒
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info(Thread.currentThread().getName() + " 获取一个 signal, 继续执行 ");
lock.unlock(); // 释放锁
}
};
thread1.start(); // 线程 1 线运行
Thread.sleep(1 * 1000);
Thread thread2 = new Thread("Thread 2 ") {
@Override
public void run() {
lock.lock(); // 线程 2获取lock
logger.info(Thread.currentThread().getName() + " 正在运行.....");
thread1.interrupt(); // 对线程1 进行中断 看看中断后会怎么样? 结果 线程 1还是获取lock, 并且最后还进行 lock.unlock()操作
try {
Thread.sleep(2 * 1000);
} catch (Exception e) {
}
condition.signal(); // 发送唤醒信号 从 AQS 的 Condition Queue 里面转移 Node 到 Sync Queue
logger.info(Thread.currentThread().getName() + " 发送一个 signal ");
logger.info(Thread.currentThread().getName() + " 发送 signal 结束");
lock.unlock(); // 线程 2 释放锁
}
};
thread2.start();
}
}
\ No newline at end of file
src/test/java/com/example/demo/condition/ConditionTest.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.condition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ConditionTest {
private static final Logger logger = LoggerFactory.getLogger(ConditionTest.class);
static final Lock lock = new ReentrantLock();
static final Condition condition = lock.newCondition();
public static void main(String[] args) throws Exception {
final Thread thread1 = new Thread("Thread 1 ") {
@Override
public void run() {
lock.lock(); // 线程 1获取 lock
logger.info(Thread.currentThread().getName() + " 正在运行 .....");
try {
Thread.sleep(2 * 1000);
logger.info(Thread.currentThread().getName() + " 停止运行, 等待一个 signal ");
condition.await(); // 调用 condition.await 进行释放锁, 将当前节点封装成一个 Node 放入 Condition Queue 里面, 等待唤醒
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info(Thread.currentThread().getName() + " 获取一个 signal, 继续执行 ");
lock.unlock(); // 释放锁
}
};
thread1.start(); // 线程 1 线运行
Thread.sleep(1 * 1000);
Thread thread2 = new Thread("Thread 2 ") {
@Override
public void run() {
lock.lock(); // 线程 2获取lock
logger.info(Thread.currentThread().getName() + " 正在运行.....");
thread1.interrupt(); // 对线程1 进行中断 看看中断后会怎么样? 结果 线程 1还是获取lock, 并且最后还进行 lock.unlock()操作
try {
Thread.sleep(2 * 1000);
} catch (Exception e) {
}
condition.signal(); // 发送唤醒信号 从 AQS 的 Condition Queue 里面转移 Node 到 Sync Queue
logger.info(Thread.currentThread().getName() + " 发送一个 signal ");
logger.info(Thread.currentThread().getName() + " 发送 signal 结束");
lock.unlock(); // 线程 2 释放锁
}
};
thread2.start();
}
}
\ No newline at end of file
src/test/java/com/example/demo/condition/UseManyCondition.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.condition;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class UseManyCondition {
private ReentrantLock lock = new ReentrantLock();
private Condition c1 = lock.newCondition();
private Condition c2 = lock.newCondition();
public void m1(){
try {
lock.lock();
System.out.println("当前线程：" +Thread.currentThread().getName() + "进入方法m1等待..");
c1.await();
System.out.println("当前线程：" +Thread.currentThread().getName() + "方法m1继续..");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void m2(){
try {
lock.lock();
System.out.println("当前线程：" +Thread.currentThread().getName() + "进入方法m2等待..");
c1.await();
System.out.println("当前线程：" +Thread.currentThread().getName() + "方法m2继续..");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void m3(){
try {
lock.lock();
System.out.println("当前线程：" +Thread.currentThread().getName() + "进入方法m3等待..");
c2.await();
System.out.println("当前线程：" +Thread.currentThread().getName() + "方法m3继续..");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void m4(){
try {
lock.lock();
System.out.println("当前线程：" +Thread.currentThread().getName() + "唤醒..");
c1.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public void m5(){
try {
lock.lock();
System.out.println("当前线程：" +Thread.currentThread().getName() + "唤醒..");
c2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public static void main(String[] args) {
final UseManyCondition umc = new UseManyCondition();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
umc.m1();
}
},"t1");
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
umc.m2();
}
},"t2");
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
umc.m3();
}
},"t3");
Thread t4 = new Thread(new Runnable() {
@Override
public void run() {
umc.m4();
}
},"t4");
Thread t5 = new Thread(new Runnable() {
@Override
public void run() {
umc.m5();
}
},"t5");
t1.start(); // c1
t2.start(); // c1
t3.start(); // c2
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t4.start(); // c1
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t5.start(); // c2
}
}
src/test/java/com/example/demo/limit/AccessRateLimit.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.limit;
import com.google.common.util.concurrent.RateLimiter;
/**
* @author liukai
* @Description: TODO
* @date 2018/10/265:28 PM
*/
public class AccessRateLimit {
public static void main(String[] args) {
// RateLimiter rateLimiter = RateLimiter.create(5);
// rateLimiter.
}
}
src/test/java/com/example/demo/lock/UseReentrantReadWriteLock.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
public class UseReentrantReadWriteLock {
private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
private ReadLock readLock = rwLock.readLock();
private WriteLock writeLock = rwLock.writeLock();
public void read(){
try {
readLock.lock();
System.out.println("当前线程:" + Thread.currentThread().getName() + "进入...");
Thread.sleep(3000);
System.out.println("当前线程:" + Thread.currentThread().getName() + "退出...");
} catch (Exception e) {
e.printStackTrace();
} finally {
readLock.unlock();
}
}
public void write(){
try {
writeLock.lock();
System.out.println("当前线程:" + Thread.currentThread().getName() + "进入...");
Thread.sleep(3000);
System.out.println("当前线程:" + Thread.currentThread().getName() + "退出...");
} catch (Exception e) {
e.printStackTrace();
} finally {
writeLock.unlock();
}
}
public static void main(String[] args) {
final UseReentrantReadWriteLock urrw = new UseReentrantReadWriteLock();
Thread t1 = new Thread(new Runnable() {
@Override
public void run() {
urrw.read();
}
}, "t1");
Thread t2 = new Thread(new Runnable() {
@Override
public void run() {
urrw.read();
}
}, "t2");
Thread t3 = new Thread(new Runnable() {
@Override
public void run() {
urrw.write();
}
}, "t3");
Thread t4 = new Thread(new Runnable() {
@Override
public void run() {
urrw.write();
}
}, "t4");
t1.start();
t2.start();
// t1.start(); // R
// t3.start(); // W
t3.start();
t4.start();
}
}
src/test/java/com/example/demo/move/Test.java 0 → 100644
View file @ 8e94c09c
/**
* @author liukai
* @Description: TODO
* @date 2018/10/2512:46 PM
*/
public class Test {
}
src/test/java/com/example/demo/move/intTest.java 0 → 100644
View file @ 8e94c09c
package com.example.demo.move;
/**
* @author liukai
* @Description: TODO
* @date 2018/10/2512:46 PM
*/
public class intTest {
public static void main(String[] args) {
int number = 10;
//原始数二进制
printInfo(number);
number = number << 1;
//左移一位
printInfo(number);
number = 10;
number = number >> 1;
//右移一位
printInfo(number);
}
/**
* 输出一个int的二进制数
* @param num
*/
private static void printInfo(int num){
System.out.println(Integer.toBinaryString(num));
}
}
添加项目.patch 0 → 100644
View file @ 8e94c09c
Index: .idea/vcs.xml
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- .idea/vcs.xml (date 1540209232000)
+++ .idea/vcs.xml (date 1540209232000)
@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="VcsDirectoryMappings">
+ <mapping directory="" vcs="Git" />
+ </component>
+</project>
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment