rocketmq源码剖析之producer配置

梦里梦外; 2024-04-17 21:18 80阅读 0赞

在rocketmq中，`MQProducer`是承载消息发送的，消息的发送又可以分为常规消息的发送和事务消息的发送，其中常规消息发送用的是`DefaultMQProducer`，事务消息的发送用的是`TransactionMQProducer`。他们集成关系图如：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4ODUxNTAz_size_16_color_FFFFFF_t_70]

## DefaultMQProducer ##

在`DefaultMQProducer` 中有一些常规的producer端的配置，如下所示：

*  `producerGroup` : 生产者所属的组；
 *  `defaultTopicQueueNums`：在创建topic时，topic上默认的队列数，默认为4 ；
 *  `sendMsgTimeout`：发送消息的超时时间，默认为3000ms；
 *  `compressMsgBodyOverHowmuch`：压缩消息体的阙值，默认为1024\*4=4k，即当消息的body大小为4k的时候对body内容进行压缩；
 *  `retryTimesWhenSendFailed`：同步发送失败时重试的次数，默认为2次；
 *  `retryTimesWhenSendAsyncFailed`：异步发送失败时重试的次数，默认为2次；
 *  `retryAnotherBrokerWhenNotStoreOK`：表示当发送失败的时候是否重试发送到另一个broker，默认为false；
 *  `maxMessageSize`：消息body最大值，1024\*1024\*4 = 4M

`DefaultMQProducer`相关配置源码信息如下：

/** * Producer group conceptually aggregates all producer instances of exactly same role, which is particularly * important when transactional messages are involved. * </p> * * For non-transactional messages, it does not matter as long as it's unique per process. * </p> * * See {@linktourl http://rocketmq.apache.org/docs/core-concept/} for more discussion. */
        private String producerGroup;
    
        /** * Just for testing or demo program */
        private String createTopicKey = MixAll.AUTO_CREATE_TOPIC_KEY_TOPIC;
    
        /** * Number of queues to create per default topic. */
        private volatile int defaultTopicQueueNums = 4;
    
        /** * Timeout for sending messages. */
        private int sendMsgTimeout = 3000;
    
        /** * Compress message body threshold, namely, message body larger than 4k will be compressed on default. */
        private int compressMsgBodyOverHowmuch = 1024 * 4;
    
        /** * Maximum number of retry to perform internally before claiming sending failure in synchronous mode. * </p> * * This may potentially cause message duplication which is up to application developers to resolve. */
        private int retryTimesWhenSendFailed = 2;
    
        /** * Maximum number of retry to perform internally before claiming sending failure in asynchronous mode. * </p> * * This may potentially cause message duplication which is up to application developers to resolve. */
        private int retryTimesWhenSendAsyncFailed = 2;
    
        /** * Indicate whether to retry another broker on sending failure internally. */
        private boolean retryAnotherBrokerWhenNotStoreOK = false;
    
        /** * Maximum allowed message size in bytes. */
        private int maxMessageSize = 1024 * 1024 * 4; // 4M

## TransactionMQProducer ##

在`TransactionMQProducer`中的附加配置项主要是针对事务回查的，如下所示：

private int checkThreadPoolMinSize = 1;
        private int checkThreadPoolMaxSize = 1;
        private int checkRequestHoldMax = 2000;

其中`checkRequestHoldMax`是事务回查请求阻塞等待队列的最大值，即最多支持2000个请求等待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4ODUxNTAz_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/cd426517aa5146e6a759ed8d4b56498e.png