【RocketMq-生产者】消息发送者参数详解

private String clientIP = RemotingUtil.getLocalAddress();

private String instanceName = System.getProperty("rocketmq.client.name", "DEFAULT");

private int clientCallbackExecutorThreads = Runtime.getRuntime().availableProcessors();

/**  
 * Pulling topic information interval from the named server */
   private int pollNameServerInterval = 1000 * 30;

/**  
 * Heartbeat interval in microseconds with message broker */
   private int heartbeatBrokerInterval = 1000 * 30;

/**  
 * Offset persistent interval for consumer */
   private int persistConsumerOffsetInterval = 1000 * 5;

/**  
 * Wrapping internal implementations for virtually all methods presented in this class. */
protected final transient DefaultMQProducerImpl defaultMQProducerImpl;

InternalLogger log = ClientLogger.getLog()

/**  
 * 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 <a href="http://rocketmq.apache.org/docs/core-concept/">core concepts</a> for more discussion.  
 */
 private String producerGroup;

/**  
 * Number of queues to create per default topic. */private volatile int defaultTopicQueueNums = 4;

/**  
 * Timeout for sending messages. */
   private int sendMsgTimeout = 3000;

long costTimeAsync = System.currentTimeMillis() - beginStartTime;  
if (timeout < costTimeAsync) {  
    throw new RemotingTooMuchRequestException("sendKernelImpl call timeout");  
}

/**  
 * Compress message body threshold, namely, message body larger than 4k will be compressed on default. */
 // 压缩消息体阈值，即默认压缩大于4k的消息体。
   private int compressMsgBodyOverHowmuch = 1024 * 4;

private boolean tryToCompressMessage(final Message msg) {  
	// 批量数据目前不支持压缩
    if (msg instanceof MessageBatch) {  
        //batch does not support compressing right now  
        return false;  
    }  
    byte[] body = msg.getBody();  
    if (body != null) {  
        if (body.length >= this.defaultMQProducer.getCompressMsgBodyOverHowmuch()) {  
            try {  
	            // 压缩之后的数据
                byte[] data = compressor.compress(body, compressLevel);  
                if (data != null) {  
                    msg.setBody(data);  
                    return true;                }  
            } catch (IOException e) {  
                log.error("tryToCompressMessage exception", e);  
                log.warn(msg.toString());  
            }  
        }  
    }  
  
    return false;  
}

int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;

/**  
 * 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;

/**  
 * Maximum allowed message body size in bytes. */
   private int maxMessageSize = 1024 * 1024 * 4; // 4M

...

if (msg.getBody().length > defaultMQProducer.getMaxMessageSize()) {  
    throw new MQClientException(ResponseCode.MESSAGE_ILLEGAL,  
        "the message body size over max value, MAX: " + defaultMQProducer.getMaxMessageSize());  
}

// 轨迹消息中累计到3/4左右的时候就进行合并提交
if (currentMsgSize >= traceProducer.getMaxMessageSize() - 10 * 1000) {  
    List<TraceTransferBean> dataToSend = new ArrayList(traceTransferBeanList);  
    AsyncDataSendTask asyncDataSendTask = new AsyncDataSendTask(traceTopicName, regionId, dataToSend);  
    traceExecutor.submit(asyncDataSendTask);  
  
    this.clear();  
  
}

private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};

/**
-   String key：根据 key 查找 Broker，即新主题创建在哪些 Broker 上
-   String newTopic：主题名称
-   int queueNum：主题队列个数
-   int topicSysFlag：主题的系统参数
*/
void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
 
/**
	根据队列与时间戳，从消息消费队列中查找消息，返回消息的物理偏移量（在 commitlog 文件中的偏移量）。
	MessageQueue mq：消息消费队列
	long timestamp：时间戳
*/
long searchOffset(MessageQueue mq, long timestamp)

 /**
 查询消息消费队列当前最大的逻辑偏移量，在 consumequeue 文件中的偏移量。
 */
long maxOffset(final MessageQueue mq)
    
 /**
 查询消息消费队列当前最小的逻辑偏移量。
 */
long minOffset(final MessageQueue mq) 
    
/**
返回消息消费队列中第一条消息的存储时间戳。
*/
long earliestMsgStoreTime(MessageQueue mq)
    
/**
根据消息的物理偏移量查找消息
*/
MessageExt viewMessage(String offsetMsgId)
    
/**
根据主题与消息的全局唯一 ID 查找消息。
*/    
MessageExt viewMessage(String topic, String msgId)
    
/**
批量查询消息，其参数列表如下：

String topic：主题名称
String key：消息索引 Key
int maxNum：本次查询最大返回消息条数
long begin：开始时间戳
long end：结束时间戳
*/
QueryResult queryMessage(String topic, String key, int maxNum, long begin,long end)