Producer跟Consumer对于RocketMQ来说都是客户端，我们在使用的过程中初始化其中的一个实例来进行发送或消费消息。

DefaultMQProducer

DefaultMQProducer是默认的Producer实现类，实现了MQProducer接口，而MQProducer又实现了MQAdmin接口。

org.apache.rocketmq.client.MQAdmin接口结构

  1. MQAdmin是RocketMQ管理的基本接口
  2. public interface MQAdmin {
  3.     /**
  4.      * 创建Topic
  5.      *
  6.      * @param key 索引key
  7.      * @param newTopic topic名称
  8.      * @param queueNum Topic队列数量
  9.      */
  10.     void createTopic(final String key, final String newTopic, final int queueNum)
  11.         throws MQClientException;
  13.      /**
  14.      * 创建Topic
  15.      *
  16.      * @param key 索引key
  17.      * @param newTopic topic名称
  18.      * @param queueNum Topic队列数量
  19.      * @param topicSysFlag Topic系统标记
  20.      */
  21.     void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
  22.         throws MQClientException;
  24.     /**
  25.      * 根据时间戳获取消息队列偏移量
  26.      *
  27.      * @param 消息队列实例
  28.      * @param 时间戳(毫秒)
  29.      * @return 偏移量
  30.      */
  31.     long searchOffset(final MessageQueue mq, final long timestamp) throws MQClientException;
  33.     /**
  34.      * 获取最大偏移量
  35.      *
  36.      * @param 消息队列实例
  37.      * @return 最大偏移量
  38.      */
  39.     long maxOffset(final MessageQueue mq) throws MQClientException;
  41.     /**
  42.      * 获取最小偏移量
  43.      *
  44.      * @param 消息队列实例
  45.      * @return 获取最小偏移量
  46.      */
  47.     long minOffset(final MessageQueue mq) throws MQClientException;
  49.     /**
  50.      * 获取最早存储消息的时间
  51.      *
  52.      * @param 消息队列实例
  53.      * @return 消息时间戳
  54.      */
  55.     long earliestMsgStoreTime(final MessageQueue mq) throws MQClientException;
  57.     /**
  58.      * 根据消息ID获取消息
  59.      *
  60.      * @param 偏移量消息ID
  61.      * @return 消息
  62.      */
  63.     MessageExt viewMessage(final String offsetMsgId) throws RemotingException, MQBrokerException,
  64.         InterruptedException, MQClientException;
  66.     /**
  67.      * 查询消息
  68.      *
  69.      * @param topic 消息Topic
  70.      * @param key 消息索引
  71.      * @param maxNum 最大消息数
  72.      * @param begin 开始时间
  73.      * @param end 结束时间
  74.      * @return 结果集
  75.      */
  76.     QueryResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
  77.         final long end) throws MQClientException, InterruptedException;
  79.     /**
  80.      * 根据Topic和消息ID返回消息扩展实例
  81.      */
  82.     MessageExt viewMessage(String topic,
  83.         String msgId) throws RemotingException, MQBrokerException, InterruptedException, MQClientException;
  85. }

org.apache.rocketmq.client.producer.MQProducer结构

