生产者和消费者

RocketMQ中的生产者和消费者需要和Broker建立连接,进行消息通讯:

生产者需要一个线程负责与Broker通讯,发送消息。
消费者也需要一个线程负责与Broker通讯,接收消息。

线程池

RocketMQ使用线程池来管理生产者线程和消费者线程:

private ExecutorService producerExecutor = ...
private final ExecutorService consumerExecutor = ...

线程复用

当新建生产者或消费者时,会复用线程池中的线程:

public DefaultMQProducer createProducer() {

  Runnable producerRunnable = ...  

  producerExecutor.execute(producerRunnable);

  return new DefaultMQProducer();
}

public DefaultMQConsumer createConsumer() {  

  Runnable consumerRunnable = ...

  consumerExecutor.execute(consumerRunnable);     

  return new DefaultMQConsumer();   
}

直接将生产者/消费者任务提交给线程池,由可用线程运行。

享元模式

这种复用线程池线程的方式实现了享元模式:

线程池作为工厂,管理享元对象(线程资源)
线程作为可共享的享元对象
生产者/消费者任务作为享元的外部状态

通过这种方式:

可以重用有限的线程资源
降低线程创建和销毁的开销
提高性能

符合享元模式的定义:

定义 Flyweight 来实现共享的对象
将对象的外部状态移出 Flyweight

RocketMQ通过复用线程池线程,实现了生产者和消费者的享元对象。