标签：node fileread 单线程 oid 完成 目标 cat 释放 并发
AQS简介
AbstractQueuedSynchronized,虽然类名开头是Abstract,但是他不是抽象类,意义就是说,单独使用它是没有意义的,依赖他去实现同步组件才有意义--相当于没模板方法模式
AQS实现的大致思路,它内部有一个双向的链表,链表的每一个节点都是一个Node的结构,线程会来尝试的获取锁,如果失败了那么它就将当前线程包装成一个Node节点,加入到同步队列中,前一个节点释放锁后,唤醒自己的后继节点,它实现的依赖于先进先出 (FIFO) 等待队列的阻塞锁和相关同步器
此类的设计目标是成为依靠单个原子 int 值来表示状态的大多数同步器的一个有用基础
自己的锁肯定要去实现lock接口,重写里面的方法,怎么重写呢?使用AQS,将AQS作为内部的帮助器类,重写里面的tryAcquir和tryRelease方法,因为lock()我们采用帮助器acquire的方法实现,而此方法会至少调用一次tryAcquire,同理,释放锁,我们重写帮助器的tryRelease,,,,我们只是简单的使用一下,完成加锁,释放锁,锁重入即可
他面临着两个问题
public class AQSDemo01 implements Lock {
private Helper helper = new Helper();
private class Helper extends AbstractQueuedSynchronizer {
@Override
protected boolean tryAcquire(int arg) {
int state = getState();
Thread t = Thread.currentThread();
if (state == 0) {
//如果当前状态的值等于预期的值,就把 当前状态的中修改成 arg的值...... ( 刚才掉坑了, compareAndSerState方法,会帮助我们去对比 手动输进去的0 和 当前的状态)
if (compareAndSetState(0, arg)) {
System.out.println("线程来了"+Thread.currentThread().getName()+" arg=="+arg);
setExclusiveOwnerThread(t);
return true;
}
} else if (getExclusiveOwnerThread() == t) {
setState(state + 1);
return true;
}
return false;
}
@Override
protected boolean tryRelease(int arg) {
if (Thread.currentThread() != getExclusiveOwnerThread()) {
throw new RuntimeException();
}
int state = getState() - arg;
boolean flag = false;
if (state == 0) {
setExclusiveOwnerThread(null);
flag = true;
}
setState(state);
return flag;
}
Condition newCondition() {
return new ConditionObject();
}
}
@Override
public void lock() {
helper.acquire(1);
}
@Override
public void lockInterruptibly() throws InterruptedException {
helper.acquireInterruptibly(1);
}
@Override
public boolean tryLock() {
return helper.tryAcquire(1);
}
@Override
public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
return helper.tryAcquireNanos(1, unit.toNanos(time));
}
@Override
public void unlock() {
helper.release(1);
}
@Override
public Condition newCondition() {
return helper.newCondition();
}
}测试类如下,锁正常
public class textAQS {
private int value=0;
private AQSDemo01 lock = new AQSDemo01();
public int next() {
lock.lock();
try {
Thread.sleep(300);
return value++;
} catch (InterruptedException e) {
e.printStackTrace();
throw new RuntimeException();
} finally {
lock.unlock();
}
}
/*
* 经典的验证 锁的重复问题 ,在单一的线程下, a() 想在 未释放锁 的前提下 调用b(),前提就是可冲入锁
* */
public void a() {
lock.lock();
System.out.println("a");
b();
lock.unlock();
}
public void b() {
lock.lock();
System.out.println("b");
lock.unlock();
}
public static void main(String[] args) {
textAQS m = new textAQS();
//测试可重入
new Thread(new Runnable() {
@Override
public void run() {
m.a();
}
}).start();
System.out.println("主线程=="+Thread.currentThread().getName());
ExecutorService executorService = Executors.newCachedThreadPool();
//从线程池拿出四条线程执行next任务,查看结果是否同步,同步
for (int i=0;i<4;i++){
executorService.execute(new Runnable() {
@Override
public void run() {
while(true)
System.out.println(Thread.currentThread().getName()+" "+m.next());
}
});
}
}
}打个比方
课代表等所有同学交完作业再交给老师,
1: 课代表等待所有的同学(线程)交作业 CountDownLatch cdl = new CountDownLatch(int 学生数)
2: 单个学生交完作用, cdl.countDown() --> 学生数减一
3: 主线程: cdl.await() 只要学生数不为零, 就等待
public class countDownLatch02 {
private static int []nums;
public countDownLatch02(int line){
nums=new int[line];
}
//分隔字符串数组,完成 当前行 ( 一行 ) 相加
public void colculate(String s ,int index,CountDownLatch count){
System.out.println("单行线程开始执行.. "+Thread.currentThread().getName());
String[] s1 = s.split(",");
int total=0;
for (String s2:s1) {
int i = Integer.parseInt(s2);
total+=i;
}
nums[index]=total;
System.out.println(Thread.currentThread().getName() +"线程 计算结果是=="+total);
count.countDown();
}
//分别计算没行的总值
public void sum(){
System.out.println("加总线程开始执行...");
int total =0;
for(int i=0;i<nums.length;i++){
total+=nums[i];
}
System.out.println("执行的结果是=="+total);
}
public static void main(String[] args) throws IOException, InterruptedException {
// 根据行数,
List<String> contents = readFile();
int size= contents.size();
CountDownLatch c = new CountDownLatch(size);
countDownLatch02 latch = new countDownLatch02(size);
System.out.println("zhuxianc");
// //创建出相应数目的线程,
for(int i=0;i<size;i++){
final int j =i;
new Thread(new Runnable() {
@Override
public void run() {
latch.colculate(contents.get(j),j,c);
}
}).start();
}
//在主线程中加总
// System.out.println("当前活跃的实现数"+Thread.activeCount());
/* while((Thread.activeCount())>2){ //自旋,等待其他线程执行完..
System.out.println("当前活跃的实现数"+Thread.activeCount());
}*/
c.await();
latch.sum();
}
/*
* 读取文件,将每一行存放进list数组...
* */
public static List<String> readFile() throws IOException {
List<String> list = new ArrayList<>();
String line=null;
BufferedReader bufferedReader = new BufferedReader(new FileReader("D:\\SETextMaven\\textcountDownLatch.txt"));
while ((line = bufferedReader.readLine())!=null){
list.add(line);
}
return list;
}
}/*
* 字面意思: 信号量 --> Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目
* 作用: 用来控制同时访问某些特定资源的线程数量,协调各个线程合理使用公共资源
* 简介: Semaphore 可以用来维护当前访问自身的线程个数,并提供了同步机制,比如实现一个文件的允许的并发数
* 应用场景:
* 开启 30条线程 把 一万个文件的内容读取到内存
* 使用Semaphore允许10个线程可以并发执行,将内存中的数据写回数据库
*
* 模拟高并发
* */
public class semaphore {
public static void main(String[] args) {
final int tNum =30;
// ExecutorService executorService;
// executorService = new Executors.newFixedThreadPool();
Semaphore semaphore = new Semaphore(5);//允许一次性允许 并发执行的线程数
for (int i=0;i<100;i++){
new Thread(()->{
try {
semaphore.acquire(); //当前线程获取 Semaphore 的 许可证
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()+"开始任务");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
semaphore.release(); // 归还许可证
}).start();
}
}
}实例代码:
/*
* 它允许一组线程相互等待,直到达到某个公共的屏障点,
* 也就是说,所有的线程必须相互等待,--> 直到所有线程都 满足屏障的要求--> 执行后面的任务
*
* 开会:
* await() 屏障 后续的任务
* 公司里的所有人都去开会,--> 先到的人等待迟到的人--> 人到齐了,开会....
* */
/*
* 应用场景,多线程计算数据,最后合并计算结果
* */
/*
* 模拟开会...
* */
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrier01 {
public void meeting(CyclicBarrier cyclicBarrier){
System.out.println(Thread.currentThread().getName()+"到达会议室...");
try {
cyclicBarrier.await(); //此线程等待..
System.out.println(Thread.currentThread().getName()+"准备开会..");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
CyclicBarrier01 c = new CyclicBarrier01();
// 构造函数1:
// 创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，
// 但它不会在启动 barrier 时执行预定义的操作。
//参数：
//parties - 在启动 barrier 前必须调用 await() 的线程数
//抛出：
//IllegalArgumentException - 如果 parties 小于 1
//注意点:
// 如果 传入的参数为6,而所有线程一共才五条,那么主线程和 子线程,将永远处于等待状态,因为没有第六条线程执行 await方法...
// 或者, 线程在执行await()之前,出现异常, 屏障永远不会被满足
CyclicBarrier cyclicBarrier = new CyclicBarrier(6);
for (int i =0;i<5;i++) { //开启五条线程...
new Thread(new Runnable() {
@Override
public void run() {
c.meeting(cyclicBarrier);
}
}).start();
}
// 主线程
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("人都到齐了,开会...");
// cyclicBarrier.reset();
}
}
带 Runable, 当所有预期的线程都await后,先执行Runable里面的任务
public class CyclicBarrier02 {
public void meeting(CyclicBarrier cyclicBarrier){
System.out.println(Thread.currentThread().getName()+"到达会议室...");
try {
cyclicBarrier.await(); //此线程等待..
System.out.println(Thread.currentThread().getName()+"听领导讲话...");
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
CyclicBarrier02 c = new CyclicBarrier02();
// 构造函数2:
/*
创建一个新的 CyclicBarrier，它将在给定数量的参与者（线程）处于等待状态时启动，
并在启动 barrier 时执行给定的屏障操作，该操作由最后一个进入 barrier 的线程执行。
参数：
parties - 在启动 barrier 前必须调用 await() 的线程数
barrierAction - 在启动 barrier 时执行的命令；如果不执行任何操作，则该参数为 null
抛出：
*/
CyclicBarrier cyclicBarrier = new CyclicBarrier(6, new Runnable() {
@Override
public void run() {
System.out.println("开始开会...");
}
});
for (int i =0;i<5;i++) { //开启五条线程...
new Thread(new Runnable() {
@Override
public void run() {
c.meeting(cyclicBarrier);
}
}).start();
}
// 主线程
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("主线程也 await了 ... 人都到齐了,开会...");
cyclicBarrier.reset();
}
}
在 多线程二 基本技能中有详细的将讲解,使用
局限性:
标签：node fileread 单线程 oid 完成 目标 cat 释放 并发
原文地址：https://www.cnblogs.com/ZhuChangwu/p/11150332.html
