kafka.network.AbstractServerThread中的线程协作机制

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

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()

}

1

2

3

4

5

6

7

8

9

10

11

private
class Processor extends
AbstractServerThread {

  override def run() {

    while(isRunning) {

      println("processor is running")

      //执行一些操作

      Thread.sleep(1000)

    }

    shutdownComplete()

  }

}