handler Handler机制与生产者消费者模式

昨日，恒大集团以67.46亿港元收购香港时颖公司100%的股份，间接收购了Smart King公司45%的股份，成为公司第一大股东。这意味着恒大正式进入美国新能源汽车公司法拉第未来。恒大集团正式进入FF后，恒大将任命副董事长、董事局主席夏海钧为智王公司董事长。

作者简介

zcbiner的这篇贡献结合生产者-消费者模型分析了Handler机制，希望对大家有所帮助！

Zcbiner的博客地址:

https://www.jianshu.com/u/fcc2f0577270

处理器机制

Android中通常使用处理程序机制来更新UI。一个子线程执行一个任务，任务执行后发送一条消息:Handler.sendMessage()，然后在UI线程Handler.handleMessage()上调用，执行相应的处理。

处理器机制有几个非常重要的类:

Handler：用来发送消息：sendMessage等多个方法，并实现handleMessage()方法处理回调（还可以使用Message或Handler的Callback进行回调处理，具体可以看看源码）。Message：消息实体，发送的消息即为Message类型。MessageQueue：消息队列，用于存储消息。发送消息时，消息入队列，然后Looper会从这个MessageQueen取出消息进行处理。Looper：与线程绑定，不仅仅局限于主线程，绑定的线程用来处理消息。loop()方法是一个死循环，一直从MessageQueen里取出消息进行处理。

这些类的功能也可以通过下图来解释:

Looper.loop()是一个无限循环。为什么不卡主线？简单来说，当MessageQueue为空时，线程会挂起，一旦有消息，就会唤醒线程处理消息。看这个问题的答案:

为什么Android中的Looper.loop()主线程不会卡在无尽的循环里？

https://www.zhihu.com/question/34652589

处理程序不仅可以处理主线程中的消息，还可以处理子线程中的消息，前提是子线程应该与Looper相关联。标准的写法是:

classLooperThreadextendsThread {

publicHandler mHandler

publicvoidrun(){

looper . prepare()；

mhhandler = new handler(){

publicvoidhandleMessage(消息消息){

//在此处理传入消息

}

};

looper . loop()；

}

}

确保调用Looper.prepare()和Looper.loop()方法。Looper.prepare()使用ThreadLocal将当前线程与来自new的Looper相关联，Looper.loop()打开循环来处理消息。这样就可以在LooperThread中处理mHandler发送的消息。

生产者-消费者模型

在并发编程中使用生产者和消费者模式可以解决大多数并发问题。该模型通过平衡生产线程和消费线程的工作能力，提高了程序的整体数据处理速度。

在网络世界里，生产者是生产数据的线程，消费者是消费数据的线程。在多线程开发中，如果生产者的处理速度快，消费者的处理速度慢，生产者必须等待消费者完成处理后才能继续生产数据。同理，如果消费者的处理能力大于生产者，那么消费者必须等待生产者。为了解决这种生产和消费能力不平衡的问题，有一个生产者和消费者模型。请参考:

浅谈并发(10)生产者-消费者模型

http://ifeve.com/producers-and-consumers-mode/

处理者、生产者和消费者

那么Handler机制和生产者-消费者模型是什么关系呢？

处理机制是生产者-消费者模型。可以理解为，当Handler发送消息时，它是生产者，一条一条地产生消息。Looper可以理解为一个消费者，在其loop()方法中，一个无限循环从MessageQueue中取出消息进行处理。消息队列是缓冲区，处理程序生成的消息被放入消息队列，循环程序从消息队列中取出消息。

由于处理器机制本质上是生产者-消费者模型，我们可以在没有Android的情况下实现处理器机制。

处理者

用来发送和处理消息，所以主要方法是sendMessage()和handleMessage():

