Java并发编程学习笔记（3）——线程同步进阶 | 程序员的自我修养

Semaphore

Semaphore是一个控制访问多个共享资源的计数器。当计数器值大于0，代表还有可用资源，线程可以继续访问和使用资源；当计数器的值等于0，代表暂无可用资源，线程必须等待资源的释放。

一个典型的例子就是，有多台打印机，当新的打印任务来时，将检测是否还有可用的打印机，若有则使用，并将可用打印机数量减1；若无则等待；当使用完毕后，释放打印机，将可用打印机数量加1。

使用示例

public class SemaphoreTest {
	private final Semaphore semaphore = new Semaphore(4);
	public void print() {
		try {
			semaphore.acquire();
			//do something
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			semaphore.release();
		}
	}
}

public class SemaphoreTest {

private final Semaphore semaphore = new Semaphore(4);

public void print() {

try {

semaphore.acquire();

//do something

} catch (InterruptedException e) {

e.printStackTrace();

} finally {

semaphore.release();

}

同Lock一样，Semaphore也可开启公平机制，Semaphore(int permits, boolean fair)。
acquireUninterruptibly()，与acquire()的区别是：线程中断不会抛出异常
tryAcquire()，尝试获取semaphore。如果成功，返回true。如果不成功，返回false值，并不会被阻塞和等待semaphore的释放。

CountDownLatch

CountDownLatch，用于等待多个并发事件完成，相当于一个倒数计数器，初始化时传入一个整数参数，调用方法countDown()减少计数，当计数为0时，解除await()的休眠。

使用示例

public class CountDownLatchTest implements Runnable{
	private final CountDownLatch controller = new CountDownLatch(4);
	public void assemble() {
		System.out.println("come one");
		controller.countDown();
	}
	@Override
	public void run() {
		try {
			System.out.println("wait " + controller.getCount());
			controller.await();
			System.out.println("get all");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}

public class CountDownLatchTest implements Runnable{

private final CountDownLatch controller = new CountDownLatch(4);

public void assemble() {

System.out.println("come one");

controller.countDown();

}

@Override

public void run() {

try {

System.out.println("wait " + controller.getCount());

controller.await();

System.out.println("get all");

} catch (InterruptedException e) {

e.printStackTrace();

}

CountDownLatch的计数器一旦初始化就无法更改。

CyclicBarrier

CyclicBarrier，用于在同一个点同步任务，与CountDownLatch类似，但CyclicBarrier可以在计数器等于0时执行初始化时传递的Runnable对象。

使用示例

public class CyclicBarrierTest implements Runnable {
	private final CyclicBarrier barrier = new CyclicBarrier(4, new After());
	@Override
	public void run() {
		try {
			System.out.println("wait");
			barrier.await();
			System.out.println("pass");
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (BrokenBarrierException e) {
			e.printStackTrace();
		}

	}
}
class After implements Runnable {
	@Override
	public void run() {
		System.out.println("tear down");
	}
}

public class CyclicBarrierTest implements Runnable {

private final CyclicBarrier barrier = new CyclicBarrier(4, new After());

@Override

public void run() {

try {

System.out.println("wait");

barrier.await();

System.out.println("pass");

} catch (InterruptedException e) {

e.printStackTrace();

} catch (BrokenBarrierException e) {

e.printStackTrace();

}

class After implements Runnable {

@Override

public void run() {

System.out.println("tear down");

}

CyclicBarrier的计数器可以重置，reset()。

Phaser

Phaser，用于阶段性并发任务的同步。CyclicBarrier可以在一个点同步任务，当需要在运行过程中多个点同步线程，则可以用Phaser来处理。

使用示例

public class PhaserTest implements Runnable {
	private Phaser phaser;
	public PhaserTest(Phaser phaser) {
		this.phaser = phaser;
	}
	@Override
	public void run() {
		try {
			Long duration = (long)(Math.random()*10000);
			Thread.sleep(duration);
			System.out.println("1");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		phaser.arriveAndAwaitAdvance();
		try {
			Long duration = (long)(Math.random()*10000);
			Thread.sleep(duration);
			System.out.println("2");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		phaser.arriveAndDeregister();
	}
}
class Main {
	public static void main(String[] args) {
		Phaser phaser = new Phaser(2);
		PhaserTest t1 = new PhaserTest(phaser);
		PhaserTest t2 = new PhaserTest(phaser);
		Thread n1 = new Thread(t1);
		Thread n2 = new Thread(t2);
		n1.start();
		n2.start();
		try {
			n1.join();
			n2.join();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("end");
	}
}

public class PhaserTest implements Runnable {

private Phaser phaser;

public PhaserTest(Phaser phaser) {

this.phaser = phaser;

}

@Override

public void run() {

try {

Long duration = (long)(Math.random()*10000);

Thread.sleep(duration);

System.out.println("1");

} catch (InterruptedException e) {

e.printStackTrace();

}

phaser.arriveAndAwaitAdvance();

try {

Long duration = (long)(Math.random()*10000);

Thread.sleep(duration);

System.out.println("2");

} catch (InterruptedException e) {

e.printStackTrace();

}

phaser.arriveAndDeregister();

}

class Main {

public static void main(String[] args) {

Phaser phaser = new Phaser(2);

PhaserTest t1 = new PhaserTest(phaser);

PhaserTest t2 = new PhaserTest(phaser);

Thread n1 = new Thread(t1);

Thread n2 = new Thread(t2);

n1.start();

n2.start();

try {

n1.join();

n2.join();

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("end");

}

若没有Phaser来阶段性的同步，输出的结果可能是“1 1 2 2”、“1 2 1 2”。但加入Phaser来阶段性的同步后，输出结果一定是“1 1 2 2”。

Phaser提供每次phaser改变阶段都会执行的方法，onAdvance()，可以Override该方法来实现不同的业务逻辑。

使用示例

Phaser phaser = new Phaser(2){
	@Override
	protected boolean onAdvance(int phase, int registeredParties) {
		System.out.println("p:"+phase);
		return super.onAdvance(phase, registeredParties);
	}
};

Phaser phaser = new Phaser(2){

@Override

protected boolean onAdvance(int phase, int registeredParties) {

System.out.println("p:"+phase);

return super.onAdvance(phase, registeredParties);

}

};

再次运行上述程序，结果则变为“1 1 p0 2 2 p1 end”。

Exchanger

Exchanger，用于并发任务间交换数据，但只适用于1对1的交换。

使用示例

public class ExchangerTest implements Runnable {
	private final Exchanger<String> exchanger;
	private String value;
	public ExchangerTest(Exchanger<String> exchanger, String value) {
		this.exchanger = exchanger;
		this.value = value;
	}
	@Override
	public void run() {
		System.out.println(Thread.currentThread().getName() + " Before: " + value);
		try {
			value = exchanger.exchange(value);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println(Thread.currentThread().getName() + " After: " + value);
	}
	public static void main(String[] args) {
		Exchanger<String> exchanger = new Exchanger<String>();
		ExchangerTest e1 = new ExchangerTest(exchanger, "11111");
		ExchangerTest e2 = new ExchangerTest(exchanger, "22222");
		Thread t1 = new Thread(e1, "t1");
		Thread t2 = new Thread(e2, "t2");
		t1.start();
		t2.start();
	}
}

public class ExchangerTest implements Runnable {

private final Exchanger<String> exchanger;

private String value;

public ExchangerTest(Exchanger<String> exchanger, String value) {

this.exchanger = exchanger;

this.value = value;

}

@Override

public void run() {

System.out.println(Thread.currentThread().getName() + " Before: " + value);

try {

value = exchanger.exchange(value);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println(Thread.currentThread().getName() + " After: " + value);

}

public static void main(String[] args) {

Exchanger<String> exchanger = new Exchanger<String>();

ExchangerTest e1 = new ExchangerTest(exchanger, "11111");

ExchangerTest e2 = new ExchangerTest(exchanger, "22222");

Thread t1 = new Thread(e1, "t1");

Thread t2 = new Thread(e2, "t2");

t1.start();

t2.start();

}

运行结果

t1 Before: 11111
t2 Before: 22222
t2 After: 11111
t1 After: 22222

（转载本站文章请注明作者和出处程序员的自我修养 – SelfUp.cn ，请勿用于任何商业用途）

张瑞昌: 有很多，比较常见的是Jacob迭代法，一次迭代O(n^3)，迭代次数不清楚。如果是手动算的话按照定义求就可以了
Anonymous:
lc277: 你好我想问下一般删除节点要多久，要删除的datanode大概用了 1t，解除授权已经30多小时还没完成，请问是出现什么问题了吗麻烦告诉下谢谢 qq1844554123
Anonymous: 你好我想问下一般删除节点要多久，要删除的datanode大概用了 1t，解除授权已经30多小时还没完成，请问是出现什么问题了吗
Anonymous:
李雪璇: 想要完整代码，可以帮忙发给我吗
Anonymous: 请问一下,那个 user的推荐结果楼主查看了么? 为什么输入数据最高是五分,输出结果都是7分8分啥的?怎么设置输出的分数的最大值?
Anonymous: 那个 user的推荐结果楼主查看了么? 为什么输入数据最高是五分,输出结果都是7分8分啥的?
Anonymous: stopGracefullyOnShutdown在yarn- client模式下我测试的无效，你的呢
Anonymous: 另外,import的lib包能否发个列表.
Anonymous: 部分程序已经无法使用, 另外你的import代码也应该放上来哈
wzq: 赞
增达网: 受教了！呵呵！
Anonymous:
27: dsa会报错，rsa不会
Anonymous: 看错了忽略我
Anonymous: UserSideCF这个类在哪里
晴子: 在主节点初始化CM5数据库的时候报错误：Verifying that we can write to /opt/cm-5.9.0/etc/cloudera-scm -server log4j:ERROR Could not...
zhangnew: 就4题
linxh: “ 但要是遇到预先并不知道数组的长度而又需要获取正确的（或者称之为原始的）split长度时，该如何处理呢。。？ ” 印象中可以split函数参数传-1？