RocketMQ 生产者 Producer 发送消息三种方式分析
2018-12-11 本文已影响42人
jijs
概述
Producer 发送消息，RocketMQ 提供了三种模式。
- 同步发送
- 异步发送
- OneWay 发送
示例代码如下：
// 1、同步发送
SendResult sendResult = producer.send(msg);
//2、异步发送
producer.send(msg, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
}
@Override
public void onException(Throwable e) {
}
});
//3、 Oneway发送
producer.sendOneway(msg);
-
1、同步发送
Producer 向 broker 发送消息，阻塞当前线程等待 broker 响应 发送结果。 -
2、异步发送
Producer 首先构建一个向 broker 发送消息的任务，把该任务提交给线程池，等执行完该任务时，回调用户自定义的回调函数，执行处理结果。 -
3、Oneway 发送
Oneway 方式只负责发送请求，不等待应答，Producer 只负责把请求发出去，而不处理响应结果。
一、同步发送
org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl
private SendResult sendDefaultImpl(Message msg,
final CommunicationMode communicationMode,
final SendCallback sendCallback, final long timeout
) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
//省略代码
......
TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
if (topicPublishInfo != null && topicPublishInfo.ok()) {
//省略代码
......
// 1、计算重发次数
int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;
int times = 0;
String[] brokersSent = new String[timesTotal];
// 2、循环执行发送消息
for (; times < timesTotal; times++) {
String lastBrokerName = null == mq ? null : mq.getBrokerName();
// 3、选择要发送的 MessageQueue
MessageQueue mqSelected = this.selectOneMessageQueue(topicPublishInfo, lastBrokerName);
if (mqSelected != null) {
mq = mqSelected;
brokersSent[times] = mq.getBrokerName();
try {
beginTimestampPrev = System.currentTimeMillis();
long costTime = beginTimestampPrev - beginTimestampFirst;
if (timeout < costTime) {
callTimeout = true;
break;
}
// 4、调用 sendKernelImpl 方法进行发送消息
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout - costTime);
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
switch (communicationMode) {
case ASYNC:
return null;
case ONEWAY:
return null;
case SYNC:
// 5、如果发送失败，则continue，继续循环发送，发送成功则直接 return 返回
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
if (this.defaultMQProducer.isRetryAnotherBrokerWhenNotStoreOK()) {
continue;
}
}
return sendResult;
default:
break;
}
}
//省略代码
......
} else {
break;
}
}
if (sendResult != null) {
return sendResult;
}
//省略代码
......
}
1、计算消息最多发送的次数。
同步发送：org.apache.rocketmq.client.producer.DefaultMQProducer#retryTimesWhenSendFailed + 1，默认retryTimesWhenSendFailed是2，所以除了正常调用一次外，发送消息如果失败了会重试2次。总共会发送3次，如果3次都失败则返回发送失败的消息。
异步发送：不会重试（调用总次数等于1）
2、循环执行发送消息
如果发送的消息未成功发送，则循环继续发送，直到发送的次数达到 timesTotal 。
3、选择要发送的 MessageQueue
选择消息要发送的 MessageQueue。
4、调用 sendKernelImpl 方法进行发送消息
5、如果发送失败，则continue，继续循环发送，发送成功则直接 return 返回
同步发送原理
RocketMQ 通讯是使用 Netty 进行发送的，Netty 通讯默认都是异步的，那么同步是怎么实现的呢？
public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
final long timeoutMillis)
throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
final int opaque = request.getOpaque();
try {
// 1、定义一个 ResponseFuture 来处理响应结果
final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
this.responseTable.put(opaque, responseFuture);
final SocketAddress addr = channel.remoteAddress();
channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture f) throws Exception {
if (f.isSuccess()) {
// 2、通过 Netty 执行完成后回调处理请求的结果
responseFuture.setSendRequestOK(true);
return;
} else {
responseFuture.setSendRequestOK(false);
}
responseTable.remove(opaque);
responseFuture.setCause(f.cause());
// 唤醒阻塞的线程
responseFuture.putResponse(null);
log.warn("send a request command to channel <" + addr + "> failed.");
}
});
// 3、请求结果默认等待请求3秒钟，如果超过3秒则抛出异常。
RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
if (null == responseCommand) {
if (responseFuture.isSendRequestOK()) {
throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
responseFuture.getCause());
} else {
throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
}
}
return responseCommand;
} finally {
this.responseTable.remove(opaque);
}
}
1、定义一个 ResponseFuture 来处理响应结果
public class ResponseFuture {
//省略其它代码
......
private final CountDownLatch countDownLatch = new CountDownLatch(1);
public RemotingCommand waitResponse(final long timeoutMillis) throws InterruptedException {
this.countDownLatch.await(timeoutMillis, TimeUnit.MILLISECONDS);
return this.responseCommand;
}
public void putResponse(final RemotingCommand responseCommand) {
this.responseCommand = responseCommand;
this.countDownLatch.countDown();
}
}
ResponseFuture 内部使用了 CountDownLatch 来实现的。当调用waitResponse() 方法时阻塞当前线程，当返回结果时调用 putResponse() 方法存放结果，然后执行this.countDownLatch.countDown() 唤醒阻塞的线程。
2、通过 Netty 执行完成后回调处理请求的结果
使用 Netty 进行发送消息，当 Netty 收到结果后会执行自定义的 ChannelFutureListener.operationComplete()方法。
如果执行完成，调用responseFuture.putResponse(null);立即唤醒阻塞的线程，处理请求结果。
3、默认最长等待请求3秒钟，如果超过3秒则抛出异常。
调用 responseFuture.waitResponse(timeoutMillis)方法阻塞等待 Netty 返回结果。默认最长等待时间为3秒，如果超过3秒则认为调用超时，抛出 RemotingSendRequestException 异常信息。
二、异步发送
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() {
@Override
public void run() {
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeout > costTime) {
try {
// 执行发送消息
sendDefaultImpl(msg, CommunicationMode.ASYNC, sendCallback, timeout - costTime);
} catch (Exception e) {
sendCallback.onException(e);
}
} else {
sendCallback.onException(
new RemotingTooMuchRequestException("DEFAULT ASYNC send call timeout"));
}
}
});
} catch (RejectedExecutionException e) {
throw new MQClientException("executor rejected ", e);
}
}
异步发送的消息是把发送的消息提交给线程池来进行调度执行的。因为不需要同步返回结果。
异步发送原理
public void invokeAsyncImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis,
final InvokeCallback invokeCallback)
throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
long beginStartTime = System.currentTimeMillis();
final int opaque = request.getOpaque();
//1、尝试获得 semaphore 信号量，semaphore 默认为65535。
boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);
long costTime = System.currentTimeMillis() - beginStartTime;
if (timeoutMillis < costTime) {
throw new RemotingTooMuchRequestException("invokeAsyncImpl call timeout");
}
// 2、定义 ResponseFuture 来处理响应的结果
final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis - costTime, invokeCallback, once);
// 存储到把 ResponseFuture 存储到 responseTable 中。
this.responseTable.put(opaque, responseFuture);
try {
// 3、调用 Netty 发送数据
channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture f) throws Exception {
// 3.1、成功则设置 responseFuture 的状态为发送成功
if (f.isSuccess()) {
responseFuture.setSendRequestOK(true);
return;
}
// 3.2、发送失败则快速处理失败请求
requestFail(opaque);
log.warn("send a request command to channel <{}> failed.", RemotingHelper.parseChannelRemoteAddr(channel));
}
});
} catch (Exception e) {
responseFuture.release();
log.warn("send a request command to channel <" + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
}
} else {
if (timeoutMillis <= 0) {
throw new RemotingTooMuchRequestException("invokeAsyncImpl invoke too fast");
} else {
String info =
String.format("invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
timeoutMillis,
this.semaphoreAsync.getQueueLength(),
this.semaphoreAsync.availablePermits()
);
log.warn(info);
throw new RemotingTimeoutException(info);
}
}
}
1、尝试获得 semaphore 信号量，semaphore 默认为65535。
public static final int CLIENT_ASYNC_SEMAPHORE_VALUE
=Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_ASYNC_SEMAPHORE_VALUE, "65535"));
异步发送请求的并发量，默认最大为65535。
2、定义 ResponseFuture 来处理响应的结果
ResponseFuture responseFuture 定义了发送数据响应的结果，同上面介绍的同步发送。
把responseFuture 存储到 responseTable 中。
protected final ConcurrentMap<Integer /* opaque */, ResponseFuture> responseTable =
new ConcurrentHashMap<Integer, ResponseFuture>(256);
opaque 为请求的唯一标识，每次请求创建一个新的（AtomicInteger 自增的）。
3、调用 Netty 发送数据
调用 Netty 异步发送数据。
- 3.1、 Netty 发送成功则设置 responseFuture 的状态为发送成功
- 3.2、发送失败则快速处理失败请求
快速失败是相对下面的定时任务扫描处理响应结果的
处理异步的响应结果
//1、 定时间隔3秒钟扫描一次 responseTable
this.timer.scheduleAtFixedRate(new TimerTask() {
@Override
public void run() {
try {
NettyRemotingServer.this.scanResponseTable();
} catch (Throwable e) {
log.error("scanResponseTable exception", e);
}
}
}, 1000 * 3, 1000);
public void scanResponseTable() {
final List<ResponseFuture> rfList = new LinkedList<ResponseFuture>();
Iterator<Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
while (it.hasNext()) {
Entry<Integer, ResponseFuture> next = it.next();
ResponseFuture rep = next.getValue();
// 2、扫描开始执行时间大于 执行超时+1s 的 ResponseFuture 数据，并存放到 rfList 中
if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System.currentTimeMillis()) {
rep.release();
it.remove();
rfList.add(rep);
log.warn("remove timeout request, " + rep);
}
}
// 3、执行处理完成或超时的请求
for (ResponseFuture rf : rfList) {
try {
executeInvokeCallback(rf);
} catch (Throwable e) {
log.warn("scanResponseTable, operationComplete Exception", e);
}
}
}
1、 定时间隔3秒钟扫描一次 responseTable
当 Netty 客户端和服务端启动的时候，都会自动这个定时任务。定时的扫描 responseTable 的请求数据。每隔3秒扫描一次。
2、扫描开始执行时间大于 执行超时+1s 的 ResponseFuture 数据
Netty 请求的时候默认会有等待的执行超时时间（或可自己设置），超过超时时间的则认为任务超时，需要通过定时任务处理超时的任务。
异步执行完成的请求也会在定时任务中回调执行处理结果。
三、Oneway 发送
public void invokeOnewayImpl(final Channel channel, final RemotingCommand request, final long timeoutMillis)
throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
request.markOnewayRPC();
// 尝试获得 semaphore 信号量，semaphore 默认为65535。
boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
if (acquired) {
final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreOneway);
try {
// 2、调用 Netty 发送请求
channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture f) throws Exception {
once.release();
if (!f.isSuccess()) {
log.warn("send a request command to channel <" + channel.remoteAddress() + "> failed.");
}
}
});
} catch (Exception e) {
once.release();
log.warn("write send a request command to channel <" + channel.remoteAddress() + "> failed.");
throw new RemotingSendRequestException(RemotingHelper.parseChannelRemoteAddr(channel), e);
}
} else {
if (timeoutMillis <= 0) {
throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
} else {
String info = String.format(
"invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
timeoutMillis,
this.semaphoreOneway.getQueueLength(),
this.semaphoreOneway.availablePermits()
);
log.warn(info);
throw new RemotingTimeoutException(info);
}
}
}
1、尝试获得 semaphore 信号量，semaphore 默认为65535。
public static final int CLIENT_ONEWAY_SEMAPHORE_VALUE
= Integer.parseInt(System.getProperty(COM_ROCKETMQ_REMOTING_CLIENT_ONEWAY_SEMAPHORE_VALUE, "65535"));
Oneway 方式发送请求的并发量，默认最大为65535。
2、调用 Netty 发送请求
这里只是发送数据，而没有处理响应结果。
这种方式发送数据，吞吐量更高，但不管数据是否发送成功。
