一、概述
RocketMQ的producer默认有两个，一个是DefaultMQProducer，另一个是TransactionMQProducer，本文只对send方法做一个总结，其他的细节在其他章节介绍
二、DefaultMQProducer
一共定义了17种send方法，从4.x版本，事务消息被放到了TransactionMQProducer中，所以有15个send方法，这15个方法中，又有两个异步带超时时间的send方法被废弃了，所以有效的send方法有13个：
1、同步发送
/**
* 同步发送模式. 只有消息被成功接收并且被固化完成后才会收到反馈。
* 内置有重发机制, producer将会重试
* {@link #retryTimesWhenSendFailed,default=2} 次 ，然后才会报错.
* 因此，有一定的概率向broker发送重复的消息
* 使用者有责任去解决潜在的重复数据造成的影响
* @param msg 待发送数据
* @return {@link SendResult} 实体，来通知发送者发送状态等信息, 比如消息的ID
* {@link SendStatus} 指明 broker 存储/复制 的状态, 发送到了哪个队列等等
* @throws MQClientException 客户端异常
* @throws RemotingException 网络连接异常
* @throws MQBrokerException broker异常
* @throws InterruptedException 发送线程中断异常
*/
@Override
public SendResult send(
Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg);
}
2、同步发送，带超时时间
/**
* 与 {@link #send(Message)} 相同，只不过多了超时时间的指定.
*
* @param msg 待发送消息
* @param timeout 发送超时时间
* @return {@link SendResult} 同上
* {@link SendStatus} 同上
* @throws MQClientException 客户端异常
* @throws RemotingException 网络连接异常
* @throws MQBrokerException broker异常
* @throws InterruptedException 发送线程中断异常
*/
@Override
public SendResult send(Message msg,long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, timeout);
}
3、异步发送
/**
* 异步发送消息
* 消息发送后，立即返回。broker处理返程后, 触发sendCallback回调方法
* 与上面一样，在给出发送失败标志前，会尝试2次，所以开发者要处理重复发送带来的问题
* @param msg 待发送消息
* @param sendCallback 回调函数
* @throws MQClientException 客户端异常
* @throws RemotingException 网络异常
* @throws InterruptedException 发送线程中断异常
*/
@Override
public void send(Message msg,SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, sendCallback);
}
4、异步发送，带超时时间
@Override
public void send(Message msg, SendCallback sendCallback, long timeout)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, sendCallback, timeout);
}
5、单向发送，不等待broker回馈
/**
* 发送方法不会等待broker的反馈，只会一直发
* 所以有很高的吞吐量，但是有一定概率丢失消息
*
* @param msg 待发送消息
* @throws MQClientException 客户端异常
* @throws RemotingException 网络异常
* @throws InterruptedException 发送线程中断异常
*/
@Override
public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.sendOneway(msg);
}
6、同步发送，指定队列
/**
* 同步发送，指定队列
* @param msg 待发送消息
* @param mq 指定的消息队列
* @return {@link SendResult} 同上
* {@link SendStatus} 同上
*/
@Override
public SendResult send(Message msg, MessageQueue mq)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, mq);
}
7、同步发送，指定队列，并附带超时时间
@Override
public SendResult send(Message msg, MessageQueue mq, long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, mq, timeout);
}
8、异步发送，指定队列
@Override
public void send(Message msg, MessageQueue mq, SendCallback sendCallback)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, mq, sendCallback);
}
9、异步发送，指定队列，附带超时时间
这个在4.4.0版本被设置为废弃，后续版本会给出
/**
* 因为在处超时异常存在问题，所以废弃
*/
@Override
public void send(Message msg, MessageQueue mq, SendCallback sendCallback, long timeout)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, mq, sendCallback, timeout);
}
10、单向发送，指定队列
@Override
public void sendOneway(Message msg,
MessageQueue mq) throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.sendOneway(msg, mq);
}
11、同步发送，指定队列选择策略
官方的有序消息的DEMO就是基于队列选择器做的，让一些列有序的消息（相同ID）发送到同一个队列
/**
* 指定队列选择策略MessageQueueSelector
*
* @param msg 待发送消息
* @param selector 队列选择器
* @param arg 配合队列选择器选择队列的参数，一般可以是业务参数（ID等）
* @return {@link SendResult} 同上
* {@link SendStatus} 同上
*/
@Override
public SendResult send(Message msg, MessageQueueSelector selector, Object arg)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, selector, arg);
}
12、同步发送消息，指定队列选择策略，并附带超时时间
@Override
public SendResult send(Message msg, MessageQueueSelector selector, Object arg, long timeout)
throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
return this.defaultMQProducerImpl.send(msg, selector, arg, timeout);
}
13、异步发送消息，指定队列选择策略
@Override
public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, selector, arg, sendCallback);
}
14、异步发送消息，指定队列选择策略，并附带超时时间
这个方法在4.4.0版本废弃，后续提供
@Override
public void send(Message msg, MessageQueueSelector selector, Object arg, SendCallback sendCallback, long timeout)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.send(msg, selector, arg, sendCallback, timeout);
}
15、单向发送，指定队列选择策略
@Override
public void sendOneway(Message msg, MessageQueueSelector selector, Object arg)
throws MQClientException, RemotingException, InterruptedException {
this.defaultMQProducerImpl.sendOneway(msg, selector, arg);
}
三、TransactionMQProducer
1、发送事务消息
@Override
public TransactionSendResult sendMessageInTransaction(final Message msg,
final Object arg) throws MQClientException {
if (null == this.transactionListener) {
throw new MQClientException("TransactionListener is null", null);
}
return this.defaultMQProducerImpl.sendMessageInTransaction(msg, null, arg);
}
网友评论