MQProducer的方法总结：
MQProducer中消息发送大致分为三种：单向，异步，同步
其中对消息发送有多种扩展机制：超时，指定队列，指定消息选择算法
request：表示该结果的返回条件为consumer已经消费了该消息并进行回复，producer利用同步或异步的方式获得consumer的回复消息，注：该模式效率低下。

  1. public interface MQProducer extends MQAdmin {
  2.     //启动
  3.     void start() throws MQClientException;
  4.     //停止
  5.     void shutdown();
  6.     //查找主题下所有消息队列
  7.     List<MessageQueue> fetchPublishMessageQueues(final String topic) throws MQClientException;
  8.     //同步发送消息
  9.     SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,
  10.     InterruptedException;
  11.     //同步发送消息(有超时机制)
  12.     SendResult send(final Message msg, final long timeout) throws MQClientException,
  13.     RemotingException, MQBrokerException, InterruptedException;
  14.     //异步发送消息+发送回调
  15.     void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
  16.     RemotingException, InterruptedException;
  17.     //异步发送消息+发送回调(有超时机制)
  18.     void send(final Message msg, final SendCallback sendCallback, final long timeout)
  19.         throws MQClientException, RemotingException, InterruptedException;
  20.     //单向发送消息
  21.     void sendOneway(final Message msg) throws MQClientException, RemotingException,
  22.     InterruptedException;
  23.     //同步发送消息(指定消息队列)
  24.     SendResult send(final Message msg, final MessageQueue mq) throws MQClientException,
  25.     RemotingException, MQBrokerException, InterruptedException;
  26.     //同步发送消息(指定消息队列+超时机制)
  27.     SendResult send(final Message msg, final MessageQueue mq, final long timeout)
  28.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  29.     //异步发送消息+发送回调(指定消息队列)
  30.     void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback)
  31.         throws MQClientException, RemotingException, InterruptedException;
  32.     //异步发送消息+发送回调(指定消息队列+超时机制)
  33.     void send(final Message msg, final MessageQueue mq, final SendCallback sendCallback, long timeout)
  34.         throws MQClientException, RemotingException, InterruptedException;
  35.     //单向发送消息(指定消息队列)
  36.     void sendOneway(final Message msg, final MessageQueue mq) throws MQClientException,
  37.     RemotingException, InterruptedException;
  38.     //同步发送消息(指定消息选择算法,会覆盖默认的消息队列负载)
  39.     SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg)
  40.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  41.     //同步发送消息(指定消息选择算法,会覆盖默认的消息队列负载+超时机制)
  42.     SendResult send(final Message msg, final MessageQueueSelector selector, final Object arg,
  43.                     final long timeout) throws MQClientException, RemotingException, MQBrokerException,
  44.     InterruptedException;
  45.     //异步发送消息+发送回调(指定消息选择算法,会覆盖默认的消息队列负载)
  46.     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
  47.               final SendCallback sendCallback) throws MQClientException, RemotingException,
  48.     InterruptedException;
  49.     //异步发送消息+发送回调(指定消息选择算法,会覆盖默认的消息队列负载+超时机制)
  50.     void send(final Message msg, final MessageQueueSelector selector, final Object arg,
  51.               final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
  52.     InterruptedException;
  53.     //单向发送消息(指定消息选择算法,会覆盖默认的消息队列负载)
  54.     void sendOneway(final Message msg, final MessageQueueSelector selector, final Object arg)
  55.         throws MQClientException, RemotingException, InterruptedException;
  56.     //同步事务发送消息(指定事务执行器)
  57.     TransactionSendResult sendMessageInTransaction(final Message msg,
  58.                                                    final LocalTransactionExecuter tranExecuter, final Object arg) throws MQClientException;
  59.     //同步事务发送消息
  60.     TransactionSendResult sendMessageInTransaction(final Message msg,
  61.                                                    final Object arg) throws MQClientException;
  63.     //同步批量发送消息
  64.     SendResult send(final Collection<Message> msgs) throws MQClientException, RemotingException, MQBrokerException,
  65.     InterruptedException;
  66.     //同步批量发送消息(超时机制)
  67.     SendResult send(final Collection<Message> msgs, final long timeout) throws MQClientException,
  68.     RemotingException, MQBrokerException, InterruptedException;
  69.     //同步批量发送消息(指定队列)
  70.     SendResult send(final Collection<Message> msgs, final MessageQueue mq) throws MQClientException,
  71.     RemotingException, MQBrokerException, InterruptedException;
  72.     //同步批量发送消息(指定队列+超时机制)
  73.     SendResult send(final Collection<Message> msgs, final MessageQueue mq, final long timeout)
  74.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  75.     //异步批量发送消息
  76.     void send(final Collection<Message> msgs, final SendCallback sendCallback) throws MQClientException, RemotingException, MQBrokerException,
  77.     InterruptedException;
  78.     //异步批量发送消息(超时机制)
  79.     void send(final Collection<Message> msgs, final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException,
  80.     MQBrokerException, InterruptedException;
  81.     //异步批量发送消息(指定队列)
  82.     void send(final Collection<Message> msgs, final MessageQueue mq, final SendCallback sendCallback) throws MQClientException, RemotingException,
  83.     MQBrokerException, InterruptedException;
  84.     //异步批量发送消息(指定队列+超时机制)
  85.     void send(final Collection<Message> msgs, final MessageQueue mq, final SendCallback sendCallback, final long timeout) throws MQClientException,
  86.     RemotingException, MQBrokerException, InterruptedException;
  88.     //request方法解释：使用request方法的返回结果时表示消费者已经消费该消息并且进行回复
  89.     //Send request message in synchronous mode. This method returns 
  90.     //    only when the consumer consume the request message and reply a message.
  91.     //同步+超时
  92.     Message request(final Message msg, final long timeout) throws RequestTimeoutException, MQClientException,
  93.     RemotingException, MQBrokerException, InterruptedException;
  94.     //异步+超时
  95.     void request(final Message msg, final RequestCallback requestCallback, final long timeout)
  96.         throws MQClientException, RemotingException, InterruptedException, MQBrokerException;
  97.     //同步+消息选择算法+超时
  98.     Message request(final Message msg, final MessageQueueSelector selector, final Object arg,
  99.                     final long timeout) throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException,
  100.     InterruptedException;
  101.     //异步+消息选择算法+超时
  102.     void request(final Message msg, final MessageQueueSelector selector, final Object arg,
  103.                  final RequestCallback requestCallback,
  104.                  final long timeout) throws MQClientException, RemotingException,
  105.     InterruptedException, MQBrokerException;
  106.     //同步+指定队列+超时
  107.     Message request(final Message msg, final MessageQueue mq, final long timeout)
  108.         throws RequestTimeoutException, MQClientException, RemotingException, MQBrokerException, InterruptedException;
  109.     //异步+指定队列+超时
  110.     void request(final Message msg, final MessageQueue mq, final RequestCallback requestCallback, long timeout)
  111.         throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
  112. }

DefaultMQProducer核心属性

  1.     //生产者组
  2.     private String producerGroup;
  4.     //默认TopicKey
  5.     private String createTopicKey = TopicValidator.AUTO_CREATE_TOPIC_KEY_TOPIC;
  7.     //默认Topic队列数
  8.     private volatile int defaultTopicQueueNums = 4;
  10.     //发送消息超时时间,默认3s
  11.     private int sendMsgTimeout = 3000;
  13.     //压缩消息(超过默认值则启动压缩，默认4kb)
  14.     private int compressMsgBodyOverHowmuch = 1024 * 4;
  16.     //失败重试次数(同步)
  17.     private int retryTimesWhenSendFailed = 2;
  19.     //失败重试次数(异步)
  20.     private int retryTimesWhenSendAsyncFailed = 2;
  22.     //发送失败时重试发送给其他Broker开关
  23.     private boolean retryAnotherBrokerWhenNotStoreOK = false;
  25.     //消息最大长度
  26.     private int maxMessageSize = 1024 * 1024 * 4; // 4M
  28.     //消息追踪器
  29.     private TraceDispatcher traceDispatcher = null;