publicatabstractclasshandler {

private imessagequeue messageQueue；

publicHandler(Looper looper){

message queue = looper . message queue；

}

publicHandler(){

looper . my looper()；

}

公共无效发送消息(消息消息){

//指定发送消息的处理程序，方便回调

message.target = this

尝试{

messageQueue.enqueueMessage(消息)；

} catch(中断异常e) {

e . printstacktrace()；

}

}

publicatabstractvotyhandlemessage(消息消息)；

}

消息

很简单，作为使者，绑定到相应的处理程序很重要:目标变量。

publicclassMessage{

privateintcode

privateString消息；

处理程序目标；

publicMessage(){ }

publicMessage(intcode，String msg){

this.code = code

this.msg = msg

}

publiintgetcode(){

returncode

}

publicvoidsetCode(intcode){

this.code = code

}

publicString getMsg(){

returnmsg

}

publicvoidsetMsg(字符串msg){

this.msg = msg

}

}

IMessageQueue

因为消息皇后可以有不同的实现，所以接口是抽象的。

publicinterfaceIMessageQueue {

消息next()throwsInterruptedException；

无效排队消息(消息消息)抛出异常；

}

尺蠖

loop()方法是一个无限循环，在这里处理消息。当没有消息时，绑定线程将处于等待状态。使用ThreadLocal绑定到线程。

