一：发送方式讲解
RocketMQ 版提供三种方式来发送消息:同步（Sync）发送、异步（Async）发送和单向（Oneway）发送。 我们会介绍每种发送方式的原理、应用场景、代码差异，以及三种发送方式的对比。
1.1 同步发送
CommunicationMode#SYNC
原理：
同步发送是指发送者向 MQ 执行发送消息 API 时，同步等待，直到消息服务器返回发送结果 。
应用场景：
此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。
同步发送接口介绍：
MQProducer#send// 同步-发送消息SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException;
复制代码
备注：
同步发送是指消息发送方发出一条消息后，会在收到服务端同步响应之后才发下一条消息的通讯方式。1.2 异步发送
CommunicationMode#ASYNC
原理：
发送者向 MQ 执行发送消息 API 时，指定消息发送成功后的回掉函数，然后调用消息发送 API 后，立即返回，消息发送者线程不阻塞，直到运行结束，消息发送成功或失败的回调任务在一个新的线程中执行 。
应用场景：
异步发送一般用于链路耗时较长，对响应时间较为敏感的业务场景，例如，您视频上传后通知启动转码服务，转码完成后通知推送转码结果等。
1.2.1 异步发送接口介绍
MQProducer#send//异步 发送消息， sendCallback参数是消息发送成功后的回调方法 。void send(final Message msg, final SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException;
复制代码
1.2.2 异步相关核心属性构造器介绍
DefaultMQProducerImpl:异步相关核心属性构造器介绍
//异步发送队列，默认长度:5wprivate final BlockingQueue<Runnable> asyncSenderThreadPoolQueue;//默认异步发送线程池private final ExecutorService defaultAsyncSenderExecutor;//可以自定义的-异步发送消息线程池private ExecutorService asyncSenderExecutor;//构造器public DefaultMQProducerImpl(final DefaultMQProducer defaultMQProducer, RPCHook rpcHook) { this.defaultMQProducer = defaultMQProducer; this.rpcHook = rpcHook; //有界队列,长度:5w this.asyncSenderThreadPoolQueue = new LinkedBlockingQueue<Runnable>(50000); this.defaultAsyncSenderExecutor = new ThreadPoolExecutor( Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors(), 1000 * 60, TimeUnit.MILLISECONDS, this.asyncSenderThreadPoolQueue, new ThreadFactory() { private AtomicInteger threadIndex = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { return new Thread(r, "AsyncSenderExecutor_" + this.threadIndex.incrementAndGet()); } });}
复制代码
谨记:
如未设置自定义线程池：
asyncSenderExecutor
将会使用默认线程池：
defaultAsyncSenderExecutor
默认线程池任务队列默认:5w.如队列任务超出 5w，线程池拒绝策略默认为:拒绝策略，可能会有丢失消息发送的风险。
扩展：
异步发送 netty 网络发送模块使用了 Semaphore，如遇性能调优或问题排查，别忘了！！！
备注：
异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。消息队列 RocketMQ 版的异步发送，需要您实现异步发送回调的以下接口：SendCallback 消息发送方在发送了一条消息后，不需要等待服务端响应即可发送第二条消息。发送方通过回调接口接收服务端响应，并处理响应结果。
1.3 单向发送
CommunicationMode#ONEWAY
原理：
消息发送者向 MQ 执行发送消息 API 时，直接返回，不等待消息服务器的结果， 也不注册回调函数，简单地说，就是只管发，不在乎消息是否成功存储在消息服务器上 。
应用场景：
适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。单向发送接口介绍：
MQProducer#sendOneway//单向消息 发送，就是不在乎发送结果，消息发送出去后该方法立 即返回 。void sendOneway(final Message msg) throws MQClientException, RemotingException, InterruptedException;
复制代码
备注：
发送方只负责发送消息，不等待服务端返回响应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。
三种发送方式的对比
下表概括了三者的特点和主要区别：
二：消息类型讲解
目前开源版本 RocketMq 生产端支持发送的消息类型为：普通消息、批量消息、延时消息、事物消息(开源版本定时消息和顺序消息目前不支持，顺序消息可变向实现) 。
我们将简单介绍消息和使用这些消息类型的注意事项；我们先分析消息相关的类图关系：
分析：
Message 核心属性和方法分析：
public class Message implements Serializable { private static final long serialVersionUID = 8445773977080406428L; //主题 topic private String topic; //消息-Flag一些特殊的消息标记，int类型。标记的含义定义在 MessageSysFlag 中 private int flag; /** * 扩展属性 * TAGS: 消息TAG，用于消息过滤 * KEYS: Message 索引键， 多个用空格隔开， RocketMQ 可以根据这些 key 快速检索到消息 。 * WAIT: 消息发送时是否等消息存储完成后再返回 * DELAY: 消息延迟级别，用于定时消息或消息重试 。 */ private Map<String, String> properties; //消息体 private byte[] body; // 事务Id private String transactionId; ...省略...
// 消息延迟级别，用于定时消息或消息重试 。 public void setDelayTimeLevel(int level) { this.putProperty(MessageConst.PROPERTY_DELAY_TIME_LEVEL, String.valueOf(level)); }}
复制代码
备注：延迟级别对应时间的是下面的常量：
MessageConst.PROPERTY_DELAY_TIME_LEVEL
//private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";//DELAY: 消息延迟级别，用于定时消息或消息重试 。--properties属性public static final String PROPERTY_DELAY_TIME_LEVEL = "DELAY";
复制代码
MessageBath 核心属性和方法分析：
/** * 批量消息，实现了 Iterable 迭代接口 * 批量消息发送是将 同一主题 的多条消息一起 打包发送到消息服务端，减少网络调用次数，提高网络传输效率 。 * 当然，并不是在同一批次中发送的消息数量越多性能就越好，其判断依据是单条消息的长度，如果单条消息内容比较长， 则打包多条消息发送会影响其他 线程发送消息的响应时间 ， * 并且单批次消息发送总长度不能超过 DefaultMQProducer#maxMessageSize。批量消息 发送要解决的是 如何将这些消息 编码以便服务端能够正确解码出每条 消息的消息内容 。 */public class MessageBatch extends Message implements Iterable<Message> {
private static final long serialVersionUID = 621335151046335557L; private final List<Message> messages;
private MessageBatch(List<Message> messages) { this.messages = messages; } /** * 消息编码 * @return */ public byte[] encode() { return MessageDecoder.encodeMessages(messages); } public Iterator<Message> iterator() { return messages.iterator(); } /** * 消息转换， messages -> MessageBatch * 1>批量消息 不支持 延时消息 * 2>消息主题 必须一致 * 3>消息WAIT 必须一致 * @param messages * @return */ public static MessageBatch generateFromList(Collection<Message> messages) { assert messages != null; assert messages.size() > 0; List<Message> messageList = new ArrayList<Message>(messages.size()); Message first = null; for (Message message : messages) { if (message.getDelayTimeLevel() > 0) { //批量消息 不支持 延时消息 throw new UnsupportedOperationException("TimeDelayLevel in not supported for batching"); } if (message.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) { //不支持重试topic(%RETRY%) throw new UnsupportedOperationException("Retry Group is not supported for batching"); } if (first == null) { first = message; } else { if (!first.getTopic().equals(message.getTopic())) { throw new UnsupportedOperationException("The topic of the messages in one batch should be the same"); } if (first.isWaitStoreMsgOK() != message.isWaitStoreMsgOK()) { throw new UnsupportedOperationException("The waitStoreMsgOK of the messages in one batch should the same"); } } messageList.add(message); } MessageBatch messageBatch = new MessageBatch(messageList); messageBatch.setTopic(first.getTopic()); messageBatch.setWaitStoreMsgOK(first.isWaitStoreMsgOK()); return messageBatch; }}
复制代码
小结:
普通消息： 消息队列 RocketMQ 版中无特性的消息，没有其他特性属性，就是普通的 Message 对象, 区别于有特性的定时和延时消息、顺序消息和事务消息；
延时消息： 生产者对指定消息进行延时投递，如果客户端发送延时消息 Message 中的 properties 属性必须包含 DELAY 属性 key；
批量消息： 其实就是 Message 的集合，多了一些验证；
事务消息： 后续单独讲解。
三：结论
本文简单讲解了生产端消息发送方式的区别，开源版本消息类型的区别，知识点小但很重要，建议就是源码都懂了，那些高大上理论概念是不是很简单了？
程序员的核心竞争力其实还是技术，因此对技术还是要不断的学习，关注 “IT 巅峰技术” 公众号 ，该公众号内容定位：中高级开发、架构师、中层管理人员等中高端岗位服务的，除了技术交流外还有很多架构思想和实战案例。
作者是 《 消息中间件 RocketMQ 技术内幕》 一书作者，同时也是 “RocketMQ 上海社区”联合创始人，曾就职于拼多多、德邦等公司，现任上市快递公司架构负责人，主要负责开发框架的搭建、中间件相关技术的二次开发和运维管理、混合云及基础服务平台的建设。
评论