RocketMQ学习笔记：生产者Producer-Toy模板网

这篇具有很好参考价值的文章主要介绍了RocketMQ学习笔记：生产者Producer。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

DefaultMQProducer

根据上文：RocketMQ学习笔记：消息Message - 掘金 (juejin.cn)，我们定位到Producer中的这一行代码：

java

复制代码

DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start();

通过new DefaultMQProducer("ProducerGroupName")实例化一个生产者对象。这里的ProducerGroupName在5.x的版本已经移除了，他主要说的是生产者分组，主要是保证Producer发送同一类消息且发送逻辑一致。.
通过setNamesrvAddr("127.0.0.1:9876")，是指明设置设置NameServer地址。（记得这句话吗：Producer、Consumer、Broker在启动时，会自行将数据注册到NameServer中）。
启动这个生产者实例。

DefaultMQProducer是继承ClientConfig的，ClientConfig主要是做RocketMQ客户端公共配置的，setNamesrvAddr就是其方法，后续我们再介绍这个ClientConfig类。

RocketMQ学习笔记：生产者Producer

DefaultMQProducer的其他推荐资料：RocketMQ源码解析：手把手教老婆看懂DefaultMQProducer_小虚竹的博客-CSDN博客，该文章介绍的很详细。

干活的是DefaultMQProducerImpl

构造它需要初始化一些基本属性，才能方便后面干活。

在构造方法中，我们继续往下定位发现，在DefaultMQProducer中其实有一个成员属性是defaultMQProducerImpl，我们记住：它是真正做消息传输的事：

java

复制代码

public class DefaultMQProducer extends ClientConfig implements MQProducer { protected final transient DefaultMQProducerImpl defaultMQProducerImpl; ...

在DefaultMQProducer的构造方法中，实例化了DefaultMQProducerImpl

RocketMQ学习笔记：生产者Producer

同时在上面第3步中，执行的

java

复制代码

producer.start();

也是在执行DefaultMQProducerImpl的start()。

RocketMQ学习笔记：生产者Producer

了解一下它

当在DefaultMQProducer调用以下代码时，会去创建DefaultMQProducerImpl，它是 RocketMQ 生产者的实现类。该类的主要作用是提供一个异步发送线程池，用于处理生产者发送消息的异步任务。

ini

复制代码

defaultMQProducerImpl = new DefaultMQProducerImpl(this, rpcHook);

调用以下的代码，主要是构造DefaultMQProducerImpl并且初始化一些属性，主要是线程池和队列。

asyncSenderThreadPoolQueue:容量为 50000。该队列用于存储异步发送任务，当队列已满时，新的任务将被阻塞，直到队列中有空间可用。
defaultAsyncSenderExecutor：线程池中的线程数量与 CPU 核心数量相同；线程存活时间为 60 秒，即如果一个线程在 60 秒内没有执行任务，则会被回收。

java

复制代码

public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) { // 保存传入的 DefaultMQProducer 对象和 RPCHook 对象 this.defaultMQProducer = defaultMQProducer; this.rpcHook = rpcHook; // 创建一个有界阻塞队列，容量为 50000，用于存储异步发送任务 this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000); // 创建一个异步发送线程池 this.defaultAsyncSenderExecutor = new ThreadPoolExecutor( // 设置核心线程数为当前可用的处理器数量 Runtime.getRuntime().availableProcessors(), // 设置最大线程数为当前可用的处理器数量 Runtime.getRuntime().availableProcessors(), // 设置线程存活时间为 60 秒 1000 * 60, TimeUnit.MILLISECONDS, // 使用上面创建的有界阻塞队列作为任务队列 this.asyncSenderThreadPoolQueue, // 创建一个 ThreadFactory，用于创建新的线程 new ThreadFactory() { private AtomicInteger threadIndex = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { // 创建一个新的线程，名称为 AsyncSenderExecutor_1、AsyncSenderExecutor_2、... return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet()); } }); }