这个虚类是kafka.network.Acceptor和kafka.network.Processor的父类,提供了一个抽象的Sever线程。

它的有趣之处在于为子类的启动和停止提供了线程间的协作机制。

当子类的shutdown方法被调用时,子类可以得知自己被停止,在子类做了适当的处理和清理后,调用自己的shutdownComplete方法,使得对子类shutdown方法的调用从阻塞状态返回,从而使调用线程得知子类的对象已经恰当的停止。

即,在另一个线程中要关闭一个AbstractServerThread,可以执行它shutdown方法,当此方法从阻塞中返回,代表它已经恰当的关闭。

同样,对子类的awaitStartup方法调用也会阻塞,直到子类确认自己完全启动,这个方法调用才会返回。

这些功能是通过对CountdownLatch和AtomicBoolean的使用来实现的。

private[kafka] abstract class AbstractServerThread extends Runnable with Logging {

  protected val selector = Selector.open();
private val startupLatch = new CountDownLatch(1)
private val shutdownLatch = new CountDownLatch(1)
private val alive = new AtomicBoolean(false) /**
* Initiates a graceful shutdown by signaling to stop and waiting for the shutdown to complete
*/
def shutdown(): Unit = {
alive.set(false)
selector.wakeup()
shutdownLatch.await
} /**
* Wait for the thread to completely start up
*/
def awaitStartup(): Unit = startupLatch.await /**
* Record that the thread startup is complete
*/
protected def startupComplete() = {
alive.set(true)
startupLatch.countDown
} /**
* Record that the thread shutdown is complete
*/
protected def shutdownComplete() = shutdownLatch.countDown /**
* Is the server still running?
*/
protected def isRunning = alive.get /**
* Wakeup the thread for selection.
*/
def wakeup() = selector.wakeup() }

由于它代表了一个Server线程,在其内部使用了java.nio的Selector。所以在shutdown时,需要调用Selector的wakeup方法,使得对Selector的select方法的调用从阻塞中返回。 

继承它的子类必须对isRunning进行判断,来确定自己是否已经被要求关闭。以及在处理关闭请求后,调用shutdownComplete()来确认已完闭完成。

由于Acceptor和Processor的实现太长,这里写了一个例子模拟它们

private class Processor extends AbstractServerThread {
override def run() {
while(isRunning) {
println("processor is running")
//执行一些操作
Thread.sleep(1000)
}
shutdownComplete()
} }

  在工作循环中判断isRunning作为退出循环的条件。然后执行shutdownComplete, 这时对Processor 的shutdown方法的调用才会返回。

kafka.network.AbstractServerThread中的线程协作机制的更多相关文章

  1. Java中的线程协作之Condition

    一.Condition接口 1.Condition接口的常用方法介绍 /** * 已经获取到锁的线程调用该方法会进入等待状态,知道其他持有锁的线程通知(signal)等待队列中的线程或者被中断退出等待 ...

  2. Java中的线程同步机制

    一.首先为什么线程需要同步? 1.多线程安全问题的原因      A:有多线程环境      B:有共享数据      C:有多条语句操作共享数据 2. //未完待续后面会继续更新

  3. CUDA线程协作之共享存储器“__shared__”&&“__syncthreads()”

    在GPU并行编程中,一般情况下,各个处理器都需要了解其他处理器的执行状态,在各个并行副本之间进行通信和协作,这涉及到不同线程间的通信机制和并行执行线程的同步机制. 共享内存"__share_ ...

  4. Java多线程编程(4)--线程同步机制

    一.锁 1.锁的概念   线程安全问题的产生是因为多个线程并发访问共享数据造成的,如果能将多个线程对共享数据的并发访问改为串行访问,即一个共享数据同一时刻只能被一个线程访问,就可以避免线程安全问题.锁 ...

  5. Java编程的逻辑 (67) - 线程的基本协作机制 (上)

    ​本系列文章经补充和完善,已修订整理成书<Java编程的逻辑>,由机械工业出版社华章分社出版,于2018年1月上市热销,读者好评如潮!各大网店和书店有售,欢迎购买,京东自营链接:http: ...

  6. Java编程的逻辑 (68) - 线程的基本协作机制 (下)

    ​本系列文章经补充和完善,已修订整理成书<Java编程的逻辑>,由机械工业出版社华章分社出版,于2018年1月上市热销,读者好评如潮!各大网店和书店有售,欢迎购买,京东自营链接:http: ...

  7. ThreadLocal,Java中特殊的线程绑定机制

    在DRP项目中,我们使用了ThreadLocal来创建Connection连接,避免了一直以参数的形式将Connection向下传递(传递connection的目的是由于jdbc事务要求确保使用同一个 ...

  8. 浅谈利用同步机制解决Java中的线程安全问题

    我们知道大多数程序都不会是单线程程序,单线程程序的功能非常有限,我们假设一下所有的程序都是单线程程序,那么会带来怎样的结果呢?假如淘宝是单线程程序,一直都只能一个一个用户去访问,你要在网上买东西还得等 ...

  9. Java-ThreadLocal,Java中特殊的线程绑定机制

    在DRP项目中,我们使用了ThreadLocal来创建Connection连接,避免了一直以参数的形式将Connection向下传递(传递connection的目的是由于jdbc事务要求确保使用同一个 ...

随机推荐

  1. Router的创建者——RouteBuilder

    Router的创建者--RouteBuilder 在<注册URL模式与HttpHandler的映射关系>演示的实例中,我们总是利用一个RouteBuilder对象来为RouterMiddl ...

  2. fifo read

    #include <stdio.h> #include <stdlib.h> #include <string.h> #include <sys/types. ...

  3. 【USACO】calfflac

    关键:以回文中心位置为变量进行遍历 //必须把纯字母先提出来 否则肯能会出现错误 比如: lvlv= 在检查长度4时 lvlv认为不是回文 vlv=认为是回文 但实际上 lvl 出现的要更早一些 // ...

  4. 史上最全的 UIWebview 的 JS 与 OC 交互

    来源:伯乐在线 - 键盘风筝 链接:http://ios.jobbole.com/89330/ 点击 → 申请加入伯乐在线专栏作者 其实一直想给大家整理一下JS与OC的交互,但是没有合适的机会,今天借 ...

  5. C#程序中获取电脑硬件配置信息的一种方法

    本文介绍获取cpu信息和内存信息的方法,根据本文所举例的代码可以举一反三获取更多信息. 获取cpu名称的方法: public string GetCpuInfo() { ManagementObjec ...

  6. C语言实验单片机串口发送int型数据

    void SendIint(int n)reentrant { unsigned char s; while(n!=0) { s=(unsigned char)n%10+48; SendByte(s) ...

  7. webpack + vue 在dev和production模式下的小小区别

    上周的某一天,和一位同样是前端技术极度爱好的开发者朋友聊天,他在提出了一个问题,他写的vue程序为什么在dev模式运行良好,而在production模式就直接报错了.这让我感到惊讶,还有这么神奇的事情 ...

  8. 3.2.2 SpringMVC配置式开发

    SpringMVC配置式开发 1. SpringMVC运行原理(执行过程) 2. 需求 用户提交一个请求, 服务端处理器接收到请求后, 给出一条信息,在相应页面中显示该条信息 3. 开发步骤 (1) ...

  9. python杂记二

    1. 写文件可以直接使用print函数 file_name = open("file_name.txt","w") print("file conta ...

  10. Ajax传递List对象到前台展示问题遇到的坑

    后台Json转换 后台传递的List对象,如果对象是实体类,实体类和另一个表关联,就可能会出现以下错误 org.hibernate.LazyInitializationException: faile ...