publicclassLooper{

staticfinalThreadLocal & lt。Looper>。sThreadLocal = new threadlocal & lt；Looper>。();

IMessageQueue messageQueue

privatedstaticlooper sMainLooper；

publicLooper(){

message queue = NewMessagequeue(2)；

//messageQueue = new messageQueue 1(2)；

//messageQueue = new messageQueue 2(2)；

}

publicationstatidvotyprepare(){

if(sThreadLocal.get()！= null) {

thrownewRuntimeException("每个线程只能创建一个循环程序")；

}

sthreadLocal . set(NewLooper())；

}

publicationstaticvotyprepareinlooper(){

prepare()；

同步(Looper.class) {

if(sMainLooper！= null) {

thrownewIllegalStateException(“主弯针已经准备好了。”);

}

sMainLooper = my looper()；

}

}

publicationstaticlooper getMainLooper(){

returnsMainLooper

}

publicationstaticlooper my looper(){

returnsthreadlocal . get()；

}

publicationstativorloop(){

finalLooper me = my Looper()；

if(me == null) {

thrownewRuntimeException("无活套；Looper.prepare()未在此线程上调用。);

}

for(；；) {

//消费消息，如果MessageQueen为空，等待

Message message = null

尝试{

message = me . message queue . next()；

} catch(中断异常e) {

e . printstacktrace()；

}

if(消息！= null) {

message.target.handleMessage(消息)；

}

}

}

}

消息Queen由LinkedBlockingQueue实现

消息队列是一个缓冲区，需要满足以下功能:

当缓冲区满时，挂起执行enqueueMessage的线程。当缓冲区空时，挂起执行next的线程。当缓冲区非空时，唤醒被挂起在next的线程。当缓冲区不满时，唤醒被挂起在enqueueMessage的线程。

因此，消息队列最简单的实现是使用链接锁定队列:

用并发花阻塞队列

https://www.jianshu.com/p/5dfc81b10e30

publicclassmessagequeueimplesimessagequeue {

privatefinalBlockingQueue & lt消息>排队；

publicMessageQueue(intcap){

this . queue = NewLinkedBlockingqueue & lt；>。(cap)；

}

public message next()throwsInterruptedException {

return queue . take()；

}

publicvoidenqueueMessage(消息消息){

尝试{

queue.put(消息)；

} catch(中断异常e) {

e . printstacktrace()；

}

}

}

这样，就实现了一个简单的处理器机制。Android系统的Handler机制肯定没有那么简单，做了很多健壮性相关的处理，以及与底层的交互等等。另外，Android系统的MessageQueen是一个结合Message实现的无界队列，这意味着发送消息的队列不会被阻塞。

处理器机制已经建立，让我们测试一下:

publicclassMain{

public static void main(String[]args){

main thread main thread = new main thread()；

main thread . start()；

//确保主回路已经建立

while(Looper . GetMainLooper()= = null){

尝试{

thread . sleep(10)；

} catch(中断异常e) {

e . printstacktrace()；

}

}

handler handler = new handler(Looper . GetMainLooper()){

@覆盖

publicvoidhandleMessage(消息消息){

system . out . println(" execute in:"+Thread . current thread()。getName())；

switch(msg.getCode()) {

案例0:

system . out . println(" code 0:"+msg . GetMSg())；

打破；

案例1:

system . out . println(" code 1:"+msg . GetMSg())；

打破；

默认:

System.out.println("其他代码:"+msg . GetMSg())；

}

}

};

Message message1 = newMessage( 0，“我是第一条消息！”);

工作线程工作线程1 =新工作线程(处理程序，消息1)；

Message message2 = newMessage( 1，“我是第二条消息！”);

工作线程工作线程2 =新工作线程(处理程序，消息2)；

Message message3 = newMessage( 34，“我是消息！”);

工作线程工作线程3 =新工作线程(处理程序，消息3)；

workthread 1 . start()；

workthread 2 . start()；

workthread 3 . start()；

}

/* *模拟工作线程* */

publicationstaticclassworkthreadextendsthread {

privateHandler处理程序；

privateMessage消息；

公共工作线程(处理程序，消息消息){

setName(" WorkThread ")；

this.handler = handler

this.message = message

}

@覆盖

publicvoidrun(){

super . run()；

//模拟耗时的操作

random random = new random()；

尝试{

thread . sleep(random . nextint(10)* 300)；

} catch(中断异常e) {

e . printstacktrace()；

}

//任务执行后，发送消息

handler.sendMessage(消息)；

}

}

/* *模拟主线*/

publicationstaticlasmainthreadextendsthread {

publicMainThread(){

setName(" main thread ")；

}

@覆盖

publicvoidrun(){

super . run()；

//这和系统调用一样吗，哈哈

looper . prepareinlooper()；

System.out.println(getName() +“弯针准备好了”)；

looper . loop()；

}

}

}

这里，我们模拟一个场景，其中一个子线程执行一个任务，并向主线程发送一条消息进行处理。那么执行结果就是:

主线弯针准备好了

执行于:主线程

其他代码:我是信息！

执行于:主线程

代码0:我是第一条消息！

执行于:主线程

代码1:我是第二条消息！

可以看到handleMessage的执行在主线程，简单的Handler机制就完成了！！！

本文到此结束，但是MessageQueen有许多实现，所以让我们看看它们的不同实现。

等待/通知的实现

publicclass messagequeue1 implementimessagequeue {

privateQueue<。消息>排队；

privatefinitionatomicinteger integer = Newatomicinteger(0)；

privatevolatileintcount

privatefileobject BUFFER _ LOCK = NewObject()；

publicMessageQueue1(intcap){

this.count = cap

queue = newLinkedList & lt>。();

}

@覆盖

public message next()throwsInterruptedException {

同步(BUFFER_LOCK) {

while(queue.size() == 0) {

BUFFER _ LOCK . wait()；

}

message = queue . poll()；

BUFFER _ LOCK . NotifyAll()；

returnmessage

}

}

@覆盖

publicvoidenqueueMessage(消息消息)throwsInterruptedException {

同步(BUFFER_LOCK) {

while(queue.size() == count) {

BUFFER _ LOCK . wait()；

}

queue.offer(消息)；

BUFFER _ LOCK . NotifyAll()；

}

}

}

BUFFER_LOCK用于处理与锁相关的逻辑。

在next()方法中，如果队列的大小为0，说明现在没有需要处理的消息，那么执行BUFFER_LOCK.wait()挂起线程，当queue . poll()；当队列中有消息时，需要唤醒因为没有消息而挂起的线程，所以执行BUFFER _ LOCK . NotifyAll()；

在enqueueMessage()方法中，如果queue.size() == count表示消息已满，如果Handler继续发送消息，则队列无法继续加载，因此需要挂起线程:BUFFER _ LOCK . wait()；。执行queue.offer(消息)时；，队列中保存的消息少一条，可以插入新的消息，所以BUFFER _ lock . notifyall()；唤醒因队列已满而暂停的线程。

锁定实现

既然wait/notify可以实现MessageQueue，那么ReentrantLock当然可以实现。下面是一个实现可重入锁的例子，原则上是一致的。

publicclassmessagequeue2 implementimessagequeue {

privatefinalQueue<。消息>排队；

private int cap = 0；

privatefilelock lock = new recentrantlock()；

private FinanceCondition BUFFER _ CONTENT = lock . New Condition()；

publicMessageQueue2(intcap){

this.cap = cap

queue = newLinkedList & lt>。();

}

@覆盖

public message next()throwsInterruptedException {

尝试{

lock . lock()；

while(queue.size() == 0) {

BUFFER _ CONTINUE . await()；

}

message = queue . poll()；

BUFFER _ condition . signal all()；

returnmessage

}最后{

lock . unlock()；

}

}

@覆盖

publicvoidenqueueMessage(消息消息)throwsInterruptedException {

尝试{

lock . lock()；

while(queue.size() == cap) {

BUFFER _ CONTINUE . await()；

}

queue.offer(消息)；

BUFFER _ condition . signal all()；

}最后{

lock . unlock()；

}

}

}

锁实现的优化

在上面的实现过程中，排队和出列是基于同一个锁的，这意味着如果一个线程在排队，其他线程就不能出列；一个线程出列，但其他线程不能出列。一次只能有一个线程操作缓冲区，在多线程环境下明显影响性能。所以是优化的。

优化思路是入队和离队互不干扰。因此，进入队列和离开队列分别需要一个锁。进入队列在队列末尾执行，离开队列在头执行。代码如下:

publicclass messagequeue 3 implementimessagequeue {

privatefilelock putLock = new recentrantlock()；

private final condition not full = put lock . New condition()；

privatefilelock take lock = new recentrantlock()；

private final condition notEmpty = take lock . New condition()；

privateNode头；//组长

privateNode最后；//团队尾巴

privateAtomicInteger count = Newatomicinteger(0)；//记录大小

privateintcap = 10//容量，默认为10

publicMessageQueue3(intcap){

this.cap = cap

}

@覆盖

public message next()throwsInterruptedException {

节点节点；

intc =-1；

take lock . lock()；

尝试{

while(count.get() == 0) {

notempty . await()；

}

node = head

head = head.next

c = count.getAndDecrement

if(c >;0) {

notempty . signal()；

}

}最后{

take lock . unlock()；

}

if(count . get()& lt；cap) {

signal notfull()；

}

returnnode.data

}

@覆盖

publicvoidenqueueMessage(消息消息)throwsInterruptedException {

Node node = newNode(消息)；

intc =-1；

put lock . lock()；

尝试{

while(count.get() == cap) {

notfull . await()；

}

//初始状态

if(head = = null & amp；& amplast == null) {

head = last = node

} else{

last.next =节点；

last = last.next

}

c = count . GetAnDicrement()；

if(c & lt；cap) {

notfull . signal()；

}

}最后{

putlock . unlock()；

}

if(c >;0) {

signal notempty()；

}

}

privatevoidsignalNotEmpty(){

take lock . lock()；

尝试{

notempty . signal()；

}最后{

take lock . unlock()；

}

}

privatevoidsignalNotFull(){

put lock . lock()；

尝试{

notfull . signal()；

}最后{

putlock . unlock()；

}

}

staticclassNode{

消息数据；

节点next

公共节点(消息数据){

this.data = data

}

}

}

全文结束~