Code refactoring and some new feature support
zkzlx
4 years ago
parent
68c25bc9a7
commit
cbd64061ce
@ -1,89 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
|
||||
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public final class RocketMQBinderUtils {
|
||||
|
||||
private RocketMQBinderUtils() {
|
||||
|
||||
}
|
||||
|
||||
public static RocketMQBinderConfigurationProperties mergeProperties(
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
RocketMQProperties rocketMQProperties) {
|
||||
RocketMQBinderConfigurationProperties result = new RocketMQBinderConfigurationProperties();
|
||||
if (StringUtils.isEmpty(rocketMQProperties.getNameServer())) {
|
||||
result.setNameServer(rocketBinderConfigurationProperties.getNameServer());
|
||||
}
|
||||
else {
|
||||
result.setNameServer(
|
||||
Arrays.asList(rocketMQProperties.getNameServer().split(";")));
|
||||
}
|
||||
if (rocketMQProperties.getProducer() == null
|
||||
|| StringUtils.isEmpty(rocketMQProperties.getProducer().getAccessKey())) {
|
||||
result.setAccessKey(rocketBinderConfigurationProperties.getAccessKey());
|
||||
}
|
||||
else {
|
||||
result.setAccessKey(rocketMQProperties.getProducer().getAccessKey());
|
||||
}
|
||||
if (rocketMQProperties.getProducer() == null
|
||||
|| StringUtils.isEmpty(rocketMQProperties.getProducer().getSecretKey())) {
|
||||
result.setSecretKey(rocketBinderConfigurationProperties.getSecretKey());
|
||||
}
|
||||
else {
|
||||
result.setSecretKey(rocketMQProperties.getProducer().getSecretKey());
|
||||
}
|
||||
if (rocketMQProperties.getProducer() == null || StringUtils
|
||||
.isEmpty(rocketMQProperties.getProducer().getCustomizedTraceTopic())) {
|
||||
result.setCustomizedTraceTopic(
|
||||
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
|
||||
}
|
||||
else {
|
||||
result.setCustomizedTraceTopic(
|
||||
rocketMQProperties.getProducer().getCustomizedTraceTopic());
|
||||
}
|
||||
if (rocketMQProperties.getProducer() != null
|
||||
&& rocketMQProperties.getProducer().isEnableMsgTrace()) {
|
||||
result.setEnableMsgTrace(Boolean.TRUE);
|
||||
}
|
||||
else {
|
||||
result.setEnableMsgTrace(
|
||||
rocketBinderConfigurationProperties.isEnableMsgTrace());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public static String getNameServerStr(List<String> nameServerList) {
|
||||
if (CollectionUtils.isEmpty(nameServerList)) {
|
||||
return null;
|
||||
}
|
||||
return String.join(";", nameServerList);
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,465 +1,470 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq.consuming;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderUtils;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
|
||||
import org.apache.rocketmq.acl.common.AclClientRPCHook;
|
||||
import org.apache.rocketmq.acl.common.SessionCredentials;
|
||||
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
|
||||
import org.apache.rocketmq.client.consumer.MessageSelector;
|
||||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
|
||||
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
|
||||
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
|
||||
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
|
||||
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.common.UtilAll;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.apache.rocketmq.remoting.RPCHook;
|
||||
import org.apache.rocketmq.spring.annotation.ConsumeMode;
|
||||
import org.apache.rocketmq.spring.annotation.MessageModel;
|
||||
import org.apache.rocketmq.spring.annotation.SelectorType;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
|
||||
import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
|
||||
import org.apache.rocketmq.spring.support.RocketMQUtil;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.context.SmartLifecycle;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import static com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKETMQ_RECONSUME_TIMES;
|
||||
|
||||
/**
|
||||
* A class that Listen on rocketmq message.
|
||||
* <p>
|
||||
* this class will delegate {@link RocketMQListener} to handle message
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
|
||||
* @see RocketMQListener
|
||||
*/
|
||||
public class RocketMQListenerBindingContainer
|
||||
implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
|
||||
|
||||
private final static Logger log = LoggerFactory
|
||||
.getLogger(RocketMQListenerBindingContainer.class);
|
||||
|
||||
private long suspendCurrentQueueTimeMillis = 1000;
|
||||
|
||||
/**
|
||||
* Message consume retry strategy<br>
|
||||
* -1,no retry,put into DLQ directly<br>
|
||||
* 0,broker control retry frequency<br>
|
||||
* >0,client control retry frequency.
|
||||
*/
|
||||
private int delayLevelWhenNextConsume = 0;
|
||||
|
||||
private List<String> nameServer;
|
||||
|
||||
private String consumerGroup;
|
||||
|
||||
private String topic;
|
||||
|
||||
private int consumeThreadMax = 64;
|
||||
|
||||
private String charset = "UTF-8";
|
||||
|
||||
private RocketMQListener rocketMQListener;
|
||||
|
||||
private RocketMQHeaderMapper headerMapper;
|
||||
|
||||
private DefaultMQPushConsumer consumer;
|
||||
|
||||
private boolean running;
|
||||
|
||||
private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
|
||||
|
||||
private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
// The following properties came from RocketMQConsumerProperties.
|
||||
private ConsumeMode consumeMode;
|
||||
|
||||
private SelectorType selectorType;
|
||||
|
||||
private String selectorExpression;
|
||||
|
||||
private MessageModel messageModel;
|
||||
|
||||
public RocketMQListenerBindingContainer(
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
RocketMQMessageChannelBinder rocketMQMessageChannelBinder) {
|
||||
this.rocketMQConsumerProperties = rocketMQConsumerProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder;
|
||||
this.consumeMode = rocketMQConsumerProperties.getExtension().getOrderly()
|
||||
? ConsumeMode.ORDERLY : ConsumeMode.CONCURRENTLY;
|
||||
if (StringUtils.isEmpty(rocketMQConsumerProperties.getExtension().getSql())) {
|
||||
this.selectorType = SelectorType.TAG;
|
||||
this.selectorExpression = rocketMQConsumerProperties.getExtension().getTags();
|
||||
}
|
||||
else {
|
||||
this.selectorType = SelectorType.SQL92;
|
||||
this.selectorExpression = rocketMQConsumerProperties.getExtension().getSql();
|
||||
}
|
||||
this.messageModel = rocketMQConsumerProperties.getExtension().getBroadcasting()
|
||||
? MessageModel.BROADCASTING : MessageModel.CLUSTERING;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setupMessageListener(RocketMQListener<?> rocketMQListener) {
|
||||
this.rocketMQListener = rocketMQListener;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() throws Exception {
|
||||
this.setRunning(false);
|
||||
if (Objects.nonNull(consumer)) {
|
||||
consumer.shutdown();
|
||||
}
|
||||
log.info("container destroyed, {}", this.toString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
initRocketMQPushConsumer();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoStartup() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop(Runnable callback) {
|
||||
stop();
|
||||
callback.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (this.isRunning()) {
|
||||
throw new IllegalStateException(
|
||||
"container already running. " + this.toString());
|
||||
}
|
||||
|
||||
try {
|
||||
consumer.start();
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
|
||||
}
|
||||
this.setRunning(true);
|
||||
|
||||
log.info("running container: {}", this.toString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (this.isRunning()) {
|
||||
if (Objects.nonNull(consumer)) {
|
||||
consumer.shutdown();
|
||||
}
|
||||
setRunning(false);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
private void setRunning(boolean running) {
|
||||
this.running = running;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getPhase() {
|
||||
return Integer.MAX_VALUE;
|
||||
}
|
||||
|
||||
private void initRocketMQPushConsumer() throws MQClientException {
|
||||
Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
|
||||
Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
|
||||
Assert.notNull(nameServer, "Property 'nameServer' is required");
|
||||
Assert.notNull(topic, "Property 'topic' is required");
|
||||
|
||||
String ak = rocketBinderConfigurationProperties.getAccessKey();
|
||||
String sk = rocketBinderConfigurationProperties.getSecretKey();
|
||||
if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
|
||||
RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk));
|
||||
consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook,
|
||||
new AllocateMessageQueueAveragely(),
|
||||
rocketBinderConfigurationProperties.isEnableMsgTrace(),
|
||||
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
|
||||
consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook,
|
||||
topic + "|" + UtilAll.getPid()));
|
||||
consumer.setVipChannelEnabled(false);
|
||||
}
|
||||
else {
|
||||
consumer = new DefaultMQPushConsumer(consumerGroup,
|
||||
rocketBinderConfigurationProperties.isEnableMsgTrace(),
|
||||
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
|
||||
}
|
||||
|
||||
consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(nameServer));
|
||||
consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
|
||||
consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());
|
||||
|
||||
switch (messageModel) {
|
||||
case BROADCASTING:
|
||||
consumer.setMessageModel(
|
||||
org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
|
||||
break;
|
||||
case CLUSTERING:
|
||||
consumer.setMessageModel(
|
||||
org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("Property 'messageModel' was wrong.");
|
||||
}
|
||||
|
||||
switch (selectorType) {
|
||||
case TAG:
|
||||
consumer.subscribe(topic, selectorExpression);
|
||||
break;
|
||||
case SQL92:
|
||||
consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("Property 'selectorType' was wrong.");
|
||||
}
|
||||
|
||||
switch (consumeMode) {
|
||||
case ORDERLY:
|
||||
consumer.setMessageListener(new DefaultMessageListenerOrderly());
|
||||
break;
|
||||
case CONCURRENTLY:
|
||||
consumer.setMessageListener(new DefaultMessageListenerConcurrently());
|
||||
break;
|
||||
default:
|
||||
throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
|
||||
}
|
||||
|
||||
if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
|
||||
((RocketMQPushConsumerLifecycleListener) rocketMQListener)
|
||||
.prepareStart(consumer);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "RocketMQListenerBindingContainer{" + "consumerGroup='" + consumerGroup
|
||||
+ '\'' + ", nameServer='" + nameServer + '\'' + ", topic='" + topic + '\''
|
||||
+ ", consumeMode=" + consumeMode + ", selectorType=" + selectorType
|
||||
+ ", selectorExpression='" + selectorExpression + '\'' + ", messageModel="
|
||||
+ messageModel + '}';
|
||||
}
|
||||
|
||||
public long getSuspendCurrentQueueTimeMillis() {
|
||||
return suspendCurrentQueueTimeMillis;
|
||||
}
|
||||
|
||||
public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
|
||||
this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
|
||||
}
|
||||
|
||||
public int getDelayLevelWhenNextConsume() {
|
||||
return delayLevelWhenNextConsume;
|
||||
}
|
||||
|
||||
public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
|
||||
this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
|
||||
}
|
||||
|
||||
public List<String> getNameServer() {
|
||||
return nameServer;
|
||||
}
|
||||
|
||||
public void setNameServer(List<String> nameServer) {
|
||||
this.nameServer = nameServer;
|
||||
}
|
||||
|
||||
public String getConsumerGroup() {
|
||||
return consumerGroup;
|
||||
}
|
||||
|
||||
public void setConsumerGroup(String consumerGroup) {
|
||||
this.consumerGroup = consumerGroup;
|
||||
}
|
||||
|
||||
public String getTopic() {
|
||||
return topic;
|
||||
}
|
||||
|
||||
public void setTopic(String topic) {
|
||||
this.topic = topic;
|
||||
}
|
||||
|
||||
public int getConsumeThreadMax() {
|
||||
return consumeThreadMax;
|
||||
}
|
||||
|
||||
public void setConsumeThreadMax(int consumeThreadMax) {
|
||||
this.consumeThreadMax = consumeThreadMax;
|
||||
}
|
||||
|
||||
public String getCharset() {
|
||||
return charset;
|
||||
}
|
||||
|
||||
public void setCharset(String charset) {
|
||||
this.charset = charset;
|
||||
}
|
||||
|
||||
public RocketMQListener getRocketMQListener() {
|
||||
return rocketMQListener;
|
||||
}
|
||||
|
||||
public void setRocketMQListener(RocketMQListener rocketMQListener) {
|
||||
this.rocketMQListener = rocketMQListener;
|
||||
}
|
||||
|
||||
public DefaultMQPushConsumer getConsumer() {
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public void setConsumer(DefaultMQPushConsumer consumer) {
|
||||
this.consumer = consumer;
|
||||
}
|
||||
|
||||
public ExtendedConsumerProperties<RocketMQConsumerProperties> getRocketMQConsumerProperties() {
|
||||
return rocketMQConsumerProperties;
|
||||
}
|
||||
|
||||
public ConsumeMode getConsumeMode() {
|
||||
return consumeMode;
|
||||
}
|
||||
|
||||
public SelectorType getSelectorType() {
|
||||
return selectorType;
|
||||
}
|
||||
|
||||
public String getSelectorExpression() {
|
||||
return selectorExpression;
|
||||
}
|
||||
|
||||
public MessageModel getMessageModel() {
|
||||
return messageModel;
|
||||
}
|
||||
|
||||
public RocketMQHeaderMapper getHeaderMapper() {
|
||||
return headerMapper;
|
||||
}
|
||||
|
||||
public void setHeaderMapper(RocketMQHeaderMapper headerMapper) {
|
||||
this.headerMapper = headerMapper;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert rocketmq {@link MessageExt} to Spring {@link Message}.
|
||||
* @param messageExt the rocketmq message
|
||||
* @return the converted Spring {@link Message}
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private Message convertToSpringMessage(MessageExt messageExt) {
|
||||
|
||||
// add reconsume-times header to messageExt
|
||||
int reconsumeTimes = messageExt.getReconsumeTimes();
|
||||
messageExt.putUserProperty(ROCKETMQ_RECONSUME_TIMES,
|
||||
String.valueOf(reconsumeTimes));
|
||||
Message message = RocketMQUtil.convertToSpringMessage(messageExt);
|
||||
return MessageBuilder.fromMessage(message)
|
||||
.copyHeaders(headerMapper.toHeaders(messageExt.getProperties())).build();
|
||||
}
|
||||
|
||||
public class DefaultMessageListenerConcurrently
|
||||
implements MessageListenerConcurrently {
|
||||
|
||||
@SuppressWarnings({ "unchecked", "Duplicates" })
|
||||
@Override
|
||||
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
|
||||
ConsumeConcurrentlyContext context) {
|
||||
for (MessageExt messageExt : msgs) {
|
||||
log.debug("received msg: {}", messageExt);
|
||||
try {
|
||||
long now = System.currentTimeMillis();
|
||||
rocketMQListener.onMessage(convertToSpringMessage(messageExt));
|
||||
long costTime = System.currentTimeMillis() - now;
|
||||
log.debug("consume {} message key:[{}] cost: {} ms",
|
||||
messageExt.getMsgId(), messageExt.getKeys(), costTime);
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("consume message failed. messageExt:{}", messageExt, e);
|
||||
context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
|
||||
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
|
||||
}
|
||||
}
|
||||
|
||||
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
|
||||
|
||||
@SuppressWarnings({ "unchecked", "Duplicates" })
|
||||
@Override
|
||||
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
|
||||
ConsumeOrderlyContext context) {
|
||||
for (MessageExt messageExt : msgs) {
|
||||
log.debug("received msg: {}", messageExt);
|
||||
try {
|
||||
long now = System.currentTimeMillis();
|
||||
rocketMQListener.onMessage(convertToSpringMessage(messageExt));
|
||||
long costTime = System.currentTimeMillis() - now;
|
||||
log.info("consume {} message key:[{}] cost: {} ms",
|
||||
messageExt.getMsgId(), messageExt.getKeys(), costTime);
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.warn("consume message failed. messageExt:{}", messageExt, e);
|
||||
context.setSuspendCurrentQueueTimeMillis(
|
||||
suspendCurrentQueueTimeMillis);
|
||||
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
|
||||
}
|
||||
}
|
||||
|
||||
return ConsumeOrderlyStatus.SUCCESS;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
/// *
|
||||
// * Copyright 2013-2018 the original author or authors.
|
||||
// *
|
||||
// * Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// * you may not use this file except in compliance with the License.
|
||||
// * You may obtain a copy of the License at
|
||||
// *
|
||||
// * https://www.apache.org/licenses/LICENSE-2.0
|
||||
// *
|
||||
// * Unless required by applicable law or agreed to in writing, software
|
||||
// * distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// * See the License for the specific language governing permissions and
|
||||
// * limitations under the License.
|
||||
// */
|
||||
//
|
||||
// package com.alibaba.cloud.stream.binder.rocketmq.consuming;
|
||||
//
|
||||
// import java.util.List;
|
||||
// import java.util.Objects;
|
||||
//
|
||||
// import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderUtils;
|
||||
// import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
|
||||
// import
|
||||
/// com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
// import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
// import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
|
||||
// import org.apache.rocketmq.acl.common.AclClientRPCHook;
|
||||
// import org.apache.rocketmq.acl.common.SessionCredentials;
|
||||
// import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
|
||||
// import org.apache.rocketmq.client.consumer.MessageSelector;
|
||||
// import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
|
||||
// import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
|
||||
// import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
|
||||
// import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
|
||||
// import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
|
||||
// import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
|
||||
// import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
|
||||
// import org.apache.rocketmq.client.exception.MQClientException;
|
||||
// import org.apache.rocketmq.common.UtilAll;
|
||||
// import org.apache.rocketmq.common.message.MessageExt;
|
||||
// import org.apache.rocketmq.remoting.RPCHook;
|
||||
// import org.apache.rocketmq.spring.annotation.ConsumeMode;
|
||||
// import org.apache.rocketmq.spring.annotation.MessageModel;
|
||||
// import org.apache.rocketmq.spring.annotation.SelectorType;
|
||||
// import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
// import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
|
||||
// import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
|
||||
// import org.apache.rocketmq.spring.support.RocketMQUtil;
|
||||
// import org.slf4j.Logger;
|
||||
// import org.slf4j.LoggerFactory;
|
||||
//
|
||||
// import org.springframework.beans.factory.InitializingBean;
|
||||
// import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
// import org.springframework.context.SmartLifecycle;
|
||||
// import org.springframework.integration.support.MessageBuilder;
|
||||
// import org.springframework.messaging.Message;
|
||||
// import org.springframework.util.Assert;
|
||||
// import org.springframework.util.StringUtils;
|
||||
//
|
||||
// import static
|
||||
/// com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKETMQ_RECONSUME_TIMES;
|
||||
//
|
||||
/// **
|
||||
// * A class that Listen on rocketmq message.
|
||||
// * <p>
|
||||
// * this class will delegate {@link RocketMQListener} to handle message
|
||||
// *
|
||||
// * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
// * @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
|
||||
// * @see RocketMQListener
|
||||
// */
|
||||
// public class RocketMQListenerBindingContainer
|
||||
// implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
|
||||
//
|
||||
// private final static Logger log = LoggerFactory
|
||||
// .getLogger(RocketMQListenerBindingContainer.class);
|
||||
//
|
||||
// private long suspendCurrentQueueTimeMillis = 1000;
|
||||
//
|
||||
// /**
|
||||
// * Message consume retry strategy<br>
|
||||
// * -1,no retry,put into DLQ directly<br>
|
||||
// * 0,broker control retry frequency<br>
|
||||
// * >0,client control retry frequency.
|
||||
// */
|
||||
// private int delayLevelWhenNextConsume = 0;
|
||||
//
|
||||
// private List<String> nameServer;
|
||||
//
|
||||
// private String consumerGroup;
|
||||
//
|
||||
// private String topic;
|
||||
//
|
||||
// private int consumeThreadMax = 64;
|
||||
//
|
||||
// private String charset = "UTF-8";
|
||||
//
|
||||
// private RocketMQListener rocketMQListener;
|
||||
//
|
||||
// private RocketMQHeaderMapper headerMapper;
|
||||
//
|
||||
// private DefaultMQPushConsumer consumer;
|
||||
//
|
||||
// private boolean running;
|
||||
//
|
||||
// private final ExtendedConsumerProperties<RocketMQConsumerProperties>
|
||||
/// rocketMQConsumerProperties;
|
||||
//
|
||||
// private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
|
||||
//
|
||||
// private final RocketMQBinderConfigurationProperties
|
||||
/// rocketBinderConfigurationProperties;
|
||||
//
|
||||
// // The following properties came from RocketMQConsumerProperties.
|
||||
// private ConsumeMode consumeMode;
|
||||
//
|
||||
// private SelectorType selectorType;
|
||||
//
|
||||
// private String selectorExpression;
|
||||
//
|
||||
// private MessageModel messageModel;
|
||||
//
|
||||
// public RocketMQListenerBindingContainer(
|
||||
// ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
|
||||
// RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
// RocketMQMessageChannelBinder rocketMQMessageChannelBinder) {
|
||||
// this.rocketMQConsumerProperties = rocketMQConsumerProperties;
|
||||
// this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
// this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder;
|
||||
// this.consumeMode = rocketMQConsumerProperties.getExtension().getOrderly()
|
||||
// ? ConsumeMode.ORDERLY : ConsumeMode.CONCURRENTLY;
|
||||
// if (StringUtils.isEmpty(rocketMQConsumerProperties.getExtension().getSql())) {
|
||||
// this.selectorType = SelectorType.TAG;
|
||||
// this.selectorExpression = rocketMQConsumerProperties.getExtension().getTags();
|
||||
// }
|
||||
// else {
|
||||
// this.selectorType = SelectorType.SQL92;
|
||||
// this.selectorExpression = rocketMQConsumerProperties.getExtension().getSql();
|
||||
// }
|
||||
// this.messageModel = rocketMQConsumerProperties.getExtension().getBroadcasting()
|
||||
// ? MessageModel.BROADCASTING : MessageModel.CLUSTERING;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void setupMessageListener(RocketMQListener<?> rocketMQListener) {
|
||||
// this.rocketMQListener = rocketMQListener;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void destroy() throws Exception {
|
||||
// this.setRunning(false);
|
||||
// if (Objects.nonNull(consumer)) {
|
||||
// consumer.shutdown();
|
||||
// }
|
||||
// log.info("container destroyed, {}", this.toString());
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void afterPropertiesSet() throws Exception {
|
||||
// initRocketMQPushConsumer();
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public boolean isAutoStartup() {
|
||||
// return true;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void stop(Runnable callback) {
|
||||
// stop();
|
||||
// callback.run();
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void start() {
|
||||
// if (this.isRunning()) {
|
||||
// throw new IllegalStateException(
|
||||
// "container already running. " + this.toString());
|
||||
// }
|
||||
//
|
||||
// try {
|
||||
// consumer.start();
|
||||
// }
|
||||
// catch (MQClientException e) {
|
||||
// throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
|
||||
// }
|
||||
// this.setRunning(true);
|
||||
//
|
||||
// log.info("running container: {}", this.toString());
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public void stop() {
|
||||
// if (this.isRunning()) {
|
||||
// if (Objects.nonNull(consumer)) {
|
||||
// consumer.shutdown();
|
||||
// }
|
||||
// setRunning(false);
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public boolean isRunning() {
|
||||
// return running;
|
||||
// }
|
||||
//
|
||||
// private void setRunning(boolean running) {
|
||||
// this.running = running;
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public int getPhase() {
|
||||
// return Integer.MAX_VALUE;
|
||||
// }
|
||||
//
|
||||
// private void initRocketMQPushConsumer() throws MQClientException {
|
||||
// Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
|
||||
// Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
|
||||
// Assert.notNull(nameServer, "Property 'nameServer' is required");
|
||||
// Assert.notNull(topic, "Property 'topic' is required");
|
||||
//
|
||||
// String ak = rocketBinderConfigurationProperties.getAccessKey();
|
||||
// String sk = rocketBinderConfigurationProperties.getSecretKey();
|
||||
// if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
|
||||
// RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk));
|
||||
// consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook,
|
||||
// new AllocateMessageQueueAveragely(),
|
||||
// rocketBinderConfigurationProperties.isEnableMsgTrace(),
|
||||
// rocketBinderConfigurationProperties.getCustomizedTraceTopic());
|
||||
// consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook,
|
||||
// topic + "|" + UtilAll.getPid()));
|
||||
// consumer.setVipChannelEnabled(false);
|
||||
// }
|
||||
// else {
|
||||
// consumer = new DefaultMQPushConsumer(consumerGroup,
|
||||
// rocketBinderConfigurationProperties.isEnableMsgTrace(),
|
||||
// rocketBinderConfigurationProperties.getCustomizedTraceTopic());
|
||||
// }
|
||||
//
|
||||
// consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(nameServer));
|
||||
// consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
|
||||
// consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());
|
||||
//
|
||||
// switch (messageModel) {
|
||||
// case BROADCASTING:
|
||||
// consumer.setMessageModel(
|
||||
// org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
|
||||
// break;
|
||||
// case CLUSTERING:
|
||||
// consumer.setMessageModel(
|
||||
// org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
|
||||
// break;
|
||||
// default:
|
||||
// throw new IllegalArgumentException("Property 'messageModel' was wrong.");
|
||||
// }
|
||||
//
|
||||
// switch (selectorType) {
|
||||
// case TAG:
|
||||
// consumer.subscribe(topic, selectorExpression);
|
||||
// break;
|
||||
// case SQL92:
|
||||
// consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
|
||||
// break;
|
||||
// default:
|
||||
// throw new IllegalArgumentException("Property 'selectorType' was wrong.");
|
||||
// }
|
||||
//
|
||||
// switch (consumeMode) {
|
||||
// case ORDERLY:
|
||||
// consumer.setMessageListener(new DefaultMessageListenerOrderly());
|
||||
// break;
|
||||
// case CONCURRENTLY:
|
||||
// consumer.setMessageListener(new DefaultMessageListenerConcurrently());
|
||||
// break;
|
||||
// default:
|
||||
// throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
|
||||
// }
|
||||
//
|
||||
// if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
|
||||
// ((RocketMQPushConsumerLifecycleListener) rocketMQListener)
|
||||
// .prepareStart(consumer);
|
||||
// }
|
||||
//
|
||||
// }
|
||||
//
|
||||
// @Override
|
||||
// public String toString() {
|
||||
// return "RocketMQListenerBindingContainer{" + "consumerGroup='" + consumerGroup
|
||||
// + '\'' + ", nameServer='" + nameServer + '\'' + ", topic='" + topic + '\''
|
||||
// + ", consumeMode=" + consumeMode + ", selectorType=" + selectorType
|
||||
// + ", selectorExpression='" + selectorExpression + '\'' + ", messageModel="
|
||||
// + messageModel + '}';
|
||||
// }
|
||||
//
|
||||
// public long getSuspendCurrentQueueTimeMillis() {
|
||||
// return suspendCurrentQueueTimeMillis;
|
||||
// }
|
||||
//
|
||||
// public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
|
||||
// this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
|
||||
// }
|
||||
//
|
||||
// public int getDelayLevelWhenNextConsume() {
|
||||
// return delayLevelWhenNextConsume;
|
||||
// }
|
||||
//
|
||||
// public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
|
||||
// this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
|
||||
// }
|
||||
//
|
||||
// public List<String> getNameServer() {
|
||||
// return nameServer;
|
||||
// }
|
||||
//
|
||||
// public void setNameServer(List<String> nameServer) {
|
||||
// this.nameServer = nameServer;
|
||||
// }
|
||||
//
|
||||
// public String getConsumerGroup() {
|
||||
// return consumerGroup;
|
||||
// }
|
||||
//
|
||||
// public void setConsumerGroup(String consumerGroup) {
|
||||
// this.consumerGroup = consumerGroup;
|
||||
// }
|
||||
//
|
||||
// public String getTopic() {
|
||||
// return topic;
|
||||
// }
|
||||
//
|
||||
// public void setTopic(String topic) {
|
||||
// this.topic = topic;
|
||||
// }
|
||||
//
|
||||
// public int getConsumeThreadMax() {
|
||||
// return consumeThreadMax;
|
||||
// }
|
||||
//
|
||||
// public void setConsumeThreadMax(int consumeThreadMax) {
|
||||
// this.consumeThreadMax = consumeThreadMax;
|
||||
// }
|
||||
//
|
||||
// public String getCharset() {
|
||||
// return charset;
|
||||
// }
|
||||
//
|
||||
// public void setCharset(String charset) {
|
||||
// this.charset = charset;
|
||||
// }
|
||||
//
|
||||
// public RocketMQListener getRocketMQListener() {
|
||||
// return rocketMQListener;
|
||||
// }
|
||||
//
|
||||
// public void setRocketMQListener(RocketMQListener rocketMQListener) {
|
||||
// this.rocketMQListener = rocketMQListener;
|
||||
// }
|
||||
//
|
||||
// public DefaultMQPushConsumer getConsumer() {
|
||||
// return consumer;
|
||||
// }
|
||||
//
|
||||
// public void setConsumer(DefaultMQPushConsumer consumer) {
|
||||
// this.consumer = consumer;
|
||||
// }
|
||||
//
|
||||
// public ExtendedConsumerProperties<RocketMQConsumerProperties>
|
||||
/// getRocketMQConsumerProperties() {
|
||||
// return rocketMQConsumerProperties;
|
||||
// }
|
||||
//
|
||||
// public ConsumeMode getConsumeMode() {
|
||||
// return consumeMode;
|
||||
// }
|
||||
//
|
||||
// public SelectorType getSelectorType() {
|
||||
// return selectorType;
|
||||
// }
|
||||
//
|
||||
// public String getSelectorExpression() {
|
||||
// return selectorExpression;
|
||||
// }
|
||||
//
|
||||
// public MessageModel getMessageModel() {
|
||||
// return messageModel;
|
||||
// }
|
||||
//
|
||||
// public RocketMQHeaderMapper getHeaderMapper() {
|
||||
// return headerMapper;
|
||||
// }
|
||||
//
|
||||
// public void setHeaderMapper(RocketMQHeaderMapper headerMapper) {
|
||||
// this.headerMapper = headerMapper;
|
||||
// }
|
||||
//
|
||||
// /**
|
||||
// * Convert rocketmq {@link MessageExt} to Spring {@link Message}.
|
||||
// * @param messageExt the rocketmq message
|
||||
// * @return the converted Spring {@link Message}
|
||||
// */
|
||||
// @SuppressWarnings("unchecked")
|
||||
// private Message convertToSpringMessage(MessageExt messageExt) {
|
||||
//
|
||||
// // add reconsume-times header to messageExt
|
||||
// int reconsumeTimes = messageExt.getReconsumeTimes();
|
||||
// messageExt.putUserProperty(ROCKETMQ_RECONSUME_TIMES,
|
||||
// String.valueOf(reconsumeTimes));
|
||||
// Message message = RocketMQUtil.convertToSpringMessage(messageExt);
|
||||
// return MessageBuilder.fromMessage(message)
|
||||
// .copyHeaders(headerMapper.toHeaders(messageExt.getProperties())).build();
|
||||
// }
|
||||
//
|
||||
// public class DefaultMessageListenerConcurrently
|
||||
// implements MessageListenerConcurrently {
|
||||
//
|
||||
// @SuppressWarnings({ "unchecked", "Duplicates" })
|
||||
// @Override
|
||||
// public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
|
||||
// ConsumeConcurrentlyContext context) {
|
||||
// for (MessageExt messageExt : msgs) {
|
||||
// log.debug("received msg: {}", messageExt);
|
||||
// try {
|
||||
// long now = System.currentTimeMillis();
|
||||
// rocketMQListener.onMessage(convertToSpringMessage(messageExt));
|
||||
// long costTime = System.currentTimeMillis() - now;
|
||||
// log.debug("consume {} message key:[{}] cost: {} ms",
|
||||
// messageExt.getMsgId(), messageExt.getKeys(), costTime);
|
||||
// }
|
||||
// catch (Exception e) {
|
||||
// log.warn("consume message failed. messageExt:{}", messageExt, e);
|
||||
// context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
|
||||
// return ConsumeConcurrentlyStatus.RECONSUME_LATER;
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||||
// }
|
||||
//
|
||||
// }
|
||||
//
|
||||
// public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
|
||||
//
|
||||
// @SuppressWarnings({ "unchecked", "Duplicates" })
|
||||
// @Override
|
||||
// public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
|
||||
// ConsumeOrderlyContext context) {
|
||||
// for (MessageExt messageExt : msgs) {
|
||||
// log.debug("received msg: {}", messageExt);
|
||||
// try {
|
||||
// long now = System.currentTimeMillis();
|
||||
// rocketMQListener.onMessage(convertToSpringMessage(messageExt));
|
||||
// long costTime = System.currentTimeMillis() - now;
|
||||
// log.info("consume {} message key:[{}] cost: {} ms",
|
||||
// messageExt.getMsgId(), messageExt.getKeys(), costTime);
|
||||
// }
|
||||
// catch (Exception e) {
|
||||
// log.warn("consume message failed. messageExt:{}", messageExt, e);
|
||||
// context.setSuspendCurrentQueueTimeMillis(
|
||||
// suspendCurrentQueueTimeMillis);
|
||||
// return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
|
||||
// }
|
||||
// }
|
||||
//
|
||||
// return ConsumeOrderlyStatus.SUCCESS;
|
||||
// }
|
||||
//
|
||||
// }
|
||||
//
|
||||
// }
|
||||
|
||||
@ -1,176 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq.integration;
|
||||
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.consuming.RocketMQListenerBindingContainer;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.metrics.Instrumentation;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.integration.endpoint.MessageProducerSupport;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.retry.RecoveryCallback;
|
||||
import org.springframework.retry.RetryCallback;
|
||||
import org.springframework.retry.RetryContext;
|
||||
import org.springframework.retry.RetryListener;
|
||||
import org.springframework.retry.support.RetryTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
private static final Logger log = LoggerFactory
|
||||
.getLogger(RocketMQInboundChannelAdapter.class);
|
||||
|
||||
private RetryTemplate retryTemplate;
|
||||
|
||||
private RecoveryCallback<? extends Object> recoveryCallback;
|
||||
|
||||
private RocketMQListenerBindingContainer rocketMQListenerContainer;
|
||||
|
||||
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
public RocketMQInboundChannelAdapter(
|
||||
RocketMQListenerBindingContainer rocketMQListenerContainer,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
this.rocketMQListenerContainer = rocketMQListenerContainer;
|
||||
this.consumerProperties = consumerProperties;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void onInit() {
|
||||
if (consumerProperties == null
|
||||
|| !consumerProperties.getExtension().getEnabled()) {
|
||||
return;
|
||||
}
|
||||
super.onInit();
|
||||
if (this.retryTemplate != null) {
|
||||
Assert.state(getErrorChannel() == null,
|
||||
"Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
|
||||
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
|
||||
+ "send an error message when retries are exhausted");
|
||||
}
|
||||
|
||||
BindingRocketMQListener listener = new BindingRocketMQListener();
|
||||
rocketMQListenerContainer.setRocketMQListener(listener);
|
||||
|
||||
if (retryTemplate != null) {
|
||||
this.retryTemplate.registerListener(listener);
|
||||
}
|
||||
|
||||
try {
|
||||
rocketMQListenerContainer.afterPropertiesSet();
|
||||
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
|
||||
throw new IllegalArgumentException(
|
||||
"rocketMQListenerContainer init error: " + e.getMessage(), e);
|
||||
}
|
||||
|
||||
instrumentationManager.addHealthInstrumentation(
|
||||
new Instrumentation(rocketMQListenerContainer.getTopic()
|
||||
+ rocketMQListenerContainer.getConsumerGroup()));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
if (consumerProperties == null
|
||||
|| !consumerProperties.getExtension().getEnabled()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
rocketMQListenerContainer.start();
|
||||
instrumentationManager
|
||||
.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
|
||||
+ rocketMQListenerContainer.getConsumerGroup())
|
||||
.markStartedSuccessfully();
|
||||
}
|
||||
catch (Exception e) {
|
||||
instrumentationManager
|
||||
.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
|
||||
+ rocketMQListenerContainer.getConsumerGroup())
|
||||
.markStartFailed(e);
|
||||
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
|
||||
throw new MessagingException(MessageBuilder.withPayload(
|
||||
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
|
||||
.build(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
rocketMQListenerContainer.stop();
|
||||
}
|
||||
|
||||
public void setRetryTemplate(RetryTemplate retryTemplate) {
|
||||
this.retryTemplate = retryTemplate;
|
||||
}
|
||||
|
||||
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
|
||||
this.recoveryCallback = recoveryCallback;
|
||||
}
|
||||
|
||||
protected class BindingRocketMQListener
|
||||
implements RocketMQListener<Message>, RetryListener {
|
||||
|
||||
@Override
|
||||
public void onMessage(Message message) {
|
||||
boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
|
||||
if (enableRetry) {
|
||||
RocketMQInboundChannelAdapter.this.retryTemplate.execute(context -> {
|
||||
RocketMQInboundChannelAdapter.this.sendMessage(message);
|
||||
return null;
|
||||
}, (RecoveryCallback<Object>) RocketMQInboundChannelAdapter.this.recoveryCallback);
|
||||
}
|
||||
else {
|
||||
RocketMQInboundChannelAdapter.this.sendMessage(message);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> boolean open(RetryContext context,
|
||||
RetryCallback<T, E> callback) {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> void close(RetryContext context,
|
||||
RetryCallback<T, E> callback, Throwable throwable) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> void onError(RetryContext context,
|
||||
RetryCallback<T, E> callback, Throwable throwable) {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,302 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq.integration;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.metrics.Instrumentation;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.support.RocketMQHeaderMapper;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.client.producer.SendCallback;
|
||||
import org.apache.rocketmq.client.producer.SendResult;
|
||||
import org.apache.rocketmq.client.producer.SendStatus;
|
||||
import org.apache.rocketmq.common.message.MessageConst;
|
||||
import org.apache.rocketmq.common.message.MessageQueue;
|
||||
import org.apache.rocketmq.spring.core.RocketMQTemplate;
|
||||
import org.apache.rocketmq.spring.support.RocketMQHeaders;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.cloud.stream.binder.BinderHeaders;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.integration.handler.AbstractMessageHandler;
|
||||
import org.springframework.integration.support.DefaultErrorMessageStrategy;
|
||||
import org.springframework.integration.support.ErrorMessageStrategy;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.support.ErrorMessage;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
|
||||
|
||||
private final static Logger log = LoggerFactory
|
||||
.getLogger(RocketMQMessageHandler.class);
|
||||
|
||||
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
|
||||
|
||||
private MessageChannel sendFailureChannel;
|
||||
|
||||
private final RocketMQTemplate rocketMQTemplate;
|
||||
|
||||
private RocketMQHeaderMapper headerMapper;
|
||||
|
||||
private final Boolean transactional;
|
||||
|
||||
private final String destination;
|
||||
|
||||
private final String groupName;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
private boolean sync = false;
|
||||
|
||||
private volatile boolean running = false;
|
||||
|
||||
private ExtendedProducerProperties<RocketMQProducerProperties> producerProperties;
|
||||
|
||||
private MessageConverterConfigurer.PartitioningInterceptor partitioningInterceptor;
|
||||
|
||||
public RocketMQMessageHandler(RocketMQTemplate rocketMQTemplate, String destination,
|
||||
String groupName, Boolean transactional,
|
||||
InstrumentationManager instrumentationManager,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
|
||||
MessageConverterConfigurer.PartitioningInterceptor partitioningInterceptor) {
|
||||
this.rocketMQTemplate = rocketMQTemplate;
|
||||
this.destination = destination;
|
||||
this.groupName = groupName;
|
||||
this.transactional = transactional;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
this.producerProperties = producerProperties;
|
||||
this.partitioningInterceptor = partitioningInterceptor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
if (!transactional) {
|
||||
instrumentationManager
|
||||
.addHealthInstrumentation(new Instrumentation(destination));
|
||||
try {
|
||||
rocketMQTemplate.afterPropertiesSet();
|
||||
instrumentationManager.getHealthInstrumentation(destination)
|
||||
.markStartedSuccessfully();
|
||||
}
|
||||
catch (Exception e) {
|
||||
instrumentationManager.getHealthInstrumentation(destination)
|
||||
.markStartFailed(e);
|
||||
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
|
||||
throw new MessagingException(MessageBuilder.withPayload(
|
||||
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
|
||||
.build(), e);
|
||||
}
|
||||
}
|
||||
if (producerProperties.isPartitioned()) {
|
||||
try {
|
||||
List<MessageQueue> messageQueues = rocketMQTemplate.getProducer()
|
||||
.fetchPublishMessageQueues(destination);
|
||||
if (producerProperties.getPartitionCount() != messageQueues.size()) {
|
||||
logger.info(String.format(
|
||||
"The partition count of topic '%s' will change from '%s' to '%s'",
|
||||
destination, producerProperties.getPartitionCount(),
|
||||
messageQueues.size()));
|
||||
producerProperties.setPartitionCount(messageQueues.size());
|
||||
partitioningInterceptor
|
||||
.setPartitionCount(producerProperties.getPartitionCount());
|
||||
}
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
logger.error("fetch publish message queues fail", e);
|
||||
}
|
||||
}
|
||||
running = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (!transactional) {
|
||||
rocketMQTemplate.destroy();
|
||||
}
|
||||
running = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleMessageInternal(
|
||||
org.springframework.messaging.Message<?> message) {
|
||||
try {
|
||||
// issue 737 fix
|
||||
Map<String, String> jsonHeaders = headerMapper
|
||||
.fromHeaders(message.getHeaders());
|
||||
message = org.springframework.messaging.support.MessageBuilder
|
||||
.fromMessage(message).copyHeaders(jsonHeaders).build();
|
||||
|
||||
final StringBuilder topicWithTags = new StringBuilder(destination);
|
||||
String tags = Optional
|
||||
.ofNullable(message.getHeaders().get(RocketMQHeaders.TAGS)).orElse("")
|
||||
.toString();
|
||||
if (!StringUtils.isEmpty(tags)) {
|
||||
topicWithTags.append(":").append(tags);
|
||||
}
|
||||
|
||||
SendResult sendRes = null;
|
||||
if (transactional) {
|
||||
sendRes = rocketMQTemplate.sendMessageInTransaction(groupName,
|
||||
topicWithTags.toString(), message, message.getHeaders()
|
||||
.get(RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG));
|
||||
log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
|
||||
}
|
||||
else {
|
||||
int delayLevel = 0;
|
||||
try {
|
||||
Object delayLevelObj = message.getHeaders()
|
||||
.getOrDefault(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 0);
|
||||
if (delayLevelObj instanceof Number) {
|
||||
delayLevel = ((Number) delayLevelObj).intValue();
|
||||
}
|
||||
else if (delayLevelObj instanceof String) {
|
||||
delayLevel = Integer.parseInt((String) delayLevelObj);
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
boolean needSelectQueue = message.getHeaders()
|
||||
.containsKey(BinderHeaders.PARTITION_HEADER);
|
||||
if (sync) {
|
||||
if (needSelectQueue) {
|
||||
sendRes = rocketMQTemplate.syncSendOrderly(
|
||||
topicWithTags.toString(), message, "",
|
||||
rocketMQTemplate.getProducer().getSendMsgTimeout());
|
||||
}
|
||||
else {
|
||||
sendRes = rocketMQTemplate.syncSend(topicWithTags.toString(),
|
||||
message,
|
||||
rocketMQTemplate.getProducer().getSendMsgTimeout(),
|
||||
delayLevel);
|
||||
}
|
||||
log.debug("sync send to topic " + topicWithTags + " " + sendRes);
|
||||
}
|
||||
else {
|
||||
Message<?> finalMessage = message;
|
||||
SendCallback sendCallback = new SendCallback() {
|
||||
@Override
|
||||
public void onSuccess(SendResult sendResult) {
|
||||
log.debug("async send to topic " + topicWithTags + " "
|
||||
+ sendResult);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onException(Throwable e) {
|
||||
log.error("RocketMQ Message hasn't been sent. Caused by "
|
||||
+ e.getMessage());
|
||||
if (getSendFailureChannel() != null) {
|
||||
getSendFailureChannel().send(
|
||||
RocketMQMessageHandler.this.errorMessageStrategy
|
||||
.buildErrorMessage(new MessagingException(
|
||||
finalMessage, e), null));
|
||||
}
|
||||
}
|
||||
};
|
||||
if (needSelectQueue) {
|
||||
rocketMQTemplate.asyncSendOrderly(topicWithTags.toString(),
|
||||
message, "", sendCallback,
|
||||
rocketMQTemplate.getProducer().getSendMsgTimeout());
|
||||
}
|
||||
else {
|
||||
rocketMQTemplate.asyncSend(topicWithTags.toString(), message,
|
||||
sendCallback);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (sendRes != null && !sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
|
||||
if (getSendFailureChannel() != null) {
|
||||
this.getSendFailureChannel().send(message);
|
||||
}
|
||||
else {
|
||||
throw new MessagingException(message,
|
||||
new MQClientException("message hasn't been sent", null));
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
|
||||
if (getSendFailureChannel() != null) {
|
||||
getSendFailureChannel().send(this.errorMessageStrategy
|
||||
.buildErrorMessage(new MessagingException(message, e), null));
|
||||
}
|
||||
else {
|
||||
throw new MessagingException(message, e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent
|
||||
* to this channel with a payload of a {@link MessagingException} with the failed
|
||||
* message and cause.
|
||||
* @param sendFailureChannel the failure channel.
|
||||
* @since 0.2.2
|
||||
*/
|
||||
public void setSendFailureChannel(MessageChannel sendFailureChannel) {
|
||||
this.sendFailureChannel = sendFailureChannel;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the error message strategy implementation to use when sending error messages
|
||||
* after send failures. Cannot be null.
|
||||
* @param errorMessageStrategy the implementation.
|
||||
* @since 0.2.2
|
||||
*/
|
||||
public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
|
||||
Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null");
|
||||
this.errorMessageStrategy = errorMessageStrategy;
|
||||
}
|
||||
|
||||
public MessageChannel getSendFailureChannel() {
|
||||
return sendFailureChannel;
|
||||
}
|
||||
|
||||
public void setSync(boolean sync) {
|
||||
this.sync = sync;
|
||||
}
|
||||
|
||||
public RocketMQHeaderMapper getHeaderMapper() {
|
||||
return headerMapper;
|
||||
}
|
||||
|
||||
public void setHeaderMapper(RocketMQHeaderMapper headerMapper) {
|
||||
this.headerMapper = headerMapper;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,382 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq.integration;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderUtils;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.consuming.RocketMQMessageQueueChooser;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
|
||||
import org.apache.rocketmq.client.consumer.MessageQueueListener;
|
||||
import org.apache.rocketmq.client.consumer.MessageSelector;
|
||||
import org.apache.rocketmq.client.consumer.PullResult;
|
||||
import org.apache.rocketmq.client.consumer.PullStatus;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.apache.rocketmq.common.message.MessageQueue;
|
||||
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
|
||||
import org.apache.rocketmq.spring.support.RocketMQUtil;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.integration.IntegrationMessageHeaderAccessor;
|
||||
import org.springframework.integration.acks.AcknowledgmentCallback;
|
||||
import org.springframework.integration.acks.AcknowledgmentCallbackFactory;
|
||||
import org.springframework.integration.endpoint.AbstractMessageSource;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageSource extends AbstractMessageSource<Object>
|
||||
implements DisposableBean, Lifecycle {
|
||||
|
||||
private final static Logger log = LoggerFactory
|
||||
.getLogger(RocketMQMessageSource.class);
|
||||
|
||||
private final RocketMQCallbackFactory ackCallbackFactory;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties;
|
||||
|
||||
private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
|
||||
|
||||
private final String topic;
|
||||
|
||||
private final String group;
|
||||
|
||||
private final Object consumerMonitor = new Object();
|
||||
|
||||
private DefaultMQPullConsumer consumer;
|
||||
|
||||
private boolean running;
|
||||
|
||||
private MessageSelector messageSelector;
|
||||
|
||||
private RocketMQMessageQueueChooser messageQueueChooser = new RocketMQMessageQueueChooser();
|
||||
|
||||
public RocketMQMessageSource(
|
||||
RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
|
||||
String topic, String group) {
|
||||
this(new RocketMQCallbackFactory(), rocketMQBinderConfigurationProperties,
|
||||
rocketMQConsumerProperties, topic, group);
|
||||
}
|
||||
|
||||
public RocketMQMessageSource(RocketMQCallbackFactory ackCallbackFactory,
|
||||
RocketMQBinderConfigurationProperties rocketMQBinderConfigurationProperties,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
|
||||
String topic, String group) {
|
||||
this.ackCallbackFactory = ackCallbackFactory;
|
||||
this.rocketMQBinderConfigurationProperties = rocketMQBinderConfigurationProperties;
|
||||
this.rocketMQConsumerProperties = rocketMQConsumerProperties;
|
||||
this.topic = topic;
|
||||
this.group = group;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void start() {
|
||||
if (this.isRunning()) {
|
||||
throw new IllegalStateException(
|
||||
"pull consumer already running. " + this.toString());
|
||||
}
|
||||
try {
|
||||
consumer = new DefaultMQPullConsumer(group);
|
||||
consumer.setNamesrvAddr(RocketMQBinderUtils.getNameServerStr(
|
||||
rocketMQBinderConfigurationProperties.getNameServer()));
|
||||
consumer.setConsumerPullTimeoutMillis(
|
||||
rocketMQConsumerProperties.getExtension().getPullTimeout());
|
||||
consumer.setMessageModel(MessageModel.CLUSTERING);
|
||||
|
||||
String tags = rocketMQConsumerProperties.getExtension().getTags();
|
||||
String sql = rocketMQConsumerProperties.getExtension().getSql();
|
||||
|
||||
if (!StringUtils.isEmpty(tags) && !StringUtils.isEmpty(sql)) {
|
||||
messageSelector = MessageSelector.byTag(tags);
|
||||
}
|
||||
else if (!StringUtils.isEmpty(tags)) {
|
||||
messageSelector = MessageSelector.byTag(tags);
|
||||
}
|
||||
else if (!StringUtils.isEmpty(sql)) {
|
||||
messageSelector = MessageSelector.bySql(sql);
|
||||
}
|
||||
|
||||
consumer.registerMessageQueueListener(topic, new MessageQueueListener() {
|
||||
@Override
|
||||
public void messageQueueChanged(String topic, Set<MessageQueue> mqAll,
|
||||
Set<MessageQueue> mqDivided) {
|
||||
log.info(
|
||||
"messageQueueChanged, topic='{}', mqAll=`{}`, mqDivided=`{}`",
|
||||
topic, mqAll, mqDivided);
|
||||
switch (consumer.getMessageModel()) {
|
||||
case BROADCASTING:
|
||||
RocketMQMessageSource.this.resetMessageQueues(mqAll);
|
||||
break;
|
||||
case CLUSTERING:
|
||||
RocketMQMessageSource.this.resetMessageQueues(mqDivided);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
consumer.start();
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
log.error("DefaultMQPullConsumer startup error: " + e.getMessage(), e);
|
||||
}
|
||||
this.setRunning(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void stop() {
|
||||
if (this.isRunning()) {
|
||||
this.setRunning(false);
|
||||
consumer.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected synchronized Object doReceive() {
|
||||
if (messageQueueChooser.getMessageQueues() == null
|
||||
|| messageQueueChooser.getMessageQueues().size() == 0) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
int count = 0;
|
||||
while (count < messageQueueChooser.getMessageQueues().size()) {
|
||||
MessageQueue messageQueue;
|
||||
synchronized (this.consumerMonitor) {
|
||||
messageQueue = messageQueueChooser.choose();
|
||||
messageQueueChooser.increment();
|
||||
}
|
||||
|
||||
long offset = consumer.fetchConsumeOffset(messageQueue,
|
||||
rocketMQConsumerProperties.getExtension().isFromStore());
|
||||
|
||||
log.debug("topic='{}', group='{}', messageQueue='{}', offset now='{}'",
|
||||
this.topic, this.group, messageQueue, offset);
|
||||
|
||||
PullResult pullResult;
|
||||
if (messageSelector != null) {
|
||||
pullResult = consumer.pull(messageQueue, messageSelector, offset, 1);
|
||||
}
|
||||
else {
|
||||
pullResult = consumer.pull(messageQueue, (String) null, offset, 1);
|
||||
}
|
||||
|
||||
if (pullResult.getPullStatus() == PullStatus.FOUND) {
|
||||
List<MessageExt> messageExtList = pullResult.getMsgFoundList();
|
||||
|
||||
Message message = RocketMQUtil
|
||||
.convertToSpringMessage(messageExtList.get(0));
|
||||
|
||||
AcknowledgmentCallback ackCallback = this.ackCallbackFactory
|
||||
.createCallback(new RocketMQAckInfo(messageQueue, pullResult,
|
||||
consumer, offset));
|
||||
|
||||
Message messageResult = MessageBuilder.fromMessage(message).setHeader(
|
||||
IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK,
|
||||
ackCallback).build();
|
||||
return messageResult;
|
||||
}
|
||||
else {
|
||||
log.debug("messageQueue='{}' PullResult='{}' with topic `{}`",
|
||||
messageQueueChooser.getMessageQueues(),
|
||||
pullResult.getPullStatus(), topic);
|
||||
}
|
||||
count++;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
log.error("Consumer pull error: " + e.getMessage(), e);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getComponentType() {
|
||||
return "rocketmq:message-source";
|
||||
}
|
||||
|
||||
public synchronized void setRunning(boolean running) {
|
||||
this.running = running;
|
||||
}
|
||||
|
||||
public synchronized void resetMessageQueues(Set<MessageQueue> queueSet) {
|
||||
log.info("resetMessageQueues, topic='{}', messageQueue=`{}`", topic, queueSet);
|
||||
synchronized (this.consumerMonitor) {
|
||||
this.messageQueueChooser.reset(queueSet);
|
||||
}
|
||||
}
|
||||
|
||||
public static class RocketMQCallbackFactory
|
||||
implements AcknowledgmentCallbackFactory<RocketMQAckInfo> {
|
||||
|
||||
@Override
|
||||
public AcknowledgmentCallback createCallback(RocketMQAckInfo info) {
|
||||
return new RocketMQAckCallback(info);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class RocketMQAckCallback implements AcknowledgmentCallback {
|
||||
|
||||
private final RocketMQAckInfo ackInfo;
|
||||
|
||||
private boolean acknowledged;
|
||||
|
||||
private boolean autoAckEnabled = true;
|
||||
|
||||
public RocketMQAckCallback(RocketMQAckInfo ackInfo) {
|
||||
this.ackInfo = ackInfo;
|
||||
}
|
||||
|
||||
protected void setAcknowledged(boolean acknowledged) {
|
||||
this.acknowledged = acknowledged;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAcknowledged() {
|
||||
return this.acknowledged;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void noAutoAck() {
|
||||
this.autoAckEnabled = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAutoAck() {
|
||||
return this.autoAckEnabled;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void acknowledge(Status status) {
|
||||
Assert.notNull(status, "'status' cannot be null");
|
||||
if (this.acknowledged) {
|
||||
throw new IllegalStateException("Already acknowledged");
|
||||
}
|
||||
log.debug("acknowledge(" + status.name() + ") for " + this);
|
||||
synchronized (this.ackInfo.getConsumerMonitor()) {
|
||||
try {
|
||||
switch (status) {
|
||||
case ACCEPT:
|
||||
case REJECT:
|
||||
ackInfo.getConsumer().updateConsumeOffset(
|
||||
ackInfo.getMessageQueue(),
|
||||
ackInfo.getPullResult().getNextBeginOffset());
|
||||
log.debug("messageQueue='{}' offset update to `{}`",
|
||||
ackInfo.getMessageQueue(), String.valueOf(
|
||||
ackInfo.getPullResult().getNextBeginOffset()));
|
||||
break;
|
||||
case REQUEUE:
|
||||
// decrease index and update offset of messageQueue of ackInfo
|
||||
int oldIndex = ackInfo.getMessageQueueChooser().requeue();
|
||||
ackInfo.getConsumer().updateConsumeOffset(
|
||||
ackInfo.getMessageQueue(), ackInfo.getOldOffset());
|
||||
log.debug(
|
||||
"messageQueue='{}' offset requeue to index:`{}`, oldOffset:'{}'",
|
||||
ackInfo.getMessageQueue(), oldIndex,
|
||||
ackInfo.getOldOffset());
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (MQClientException e) {
|
||||
log.error("acknowledge error: " + e.getErrorMessage(), e);
|
||||
}
|
||||
finally {
|
||||
this.acknowledged = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "RocketMQAckCallback{" + "ackInfo=" + ackInfo + ", acknowledged="
|
||||
+ acknowledged + ", autoAckEnabled=" + autoAckEnabled + '}';
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class RocketMQAckInfo {
|
||||
|
||||
private final MessageQueue messageQueue;
|
||||
|
||||
private final PullResult pullResult;
|
||||
|
||||
private final DefaultMQPullConsumer consumer;
|
||||
|
||||
private final long oldOffset;
|
||||
|
||||
public RocketMQAckInfo(MessageQueue messageQueue, PullResult pullResult,
|
||||
DefaultMQPullConsumer consumer, long oldOffset) {
|
||||
this.messageQueue = messageQueue;
|
||||
this.pullResult = pullResult;
|
||||
this.consumer = consumer;
|
||||
this.oldOffset = oldOffset;
|
||||
}
|
||||
|
||||
public MessageQueue getMessageQueue() {
|
||||
return messageQueue;
|
||||
}
|
||||
|
||||
public PullResult getPullResult() {
|
||||
return pullResult;
|
||||
}
|
||||
|
||||
public DefaultMQPullConsumer getConsumer() {
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public RocketMQMessageQueueChooser getMessageQueueChooser() {
|
||||
return RocketMQMessageSource.this.messageQueueChooser;
|
||||
}
|
||||
|
||||
public long getOldOffset() {
|
||||
return oldOffset;
|
||||
}
|
||||
|
||||
public Object getConsumerMonitor() {
|
||||
return RocketMQMessageSource.this.consumerMonitor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "RocketMQAckInfo{" + "messageQueue=" + messageQueue + ", pullResult="
|
||||
+ pullResult + ", consumer=" + consumer + ", oldOffset=" + oldOffset
|
||||
+ '}';
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,49 +0,0 @@
|
||||
/*
|
||||
* Copyright 2013-2018 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.alibaba.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQBindingProperties implements BinderSpecificPropertiesProvider {
|
||||
|
||||
private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties();
|
||||
|
||||
private RocketMQProducerProperties producer = new RocketMQProducerProperties();
|
||||
|
||||
@Override
|
||||
public RocketMQConsumerProperties getConsumer() {
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public void setConsumer(RocketMQConsumerProperties consumer) {
|
||||
this.consumer = consumer;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQProducerProperties getProducer() {
|
||||
return producer;
|
||||
}
|
||||
|
||||
public void setProducer(RocketMQProducerProperties producer) {
|
||||
this.producer = producer;
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue