聊聊rocketmq5的RocketMQClientTemplate
序
本文主要研究一下rocketmq5的RocketMQClientTemplate
rocketmq-spring
自从rocketmq推出了5版本之后,rocketmq-spring就区分了4.x与5.x版本,4.x版本是rocketmq-spring开头,5.x版本是rocketmq-v5-client开头;rocketmq5的RocketMQClientTemplate对应的是旧版的RocketMQTemplate,他们都同样继承了org.springframework.messaging.core.AbstractMessageSendingTemplate,只是各自扩展的方法发生的变化。
RocketMQTemplate
主要是:
- sendAndReceive
- syncSend、syncSendDelayTimeSeconds、syncSendDelayTimeMills、syncSendDeliverTimeMills、syncSendOrderly
- asyncSend、asyncSendOrderly
- sendOneWay、sendOneWayOrderly
- sendMessageInTransaction
RocketMQClientTemplate
主要是:
- syncSendNormalMessage、syncSendFifoMessage、syncSendDelayMessage
- asyncSendNormalMessage、asyncSendFifoMessage、asyncSendDelayMessage
- asyncSendWithObjectPayload、asyncSendWithStringPayload、asyncSendWithBytePayload、asyncSendWithMessagePayload
- sendMessageInTransaction
- receive、receiveAsync
- ack、ackAsync
MessageSendingOperations
org/springframework/messaging/core/MessageSendingOperations.java
public interface MessageSendingOperations<D> {
/**
* Send a message to a default destination.
* @param message the message to send
*/
void send(Message<?> message) throws MessagingException;
/**
* Send a message to the given destination.
* @param destination the target destination
* @param message the message to send
*/
void send(D destination, Message<?> message) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message and send it to a default destination.
* @param payload the Object to use as payload
*/
void convertAndSend(Object payload) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message and send it to the given destination.
* @param destination the target destination
* @param payload the Object to use as payload
*/
void convertAndSend(D destination, Object payload) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message with the given headers and send it to
* the given destination.
* @param destination the target destination
* @param payload the Object to use as payload
* @param headers the headers for the message to send
*/
void convertAndSend(D destination, Object payload, Map<String, Object> headers) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message, apply the given post processor, and send
* the resulting message to a default destination.
* @param payload the Object to use as payload
* @param postProcessor the post processor to apply to the message
*/
void convertAndSend(Object payload, @Nullable MessagePostProcessor postProcessor) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message, apply the given post processor, and send
* the resulting message to the given destination.
* @param destination the target destination
* @param payload the Object to use as payload
* @param postProcessor the post processor to apply to the message
*/
void convertAndSend(D destination, Object payload, MessagePostProcessor postProcessor) throws MessagingException;
/**
* Convert the given Object to serialized form, possibly using a
* {@link org.springframework.messaging.converter.MessageConverter},
* wrap it as a message with the given headers, apply the given post processor,
* and send the resulting message to the given destination.
* @param destination the target destination
* @param payload the Object to use as payload
* @param headers the headers for the message to send
* @param postProcessor the post processor to apply to the message
*/
void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers,
@Nullable MessagePostProcessor postProcessor) throws MessagingException;
}
spring-messaging的MessageSendingOperations接口定义了send、convertAndSend方法
AbstractMessageSendingTemplate
org/springframework/messaging/core/AbstractMessageSendingTemplate.java
public abstract class AbstractMessageSendingTemplate<D> implements MessageSendingOperations<D> {
/**
* Name of the header that can be set to provide further information
* (e.g. a {@code MethodParameter} instance) about the origin of the
* payload, to be taken into account as a conversion hint.
* @since 4.2
*/
public static final String CONVERSION_HINT_HEADER = "conversionHint";
protected final Log logger = LogFactory.getLog(getClass());
@Nullable
private D defaultDestination;
private MessageConverter converter = new SimpleMessageConverter();
/**
* Configure the default destination to use in send methods that don't have
* a destination argument. If a default destination is not configured, send methods
* without a destination argument will raise an exception if invoked.
*/
public void setDefaultDestination(@Nullable D defaultDestination) {
this.defaultDestination = defaultDestination;
}
/**
* Return the configured default destination.
*/
@Nullable
public D getDefaultDestination() {
return this.defaultDestination;
}
/**
* Set the {@link MessageConverter} to use in {@code convertAndSend} methods.
* <p>By default, {@link SimpleMessageConverter} is used.
* @param messageConverter the message converter to use
*/
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "MessageConverter must not be null");
this.converter = messageConverter;
}
/**
* Return the configured {@link MessageConverter}.
*/
public MessageConverter getMessageConverter() {
return this.converter;
}
@Override
public void send(Message<?> message) {
send(getRequiredDefaultDestination(), message);
}
protected final D getRequiredDefaultDestination() {
Assert.state(this.defaultDestination != null, "No 'defaultDestination' configured");
return this.defaultDestination;
}
@Override
public void send(D destination, Message<?> message) {
doSend(destination, message);
}
protected abstract void doSend(D destination, Message<?> message);
@Override
public void convertAndSend(Object payload) throws MessagingException {
convertAndSend(payload, null);
}
@Override
public void convertAndSend(D destination, Object payload) throws MessagingException {
convertAndSend(destination, payload, (Map<String, Object>) null);
}
@Override
public void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers)
throws MessagingException {
convertAndSend(destination, payload, headers, null);
}
@Override
public void convertAndSend(Object payload, @Nullable MessagePostProcessor postProcessor)
throws MessagingException {
convertAndSend(getRequiredDefaultDestination(), payload, postProcessor);
}
@Override
public void convertAndSend(D destination, Object payload, @Nullable MessagePostProcessor postProcessor)
throws MessagingException {
convertAndSend(destination, payload, null, postProcessor);
}
@Override
public void convertAndSend(D destination, Object payload, @Nullable Map<String, Object> headers,
@Nullable MessagePostProcessor postProcessor) throws MessagingException {
Message<?> message = doConvert(payload, headers, postProcessor);
send(destination, message);
}
/**
* Convert the given Object to serialized form, possibly using a
* {@link MessageConverter}, wrap it as a message with the given
* headers and apply the given post processor.
* @param payload the Object to use as payload
* @param headers the headers for the message to send
* @param postProcessor the post processor to apply to the message
* @return the converted message
*/
protected Message<?> doConvert(Object payload, @Nullable Map<String, Object> headers,
@Nullable MessagePostProcessor postProcessor) {
MessageHeaders messageHeaders = null;
Object conversionHint = (headers != null ? headers.get(CONVERSION_HINT_HEADER) : null);
Map<String, Object> headersToUse = processHeadersToSend(headers);
if (headersToUse != null) {
if (headersToUse instanceof MessageHeaders) {
messageHeaders = (MessageHeaders) headersToUse;
}
else {
messageHeaders = new MessageHeaders(headersToUse);
}
}
MessageConverter converter = getMessageConverter();
Message<?> message = (converter instanceof SmartMessageConverter ?
((SmartMessageConverter) converter).toMessage(payload, messageHeaders, conversionHint) :
converter.toMessage(payload, messageHeaders));
if (message == null) {
String payloadType = payload.getClass().getName();
Object contentType = (messageHeaders != null ? messageHeaders.get(MessageHeaders.CONTENT_TYPE) : null);
throw new MessageConversionException("Unable to convert payload with type='" + payloadType +
"', contentType='" + contentType + "', converter=[" + getMessageConverter() + "]");
}
if (postProcessor != null) {
message = postProcessor.postProcessMessage(message);
}
return message;
}
/**
* Provides access to the map of input headers before a send operation.
* Subclasses can modify the headers and then return the same or a different map.
* <p>This default implementation in this class returns the input map.
* @param headers the headers to send (or {@code null} if none)
* @return the actual headers to send (or {@code null} if none)
*/
@Nullable
protected Map<String, Object> processHeadersToSend(@Nullable Map<String, Object> headers) {
return headers;
}
}
AbstractMessageSendingTemplate对MessageSendingOperations的接口做了实现,它定义了
doSend(D destination, Message<?> message)
需要子类实现
RocketMQClientTemplate
rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/core/RocketMQClientTemplate.java
@SuppressWarnings({"WeakerAccess", "unused"})
public class RocketMQClientTemplate extends AbstractMessageSendingTemplate<String> implements DisposableBean {
private static final Logger log = LoggerFactory.getLogger(RocketMQClientTemplate.class);
private ProducerBuilder producerBuilder;
private SimpleConsumerBuilder simpleConsumerBuilder;
private volatile Producer producer;
private volatile SimpleConsumer simpleConsumer;
private RocketMQMessageConverter rocketMQMessageConverter = new RocketMQMessageConverter();
private String charset = "UTF-8";
@Override
protected void doSend(String destination, Message<?> message) {
SendReceipt sendReceipt = syncSendGrpcMessage(destination, message, null, null);
if (log.isDebugEnabled()) {
log.debug("send message to `{}` finished. result:{}", destination, sendReceipt);
}
}
/**
* @param destination formats: `topicName:tags`
* @param message {@link Message} the message to be sent.
* @param messageDelayTime Time for message delay
* @param messageGroup message group name
* @return SendReceipt Synchronous Task Results
*/
public SendReceipt syncSendGrpcMessage(String destination, Message<?> message, Duration messageDelayTime, String messageGroup) {
if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
log.error("send request message failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
SendReceipt sendReceipt = null;
try {
org.apache.rocketmq.client.apis.message.Message rocketMsg = this.createRocketMQMessage(destination, message, messageDelayTime, messageGroup);
Producer grpcProducer = this.getProducer();
try {
sendReceipt = grpcProducer.send(rocketMsg);
log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
} catch (Throwable t) {
log.error("Failed to send message", t);
}
} catch (Exception e) {
log.error("send request message failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(e.getMessage(), e);
}
return sendReceipt;
}
private org.apache.rocketmq.client.apis.message.Message createRocketMQMessage(String destination, Message<?> message, Duration messageDelayTime, String messageGroup) {
Message<?> msg = this.doConvert(message.getPayload(), message.getHeaders(), null);
return RocketMQUtil.convertToClientMessage(getMessageConverter(), charset,
destination, msg, messageDelayTime, messageGroup);
}
//......
}
RocketMQClientTemplate继承了AbstractMessageSendingTemplate,其doSend方法调用的是syncSendGrpcMessage;该方法主要是调用父类的doConvert转换为spring-messaging的Message,之后再通过RocketMQUtil.convertToClientMessage转换为org.apache.rocketmq.client.apis.message.Message,最后通过grpcProducer.send(rocketMsg)进行发送
syncSendNormalMessage
public SendReceipt syncSendNormalMessage(String destination, Object payload) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, null);
}
public SendReceipt syncSendNormalMessage(String destination, String payload) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, null);
}
public SendReceipt syncSendNormalMessage(String destination, Message<?> message) {
return syncSendGrpcMessage(destination, message, null, null);
}
public SendReceipt syncSendNormalMessage(String destination, byte[] payload) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, null);
}
syncSendNormalMessage主要是通过MessageBuilder构建Message,然后通过syncSendGrpcMessage进行发送
syncSendFifoMessage
public SendReceipt syncSendFifoMessage(String destination, Object payload, String messageGroup) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, messageGroup);
}
public SendReceipt syncSendFifoMessage(String destination, String payload, String messageGroup) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, messageGroup);
}
public SendReceipt syncSendFifoMessage(String destination, byte[] payload, String messageGroup) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, null, messageGroup);
}
public SendReceipt syncSendFifoMessage(String destination, Message<?> message, String messageGroup) {
return syncSendGrpcMessage(destination, message, null, messageGroup);
}
syncSendFifoMessage也是通过MessageBuilder构建Message,只是通过syncSendGrpcMessage进行发送时指定了messageGroup
syncSendDelayMessage
public SendReceipt syncSendDelayMessage(String destination, Object payload, Duration messageDelayTime) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, messageDelayTime, null);
}
public SendReceipt syncSendDelayMessage(String destination, String payload, Duration messageDelayTime) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, messageDelayTime, null);
}
public SendReceipt syncSendDelayMessage(String destination, byte[] payload, Duration messageDelayTime) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return syncSendGrpcMessage(destination, message, messageDelayTime, null);
}
public SendReceipt syncSendDelayMessage(String destination, Message<?> message, Duration messageDelayTime) {
return syncSendGrpcMessage(destination, message, messageDelayTime, null);
}
syncSendDelayMessage也是通过MessageBuilder构建Message,只是通过syncSendGrpcMessage进行发送时指定了messageDelayTime
asyncSend
public CompletableFuture<SendReceipt> asyncSend(String destination, Message<?> message, Duration messageDelayTime, String messageGroup, CompletableFuture<SendReceipt> future) {
if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
log.error("send request message failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
Producer grpcProducer = this.getProducer();
try {
org.apache.rocketmq.client.apis.message.Message rocketMsg = this.createRocketMQMessage(destination, message, messageDelayTime, messageGroup);
future = grpcProducer.sendAsync(rocketMsg);
} catch (Exception e) {
log.error("send request message failed. destination:{}, message:{} ", destination, message);
throw new MessagingException(e.getMessage(), e);
}
return future;
}
public CompletableFuture<SendReceipt> asyncSendWithObjectPayload(String destination, Object payload, Duration messageDelayTime, String messageGroup, CompletableFuture<SendReceipt> future) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return asyncSend(destination, message, messageDelayTime, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendWithStringPayload(String destination, String payload, Duration messageDelayTime, String messageGroup, CompletableFuture<SendReceipt> future) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return asyncSend(destination, message, messageDelayTime, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendWithBytePayload(String destination, byte[] payload, Duration messageDelayTime, String messageGroup, CompletableFuture<SendReceipt> future) {
Message<?> message = MessageBuilder.withPayload(payload).build();
return asyncSend(destination, message, messageDelayTime, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendWithMessagePayload(String destination, Message<?> payload, Duration messageDelayTime, String messageGroup, CompletableFuture<SendReceipt> future) {
return asyncSend(destination, payload, messageDelayTime, messageGroup, future);
}
asyncSend与syncSendGrpcMessage类似先通过createRocketMQMessage转换为org.apache.rocketmq.client.apis.message.Message,只是最后调用的是grpcProducer.sendAsync来发送;asyncSendWithObjectPayload、asyncSendWithStringPayload、asyncSendWithBytePayload、asyncSendWithMessagePayload最后调用的是asyncSend方法
asyncSendNormalMessage
public CompletableFuture<SendReceipt> asyncSendNormalMessage(String destination, Object payload, CompletableFuture<SendReceipt> future) {
return asyncSendWithObjectPayload(destination, payload, null, null, future);
}
public CompletableFuture<SendReceipt> asyncSendNormalMessage(String destination, String payload, CompletableFuture<SendReceipt> future) {
return asyncSendWithStringPayload(destination, payload, null, null, future);
}
public CompletableFuture<SendReceipt> asyncSendNormalMessage(String destination, byte[] payload, CompletableFuture<SendReceipt> future) {
return asyncSendWithBytePayload(destination, payload, null, null, future);
}
public CompletableFuture<SendReceipt> asyncSendNormalMessage(String destination, Message<?> payload, CompletableFuture<SendReceipt> future) {
return asyncSendWithMessagePayload(destination, payload, null, null, future);
}
asyncSendNormalMessage委托给了asyncSendWithObjectPayload、asyncSendWithStringPayload、asyncSendWithBytePayload、asyncSendWithMessagePayload;最后调用的是asyncSend方法
asyncSendFifoMessage
public CompletableFuture<SendReceipt> asyncSendFifoMessage(String destination, Object payload, String messageGroup, CompletableFuture<SendReceipt> future) {
return asyncSendWithObjectPayload(destination, payload, null, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendFifoMessage(String destination, String payload, String messageGroup, CompletableFuture<SendReceipt> future) {
return asyncSendWithStringPayload(destination, payload, null, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendFifoMessage(String destination, byte[] payload, String messageGroup, CompletableFuture<SendReceipt> future) {
return asyncSendWithBytePayload(destination, payload, null, messageGroup, future);
}
public CompletableFuture<SendReceipt> asyncSendFifoMessage(String destination, Message<?> payload, String messageGroup, CompletableFuture<SendReceipt> future) {
return asyncSendWithMessagePayload(destination, payload, null, messageGroup, future);
}
asyncSendFifoMessage委托给了asyncSendWithObjectPayload、asyncSendWithStringPayload、asyncSendWithBytePayload、asyncSendWithMessagePayload,只是传递了messageGroup
asyncSendDelayMessage
public CompletableFuture<SendReceipt> asyncSendDelayMessage(String destination, Object payload, Duration messageDelayTime, CompletableFuture<SendReceipt> future) {
return asyncSendWithObjectPayload(destination, payload, messageDelayTime, null, future);
}
public CompletableFuture<SendReceipt> asyncSendDelayMessage(String destination, String payload, Duration messageDelayTime, CompletableFuture<SendReceipt> future) {
return asyncSendWithStringPayload(destination, payload, messageDelayTime, null, future);
}
public CompletableFuture<SendReceipt> asyncSendDelayMessage(String destination, byte[] payload, Duration messageDelayTime, CompletableFuture<SendReceipt> future) {
return asyncSendWithBytePayload(destination, payload, messageDelayTime, null, future);
}
public CompletableFuture<SendReceipt> asyncSendDelayMessage(String destination, Message<?> payload, Duration messageDelayTime, CompletableFuture<SendReceipt> future) {
return asyncSendWithMessagePayload(destination, payload, messageDelayTime, null, future);
}
asyncSendDelayMessage委托了asyncSendWithObjectPayload、asyncSendWithStringPayload、asyncSendWithBytePayload、asyncSendWithMessagePayload,只是传递了messageDelayTime
sendMessageInTransaction
public Pair<SendReceipt, Transaction> sendMessageInTransaction(String destination, Object payload) throws ClientException {
Message<?> message = MessageBuilder.withPayload(payload).build();
return sendTransactionMessage(destination, message);
}
public Pair<SendReceipt, Transaction> sendMessageInTransaction(String destination, String payload) throws ClientException {
Message<?> message = MessageBuilder.withPayload(payload).build();
return sendTransactionMessage(destination, message);
}
public Pair<SendReceipt, Transaction> sendMessageInTransaction(String destination, byte[] payload) throws ClientException {
Message<?> message = MessageBuilder.withPayload(payload).build();
return sendTransactionMessage(destination, message);
}
/**
* @param destination formats: `topicName:tags`
* @param message {@link Message} the message to be sent.
* @return CompletableFuture<SendReceipt> Asynchronous Task Results
*/
public Pair<SendReceipt, Transaction> sendTransactionMessage(String destination, Message<?> message) {
if (Objects.isNull(message) || Objects.isNull(message.getPayload())) {
log.error("send request message failed. destination:{}, message is null ", destination);
throw new IllegalArgumentException("`message` and `message.payload` cannot be null");
}
final SendReceipt sendReceipt;
Producer grpcProducer = this.getProducer();
org.apache.rocketmq.client.apis.message.Message rocketMsg = this.createRocketMQMessage(destination, message, null, null);
final Transaction transaction;
try {
transaction = grpcProducer.beginTransaction();
sendReceipt = grpcProducer.send(rocketMsg, transaction);
log.info("Send transaction message successfully, messageId={}", sendReceipt.getMessageId());
} catch (ClientException e) {
log.error("send request message failed. destination:{}, message:{} ", destination, message);
throw new RuntimeException(e);
}
return new Pair<>(sendReceipt, transaction);
}
sendMessageInTransaction先通过MessageBuilder构建消息,最后调用的是sendTransactionMessage;sendTransactionMessage也是先通过createRocketMQMessage转为org.apache.rocketmq.client.apis.message.Message,最后通过grpcProducer.beginTransaction()、grpcProducer.send(rocketMsg, transaction)来发送事务消息
receive
public List<MessageView> receive(int maxMessageNum, Duration invisibleDuration) throws ClientException {
SimpleConsumer simpleConsumer = this.getSimpleConsumer();
return simpleConsumer.receive(maxMessageNum, invisibleDuration);
}
public CompletableFuture<List<MessageView>> receiveAsync(int maxMessageNum, Duration invisibleDuration) throws ClientException, IOException {
SimpleConsumer simpleConsumer = this.getSimpleConsumer();
CompletableFuture<List<MessageView>> listCompletableFuture = simpleConsumer.receiveAsync(maxMessageNum, invisibleDuration);
simpleConsumer.close();
return listCompletableFuture;
}
receive通过simpleConsumer.receive来拉取;receiveAsync则是通过simpleConsumer.receiveAsync来拉取
ack
public void ack(MessageView message) throws ClientException {
SimpleConsumer simpleConsumer = this.getSimpleConsumer();
simpleConsumer.ack(message);
}
public CompletableFuture<Void> ackAsync(MessageView messageView) {
SimpleConsumer simpleConsumer = this.getSimpleConsumer();
return simpleConsumer.ackAsync(messageView);
}
ack是通过simpleConsumer.ack来操作;ackAsync是通过simpleConsumer.ackAsync来操作
小结
rocketmq5的RocketMQClientTemplate继承了org.springframework.messaging.core.AbstractMessageSendingTemplate,实现了spring-messaging的MessageSendingOperations接口,同时它自己扩展了如下方法
- syncSendNormalMessage、syncSendFifoMessage、syncSendDelayMessage
- asyncSendNormalMessage、asyncSendFifoMessage、asyncSendDelayMessage
- sendMessageInTransaction
- receive、receiveAsync
- ack、ackAsync
其发送消息底层依赖的是grpcProducer.send、sendAsync、send(rocketMsg, transaction),而消费相关的则是依赖的SimpleConsumer
原文地址:https://blog.csdn.net/hello_ejb3/article/details/140733608
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!