Android事件总线 浅谈 EventBus
Posted 一叶知秋V
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Android事件总线 浅谈 EventBus相关的知识，希望对你有一定的参考价值。
JUC 包（java.util.concurrent）提供了并发编程的解决方案，CAS 是 java.util.concurrent.atomic 包的基础，AQS 是 java.util.concurrent.locks 包以及一些常用类比如 Semophore，ReentrantLock 等类的基础。
JUC 包的分类：
- executor：线程执行器
- locks：锁
- atomic：原子变量类
- tools：并发工具类
- collections：并发集合
1.并发工具类
包括倒计时闭锁 CountDownLatch、栅栏 CyclicBarrier、信号量 Semaphore、交换器 Exchanger。
1.CountDownLatch
倒计时闭锁 CountDownLatch 也称为倒计时计数器，可以让主线程等待一组事件发生后继续执行，事件是指 CountDownLatch 里的 countDown() 方法。
public class CountDownLatchDemo {
public static void main(String[] args) throws InterruptedException {
new CountDownLatchDemo().execute();
}
private void execute() throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(3); // 初始化count为3
new Thread(new Task(countDownLatch), "thread1").start();
new Thread(new Task(countDownLatch), "thread2").start();
new Thread(new Task(countDownLatch), "thread3").start();
// 调用await()方法的线程会被挂起, 它会等待直到count值为0才继续执行
// 带超时参数的await(long timeout, TimeUnit unit)方法超时后不管count是否为0,都会继续执行
countDownLatch.await();
System.out.println("所有线程已到达, 主线程开始执行");
}
class Task implements Runnable {
private CountDownLatch countDownLatch;
private Task(CountDownLatch countDownLatch) {
this.countDownLatch = countDownLatch;
}
@Override
public void run() {
System.out.println("线程" + Thread.currentThread().getName() + "已经到达");
countDownLatch.countDown(); // 将count值减1
}
}
}
执行结果：
线程thread3已经到达
线程thread1已经到达
线程thread2已经到达
所有线程已到达, 主线程开始执行
2.CyclicBarrier
栅栏 CyclicBarrier 可以阻塞当前线程，等待其他线程，所有线程必须同时到达栅栏位置后，才能继续执行；所有线程到达栅栏处，可以触发执行另一个预先设置的线程。
public class CyclicBarrierDemo {
public static void main(String[] args) {
new CyclicBarrierDemo().execute();
}
private void execute() {
CyclicBarrier cyclicBarrier = new CyclicBarrier(3); // 初始化栅栏的参与者为3
new Thread(new Task(cyclicBarrier), "thread1").start();
new Thread(new Task(cyclicBarrier), "thread2").start();
new Thread(new Task(cyclicBarrier), "thread3").start();
}
class Task implements Runnable {
private CyclicBarrier cyclicBarrier;
private Task(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run() {
System.out.println("线程" + Thread.currentThread().getName() + "已经到达");
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println("线程" + Thread.currentThread().getName() + "开始处理");
}
}
}
执行结果：
线程thread3已经到达
线程thread1已经到达
线程thread2已经到达
线程thread2开始处理
线程thread3开始处理
线程thread1开始处理
3.Semaphore
信号量 Semaphore 可以控制某个资源可被同时访问的线程个数。
public class SemaphoreDemo {
public static void main(String[] args) {
// 线程池
ExecutorService exec = Executors.newCachedThreadPool();
// 只能5个线程同时访问
final Semaphore semp = new Semaphore(5);
for (int index = 0; index < 20; index++) {
final int no = index;
Runnable runnable = new Runnable() {
@Override
public void run() {
try {
// 获取许可
semp.acquire();
System.out.println("Accessing: " + no);
Thread.sleep((long) (Math.random() * 10000));
// 访问完后, 释放
semp.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
exec.execute(runnable);
}
// 退出线程池
exec.shutdown();
}
}
只有 5 个客户端可以访问，其他客户端只能等待已经获取许可的 5 个客户端调用 release() 方法释放许可，才能进行访问。
4.Exchanger
交换器 Exchanger 主要用于线程之间数据交换，只能用户两个线程，可以实现两个线程到达同步点后，相互交换数据。先到达同步点的线程会被阻塞，当两个线程都到达同步点后，开始交换数据。
public class ExchangerDemo {
private static Exchanger<String> exchanger = new Exchanger<String>();
public static void main(String[] args) {
ExecutorService exec = Executors.newFixedThreadPool(2);
// 线程一
exec.execute(() -> {
try {
String text = exchanger.exchange("hello1");
System.out.println("thread1: " + text);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
// 线程二
exec.execute(() -> {
try {
TimeUnit.SECONDS.sleep(3);
String text = exchanger.exchange("hello2");
System.out.println("thread2: " + text);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
执行结果：
thread1: hello2
thread2: hello1
#2.并发集合
包括 ConcurrentHashMap、BlockingQueue。
##1.BlockingQueue
BlockingQueue 接口提供了可阻塞的入队和出队操作，如果队列满了，入队操作将阻塞，直到有空间可用；如果队列空了，出队操作将阻塞，直到有元素可用。
BlockingQueue 接口主要用于生产者 - 消费者模式，在多线程场景时生产者线程在队列尾部添加元素，而消费者则在队列头部消费元素，通过这种方式能够达到将任务的生产和消费进行隔离的目的。
以下是 BlockingQueue 常用的实现类，都是线程安全的：
| 说明 | |
|---|---|
| ArrayBlockingQueue | 一个由数组组成的有界阻塞队列 |
| LinkedBlockingQueue | 一个由链表组成的有界/无界阻塞队列 |
| PriorityBlockingQueue | 一个支持优先级排序的无界阻塞队列 |
| DelayQueue | 一个使用优先级队列实现的无界阻塞队列 |
| SynchronousQueue | 一个不存储元素的阻塞队列 |
| LinkedTransferQueue | 一个由链表组成的无界阻塞队列 |
| LinkedBlockingDeque | 一个由链表组成的双向阻塞队列 |
3.锁
Lock（Java5新特性）提供了比synchronized方法和synchronized代码块更广泛的锁定操作，Lock允许实现更灵活的结构，可以具有差别很大的属性，并且支持多个相关的Condition对象。
主要针对一个 JVM 中的多个线程对共享资源的操作。
Java 中常见的锁有独享锁/共享锁+公平锁/非公平锁+乐观锁/悲观锁
锁 Lock 分为公平锁和非公平锁，公平锁表示线程获取锁的顺序是按照线程加锁的顺序来分配的，先进先出，而非公平锁就是一种获取锁的抢占机制。
1.ReentrantLock
ReentrantLock 是一个可重入锁。
- 位于 java.util.concurrent.locks 包；
- 和 CountDownLatch、FutureTask、Semaphore 一样基于 AQS 实现，AQS 是 Java 并发用来构建锁或其他同步组件的基础框架，JUC（ java.util.concurrent）package 的核心；
- ReentrantLock 能够实现比 synchronized 更细粒度的控制，如设置公平性 fairness；
- ReentrantLock 调用 lock() 之后，必须调用 unlock() 释放锁；
- 性能未必比 synchronized 高，并且也是可重入的。
ReentrantLock 公平性的设置：
// 参数为true时, 倾向于将锁赋予等待时间最久的线程
// 获取锁的顺序按先后调用lock方法的顺序(慎用)
ReentrantLock fairLock = new ReentrantLock(true); // 公平锁
// 获取锁的顺序按抢占的顺序, 看运气
ReentrantLock fairLock = new ReentrantLock(); // 默认非公平锁
ReentrantLock fairLock = new ReentrantLock(false); // 非公平锁
公平性是减少线程饥饿的情况发生的一个办法，线程饥饿指个别线程长期等待锁，但却始终无法获取锁的情况。
synchronized 和 ReentrantLock 的区别：
synchronized 是非公平锁。
其实大多数场景中，公平性未必有想象中的那么重要，Java 默认的调度策略很少会导致饥饿的发生，若要保证公平性，则要引入额外的开销，导致一定的吞吐量下降，所以建议只有当程序确实有公平性需要的时候，才有必要去指定公平锁。
ReentrantLock 相比 synchronized，将锁对象化，提供各种便利的方法，进行精细的同步操作，甚至可以表达 synchronized 难以表达的用例：
- 判断是否有线程，或者某个特定线程，在排队等待获取锁；
- 带超时的获取锁的尝试；
- 感知有没有成功获取到锁。
Condition
如果说 ReentrantLock 将 synchronized 转变成了可控的对象，那么是否也能将 wait/notify/notifyAll 对象化？答案是可以的，java.util.concurrent.locks.Condition 类做到了这一点，
synchronized 和 ReentrantLock 的区别总结：
- synchronized 是关键字，ReentrantLock 是类；
ReentrantLock 比 synchronized 的扩展性体现在：
- ReentrantLock 可以对获取锁的等待时间进行设置，避免死锁；
- ReentrantLock 可以获取各种锁的信息；
- ReentrantLock 可以灵活地实现多路通知；
最关键的是两者的锁机制也是不一样的，synchronized 底层操作的是 Java 对象头中的 Mark Word，ReentrantLock 底层调用 Unsafe 类的 park() 方法加锁。Unsafe 类可以用来在任意内存地址位置处读写数据，另外，Unsafe 还支持一些 CAS 的操作。
1.ReentrantLock
ReentrantLock 是一种排他锁，同一时间只有一个线程在执行 ReentrantLock.lock() 方法后面的任务。
private ReentrantLock lock = new ReentrantLock(); //默认非公平锁
// private ReentrantLock lock = new ReentrantLock(true); //公平锁, true:公平锁, false:非公平锁
public void testMethod() {
try {
lock.lock(); //获取锁
...
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock(); //释放锁
}
}
ReentrantLock 实现等待/通知，可以借助于 Condition 对象实现：
public class MyService {
private ReentrantLock lock = new ReentrantLock();
private Condition condition = lock.newCondition();
public void await() {
try {
System.out.println("await start");
lock.lock(); //获取锁
condition.await(); //等同于Object类中的wait()方法
System.out.println("await end");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock(); //释放锁
}
}
public void signal() {
try {
System.out.println("signal start");
lock.lock(); //获取锁
condition.signal(); //等同于Object类中的notify()方法, notifyAll同理
System.out.println("signal end");
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock(); //释放锁
}
}
}
public class MyThread1 extends Thread {
private MyService service;
public MyThread1(MyService service){
super();
this.service = service;
}
@Override
public void run() {
System.out.println("MyThread1 start");
service.await();
System.out.println("MyThread1 end");
}
}
public static void main(String[] args) {
MyService service = new MyService();
MyThread1 t1 = new MyThread1(service);
t1.start();
Thread.sleep(3000);
service.signal();
}
运行结果：
MyThread1 start
await start
//这里停顿了3秒
signal start
signal end
await end
MyThread1 end
如果需要唤醒的是部分线程，可以通过定义多个 Condition 实现。
ReentrantLock 一些常用的方法如下：
| 方法 | 说明 |
|---|---|
| int getHoldCount() | 查询当前线程保持此锁定的个数，也就是调用lock()方法的次数。 |
| int getQueueLength() | 返回正等待获取此锁定的线程估计数。 |
| int getWaitQueueLength(Condition condition) | 返回等待与此锁定相关的给定条件Condition的线程估计数。 |
| boolean hasQueuedThread(Thread thread) | 查询指定的线程是否正在等待获取此锁定。 |
| boolean hasQueuedThreads() | 查询是否有线程正在等待获取此锁定。 |
| boolean hasWaiters(Condition condition) | 查询是否有线程正在等待与此锁定有关的condition条件。 |
| boolean isFair() | 判断是否是公平锁。 |
| boolean isHeldByCurrentThread() | 查询当前线程是否保持此锁定。 |
| boolean isLocked() | 查询此锁定是否由任意线程保持。 |
| void lockInterruptibly() | 如果当前线程未被中断，则获取锁定，如果已经被中断则出现异常。 |
| boolean tryLock() | 仅在调用时锁定未被另一个线程保持的情况下，才获取该锁定。 |
| boolean tryLock(long timeout, TimeUnit unit) | 如果锁定在给定等待时间内没有被另一个线程保持，且当前线程未被中断，则获取该锁定。 |
3.读写锁
1.ReentrantReadWriteLock
排他锁 ReentrantLock 虽然保证了实例变量的线程安全性，但效率却是十分低下的。所以在 JDK 中提供了一种读写锁 ReentrantReadWriteLock 类，使用它可以加快运行效率，在某些不需要操作实例变量的方法中，完全可以使用读写锁 ReentrantReadWriteLock 来提升该方法的代码运行速度。
读写锁包含两个锁，一个是读操作相关的锁，也称为共享锁；另一个是写操作相关的锁，也就做排他锁。也就是多个读锁之间不互斥，读锁与写锁互斥，写锁与写锁互斥。多个 Thread 可以同时进行读取操作，但是同一时刻只允许一个 Thread 进行写入操作。
1.读读共享
public class MyService {
private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public void read() {
try {
lock.readLock().lock(); //获取锁
System.out.println("获取读锁" + Thread.currentThread().getName()
+ " " + System.currentTimeMillis());
Thread.sleep(10000);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.readLock().unlock(); //释放锁
}
}
}
public class MyThread1 extends Thread {
private MyService service;
public MyThread1(MyService service){
super();
this.service = service;
}
@Override
public void run() {
service.read();
}
}
public class MyThread2 extends Thread {
private MyService service;
public MyThread2(MyService service){
super();
this.service = service;
}
@Override
public void run() {
service.read();
}
}
public static void main(String[] args) {
MyService service = new MyService();
MyThread1 t1 = new MyThread1(service);
t1.setName("t1");
MyThread2 t2 = new MyThread2(service);
t2.setName("t2");
t1.start();
t2.start();
}
程序运行后的结果如下：
获取读锁t1 1550141547658
获取读锁t2 1550141547659
两个线程几乎同时进入 lock() 后面的代码，说明在此使用了 lock.readLock() 读锁可以提高运行效率，允许多个线程同时执行 lock 方法后面的代码。
2.写写互斥
更改类 MyService.java 代码如下：
public class MyService {
private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public void write() {
try {
lock.writeLock().lock(); //获取锁
System.out.println("获取写锁" + Thread.currentThread().getName()
+ " " + System.currentTimeMillis());
Thread.sleep(10000);
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.writeLock().unlock(); //释放锁
}
}
}
MyThread1、MyThread2 的 run() 方法改为调用 write() 方法。
程序运行后的结果如下：
获取写锁t1 1550144386681
获取写锁t2 1550144396682
说明写写操作是互斥的。
3.读写互斥
更改类 MyService.java 代码如下：
public 以上是关于Android事件总线 浅谈 EventBus的主要内容，如果未能解决你的问题，请参考以下文章