一：发送方式讲解
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;//有界队列,长度:5wthis.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);@Overridepublic 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作为消息的顶层对象，在生产端可表示各种消息；MessageBatch表示批量消息；MessageExt以及它的子类其实都是在Broker端存储或查询使用，后续可仔细分析哈；
Message 核心属性和方法分析：
public class Message implements Serializable {private static final long serialVersionUID = 8445773977080406428L;//主题 topicprivate String topic;//消息-Flag一些特殊的消息标记，int类型。标记的含义定义在 MessageSysFlag 中private int flag;/*** 扩展属性* TAGS: 消息TAG，用于消息过滤* KEYS: Message 索引键， 多个用空格隔开， RocketMQ 可以根据这些 key 快速检索到消息 。* WAIT: 消息发送时是否等消息存储完成后再返回* DELAY: 消息延迟级别，用于定时消息或消息重试 。*/private Map<String, String> properties;//消息体private byte[] body;// 事务Idprivate 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上海社区”联合创始人，曾就职于拼多多、德邦等公司，现任上市快递公司架构负责人，主要负责开发框架的搭建、中间件相关技术的二次开发和运维管理、混合云及基础服务平台的建设。
