企业中一般都会封装rocketmq 同步 异步 单向方法，你只需要配置好nameserver地址 topic tag 消息体等，然后调用封装方法进行发送即可。
流程差不多如下
1、导入mq依赖
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>4.8.0</version> </dependency>
2、消息发送者步骤
创建消息生成者producer，指定组名
指定Nameserver mq地址
启动生产者
创建消息对象，指定Topic、Tag和消息体
发送消息
关闭生产者
3、消费者步骤
创建消费者Consumer，指定组名
指定Nameserver地址
订阅主题Topic和Tag
（listener监听消息，去消费先到的消息）
设置回调函数，处理消息
启动消费者
同步 异步 单向底层区别
//同步 SendResult send = producer.send(msg); //异步 producer.send(msg, new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.println("===" + sendResult); } @Override public void onException(Throwable throwable) { System.out.println("e:" + throwable); } }); //单向 producer.sendOneway(msg);
异步相对同步来说send方法多了些参数SendCallback sendCallback。而这个sendCallback重写的方法有成功和失败的方法。
单向相对同步来说换了一个方法sendOneway.
在跟踪源码去看到
异步
同步
具体实现方法
一下源码传参异步，单向的区别在
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
以及switch里面
switch(communicationMode) { case ASYNC: return null; case ONEWAY: return null; case SYNC: if (sendResult.getSendStatus() == SendStatus.SEND_OK || !this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { return sendResult; } default: break label122; }
具体如下
sendDefaultImpl方法（核心：所有同步、异步、单向）
private SendResult sendDefaultImpl(Message msg, CommunicationMode communicationMode, SendCallback sendCallback, long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException { //校验 Producer 处于运行状态 this.makeSureStateOK(); //校验消息格式 Validators.checkMessage(msg, this.defaultMQProducer); long invokeID = this.random.nextLong(); long beginTimestampFirst = System.currentTimeMillis(); long beginTimestampPrev = beginTimestampFirst; //获取Topic路由信息 TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic()); if (topicPublishInfo != null && topicPublishInfo.ok()) { boolean callTimeout = false; //发送的队列 MessageQueue mq = null; //异常 Exception exception = null; //最后一次发送结果 SendResult sendResult = null; //计算调用发送消息到成功为止的最大次数：同步、单向调用一次 ；异步3次 int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1; //第几次发送 int times = 0; //存储每次发送消息现在broker String[] brokersSent = new String[timesTotal]; while(true) { label122: { String info; if (times < timesTotal) { info = null == mq ? null : mq.getBrokerName(); //选择要发送的消息队列 MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, info); if (mqSelected != null) { mq = mqSelected; brokersSent[times] = mqSelected.getBrokerName(); long endTimestamp; try { beginTimestampPrev = System.currentTimeMillis(); long costTime = beginTimestampPrev - beginTimestampFirst; if (timeout >= costTime) { //Kernel内核的意思，调用发送消息核心方法 sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime); endTimestamp = System.currentTimeMillis(); //更新Broker可用性信息，在选择发送到的消息队列时，会参考Broker发送消息的延迟。源码多次操作Broker,可见不懂Broker就无法真正理解mq源码。 this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false); switch(communicationMode) { case ASYNC: return null; case ONEWAY: return null; case SYNC: if (sendResult.getSendStatus() == SendStatus.SEND_OK || !this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) { return sendResult; } default: break label122; } } callTimeout = true; } catch (RemotingException var26) { //当抛出RemotingException时，如果进行消息发送失败重试，则可能导致消息发送重复。例如，发送消息超时(RemotingTimeoutException)，实际Broker接收到该消息并处理成功。因此，Consumer在消费时，需要保证幂等性。 //这也是我们在生产环境看到可能出现多次重复发送的情况。 //更新Broker可用信息，更新继续循环 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var26); this.log.warn(msg.toString()); exception = var26; break label122; } catch (MQClientException var27) { //更新Broker可用信息，继续循环 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var27); this.log.warn(msg.toString()); exception = var27; break label122; } catch (MQBrokerException var28) { //更新Broker可用信息，部分情况下异常返回，结束循环 endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, true); this.log.warn(String.format("sendKernelImpl exception, resend at once, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var28); this.log.warn(msg.toString()); exception = var28; switch(var28.getResponseCode()) { case 1: case 14: case 16: case 17: case 204: case 205: break label122; //如果有发送结果，返回，没有，抛出异常 default: if (sendResult != null) { return sendResult; } throw var28; } } catch (InterruptedException var29) { endTimestamp = System.currentTimeMillis(); this.updateFaultItem(mqSelected.getBrokerName(), endTimestamp - beginTimestampPrev, false); this.log.warn(String.format("sendKernelImpl exception, throw exception, InvokeID: %s, RT: %sms, Broker: %s", invokeID, endTimestamp - beginTimestampPrev, mqSelected), var29); this.log.warn(msg.toString()); this.log.warn("sendKernelImpl exception", var29); this.log.warn(msg.toString()); throw var29; } } } if (sendResult != null) { return sendResult; } info = String.format("Send [%d] times, still failed, cost [%d]ms, Topic: %s, BrokersSent: %s", times, System.currentTimeMillis() - beginTimestampFirst, msg.getTopic(), Arrays.toString(brokersSent)); info = info + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"); MQClientException mqClientException = new MQClientException(info, (Throwable)exception); if (callTimeout) { throw new RemotingTooMuchRequestException("sendDefaultImpl call timeout"); } if (exception instanceof MQBrokerException) { mqClientException.setResponseCode(((MQBrokerException)exception).getResponseCode()); } else if (exception instanceof RemotingConnectException) { mqClientException.setResponseCode(10001); } else if (exception instanceof RemotingTimeoutException) { mqClientException.setResponseCode(10002); } else if (exception instanceof MQClientException) { mqClientException.setResponseCode(10003); } throw mqClientException; } ++times; } } else { List<String> nsList = this.getmQClientFactory().getMQClientAPIImpl().getNameServerAddressList(); if (null != nsList && !nsList.isEmpty()) { throw (new MQClientException("No route info of this topic, " + msg.getTopic() + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10005); } else { throw (new MQClientException("No name server address, please set it." + FAQUrl.suggestTodo("http://rocketmq.apache.org/docs/faq/"), (Throwable)null)).setResponseCode(10004); } } }
但是异步是在该sendDefaultImpl方法之前做一个异步处理
ExecutorService ：那么mq异步用了线程池来提高异步请求的效率！！
@Deprecated public void send(final Message msg, final SendCallback sendCallback, final long timeout) throws MQClientException, RemotingException, InterruptedException { final long beginStartTime = System.currentTimeMillis(); ExecutorService executor = this.getCallbackExecutor(); try { executor.submit(new Runnable() { public void run() { long costTime = System.currentTimeMillis() - beginStartTime; if (timeout > costTime) { try { DefaultMQProducerImpl.this.sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout - costTime); } catch (Exception var4) { sendCallback.onException(var4); } } else { sendCallback.onException(new RemotingTooMuchRequestException("DEFAULT ASYNC send call timeout")); } } }); } catch (RejectedExecutionException var9) { throw new MQClientException("executor rejected ", var9); } }
NettyRemotingClient定义了多线程
并没有发现是哪种特殊的多线程，因此是通用的多线程。就好比map和hashmap，最后用了map一样。
倒回来看，这里是多线程执行，大大提高了运行效率，并且代码严谨和规范，有判定条件，不同程度的try catch
线程池执行sendSelectImpl方法，传了msg,async异步，sendCallback（成功和失败的方法，发送消息不用等待响应，不管有没有成功或失败，生产者还会发送消息），timeout - costTime（最大等待时间默认3000-消耗时间）
大概看一下入参的情况
这里timeTotal默认是1，异步的时候是3,影响后面循环次数。
那么for循环里面要做什么事情呢
选择消息要发送的队列，准备发送
调用核心发送消息方法
更新Broker可用性信息
上面这个类sendDefaultImpl，有详细注释。
其实还可以继续深入下去，可以模仿rocketmq，自己写一个自定义mq框架那也是很厉害了！！！
