study/Spring-Cloud-Alibaba
1
0
Fork 0
mirror of https://gitee.com/mirrors/Spring-Cloud-Alibaba.git synced 2021-06-26 13:25:11 +08:00
Code Issues Releases Wiki Activity

using eclipse code formatter to format code

Browse Source
This commit is contained in:
fangjian0423 2018-11-20 17:17:44 +08:00
parent c34ecfffd8
commit bfae54b51e
22 changed files with 946 additions and 852 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
spring-cloud-alibaba-examples/rocketmq-example/pom.xml
View File

@ -6,7 +6,6 @@
<groupId>org.springframework.cloud</groupId> <groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-alibaba-examples</artifactId> <artifactId>spring-cloud-alibaba-examples</artifactId>
<version>0.2.1.BUILD-SNAPSHOT</version> <version>0.2.1.BUILD-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent> </parent>
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>

132
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java
View File

@ -26,76 +26,84 @@ import org.springframework.messaging.MessageHandler;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class RocketMQMessageChannelBinder extends public class RocketMQMessageChannelBinder extends
AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>, AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner>
ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner> implements
implements ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> { ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
private static final Logger logger = LoggerFactory.getLogger(RocketMQMessageChannelBinder.class); private static final Logger logger = LoggerFactory
.getLogger(RocketMQMessageChannelBinder.class);
private final RocketMQExtendedBindingProperties extendedBindingProperties; private final RocketMQExtendedBindingProperties extendedBindingProperties;
private final RocketMQTopicProvisioner rocketTopicProvisioner; private final RocketMQTopicProvisioner rocketTopicProvisioner;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final InstrumentationManager instrumentationManager; private final InstrumentationManager instrumentationManager;
private final ConsumersManager consumersManager; private final ConsumersManager consumersManager;
public RocketMQMessageChannelBinder(ConsumersManager consumersManager, public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
RocketMQExtendedBindingProperties extendedBindingProperties, RocketMQExtendedBindingProperties extendedBindingProperties,
RocketMQTopicProvisioner provisioningProvider, RocketMQTopicProvisioner provisioningProvider,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties, RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
InstrumentationManager instrumentationManager) { InstrumentationManager instrumentationManager) {
super(null, provisioningProvider); super(null, provisioningProvider);
this.consumersManager = consumersManager; this.consumersManager = consumersManager;
this.extendedBindingProperties = extendedBindingProperties; this.extendedBindingProperties = extendedBindingProperties;
this.rocketTopicProvisioner = provisioningProvider; this.rocketTopicProvisioner = provisioningProvider;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties; this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.instrumentationManager = instrumentationManager; this.instrumentationManager = instrumentationManager;
} }
@Override @Override
protected MessageHandler createProducerMessageHandler(ProducerDestination destination, protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
ExtendedProducerProperties<RocketMQProducerProperties> ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
producerProperties, MessageChannel errorChannel) throws Exception {
MessageChannel errorChannel) throws Exception { if (producerProperties.getExtension().getEnabled()) {
if (producerProperties.getExtension().getEnabled()) { return new RocketMQMessageHandler(destination.getName(),
return new RocketMQMessageHandler(destination.getName(), producerProperties.getExtension(), producerProperties.getExtension(),
rocketBinderConfigurationProperties, instrumentationManager); rocketBinderConfigurationProperties, instrumentationManager);
} else { }
throw new RuntimeException( else {
"Binding for channel " + destination.getName() + "has been disabled, message can't be delivered"); throw new RuntimeException("Binding for channel " + destination.getName()
} + "has been disabled, message can't be delivered");
} }
}
@Override @Override
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, protected MessageProducer createConsumerEndpoint(ConsumerDestination destination,
ExtendedConsumerProperties<RocketMQConsumerProperties> String group,
consumerProperties) ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties)
throws Exception { throws Exception {
if (group == null || "".equals(group)) { if (group == null || "".equals(group)) {
throw new RuntimeException("'group' must be configured for channel + " + destination.getName()); throw new RuntimeException(
} "'group' must be configured for channel + " + destination.getName());
}
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(consumersManager, RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
consumerProperties, destination.getName(), group, instrumentationManager); consumersManager, consumerProperties, destination.getName(), group,
instrumentationManager);
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, group, ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
consumerProperties); group, consumerProperties);
if (consumerProperties.getMaxAttempts() > 1) { if (consumerProperties.getMaxAttempts() > 1) {
rocketInboundChannelAdapter.setRetryTemplate(buildRetryTemplate(consumerProperties)); rocketInboundChannelAdapter
rocketInboundChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); .setRetryTemplate(buildRetryTemplate(consumerProperties));
} else { rocketInboundChannelAdapter
rocketInboundChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel()); .setRecoveryCallback(errorInfrastructure.getRecoverer());
} }
else {
rocketInboundChannelAdapter
.setErrorChannel(errorInfrastructure.getErrorChannel());
}
return rocketInboundChannelAdapter; return rocketInboundChannelAdapter;
} }
@Override @Override
public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) { public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
return extendedBindingProperties.getExtendedConsumerProperties(channelName); return extendedBindingProperties.getExtendedConsumerProperties(channelName);
} }
@Override @Override
public RocketMQProducerProperties getExtendedProducerProperties(String channelName) { public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
return extendedBindingProperties.getExtendedProducerProperties(channelName); return extendedBindingProperties.getExtendedProducerProperties(channelName);
} }
} }

141
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageHeaderAccessor.java
View File

@ -1,5 +1,10 @@
package org.springframework.cloud.stream.binder.rocketmq; package org.springframework.cloud.stream.binder.rocketmq;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
@ -12,95 +17,93 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.messaging.support.MessageHeaderAccessor;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
/** /**
* @author Timur Valiev * @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class RocketMQMessageHeaderAccessor extends MessageHeaderAccessor { public class RocketMQMessageHeaderAccessor extends MessageHeaderAccessor {
public RocketMQMessageHeaderAccessor() { public RocketMQMessageHeaderAccessor() {
super(); super();
} }
public RocketMQMessageHeaderAccessor(Message<?> message) { public RocketMQMessageHeaderAccessor(Message<?> message) {
super(message); super(message);
} }
public Acknowledgement getAcknowledgement(Message message) { public Acknowledgement getAcknowledgement(Message message) {
return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class); return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
} }
public RocketMQMessageHeaderAccessor withAcknowledgment(Acknowledgement acknowledgment) { public RocketMQMessageHeaderAccessor withAcknowledgment(
setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment); Acknowledgement acknowledgment) {
return this; setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
} return this;
}
public String getTags() { public String getTags() {
return (String)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_TAGS, ""); return (String) getMessageHeaders().getOrDefault(MessageConst.PROPERTY_TAGS, "");
} }
public RocketMQMessageHeaderAccessor withTags(String tag) { public RocketMQMessageHeaderAccessor withTags(String tag) {
setHeader(MessageConst.PROPERTY_TAGS, tag); setHeader(MessageConst.PROPERTY_TAGS, tag);
return this; return this;
} }
public String getKeys() { public String getKeys() {
return (String)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_KEYS, ""); return (String) getMessageHeaders().getOrDefault(MessageConst.PROPERTY_KEYS, "");
} }
public RocketMQMessageHeaderAccessor withKeys(String keys) { public RocketMQMessageHeaderAccessor withKeys(String keys) {
setHeader(MessageConst.PROPERTY_KEYS, keys); setHeader(MessageConst.PROPERTY_KEYS, keys);
return this; return this;
} }
public MessageExt getRocketMessage() { public MessageExt getRocketMessage() {
return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class); return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
} }
public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) { public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
setHeader(ORIGINAL_ROCKET_MESSAGE, message); setHeader(ORIGINAL_ROCKET_MESSAGE, message);
return this; return this;
} }
public Integer getDelayTimeLevel() { public Integer getDelayTimeLevel() {
return (Integer)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 0); return (Integer) getMessageHeaders()
} .getOrDefault(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 0);
}
public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) { public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel); setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
return this; return this;
} }
public Integer getFlag() { public Integer getFlag() {
return (Integer)getMessageHeaders().getOrDefault(ROCKET_FLAG, 0); return (Integer) getMessageHeaders().getOrDefault(ROCKET_FLAG, 0);
} }
public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) { public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
setHeader(ROCKET_FLAG, delayTimeLevel); setHeader(ROCKET_FLAG, delayTimeLevel);
return this; return this;
} }
public SendResult getSendResult() { public SendResult getSendResult() {
return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class); return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
} }
public static void putSendResult(MutableMessage message, SendResult sendResult) { public static void putSendResult(MutableMessage message, SendResult sendResult) {
message.getHeaders().put(ROCKET_SEND_RESULT, sendResult); message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
} }
public Map<String, String> getUserProperties() { public Map<String, String> getUserProperties() {
Map<String, String> result = new HashMap<>(); Map<String, String> result = new HashMap<>();
for (Map.Entry<String, Object> entry : this.toMap().entrySet()) { for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
if (entry.getValue() instanceof String && !MessageConst.STRING_HASH_SET.contains(entry.getKey()) && !entry if (entry.getValue() instanceof String
.getKey().equals(MessageHeaders.CONTENT_TYPE)) { && !MessageConst.STRING_HASH_SET.contains(entry.getKey())
result.put(entry.getKey(), (String)entry.getValue()); && !entry.getKey().equals(MessageHeaders.CONTENT_TYPE)) {
} result.put(entry.getKey(), (String) entry.getValue());
} }
return result; }
} return result;
}
} }

35
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderEndpoint.java
View File

@ -1,14 +1,15 @@
package org.springframework.cloud.stream.binder.rocketmq.actuator; package org.springframework.cloud.stream.binder.rocketmq.actuator;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ENDPOINT_ID;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import com.codahale.metrics.MetricRegistry;
import org.springframework.boot.actuate.endpoint.annotation.Endpoint; import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
import org.springframework.boot.actuate.endpoint.annotation.ReadOperation; import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ENDPOINT_ID; import com.codahale.metrics.MetricRegistry;
/** /**
* @author Timur Valiev * @author Timur Valiev
@ -17,23 +18,23 @@ import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderCon
@Endpoint(id = ENDPOINT_ID) @Endpoint(id = ENDPOINT_ID)
public class RocketMQBinderEndpoint { public class RocketMQBinderEndpoint {
private MetricRegistry metricRegistry = new MetricRegistry(); private MetricRegistry metricRegistry = new MetricRegistry();
private Map<String, Object> runtime = new ConcurrentHashMap<>(); private Map<String, Object> runtime = new ConcurrentHashMap<>();
@ReadOperation @ReadOperation
public Map<String, Object> invoke() { public Map<String, Object> invoke() {
Map<String, Object> result = new HashMap<>(); Map<String, Object> result = new HashMap<>();
result.put("metrics", metricRegistry().getMetrics()); result.put("metrics", metricRegistry().getMetrics());
result.put("runtime", runtime()); result.put("runtime", runtime());
return result; return result;
} }
public MetricRegistry metricRegistry() { public MetricRegistry metricRegistry() {
return metricRegistry; return metricRegistry;
} }
public Map<String, Object> runtime() { public Map<String, Object> runtime() {
return runtime; return runtime;
} }
} }

43
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/actuator/RocketMQBinderHealthIndicator.java
View File

@ -11,27 +11,28 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
*/ */
public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator { public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
private final InstrumentationManager instrumentationManager; private final InstrumentationManager instrumentationManager;
public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) { public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
this.instrumentationManager = instrumentationManager; this.instrumentationManager = instrumentationManager;
} }
@Override @Override
protected void doHealthCheck(Health.Builder builder) throws Exception { protected void doHealthCheck(Health.Builder builder) throws Exception {
if (instrumentationManager.getHealthInstrumentations().stream(). if (instrumentationManager.getHealthInstrumentations().stream()
allMatch(Instrumentation::isUp)) { .allMatch(Instrumentation::isUp)) {
builder.up(); builder.up();
return; return;
} }
if (instrumentationManager.getHealthInstrumentations().stream(). if (instrumentationManager.getHealthInstrumentations().stream()
allMatch(Instrumentation::isOutOfService)) { .allMatch(Instrumentation::isOutOfService)) {
builder.outOfService(); builder.outOfService();
return; return;
} }
builder.down(); builder.down();
instrumentationManager.getHealthInstrumentations().stream(). instrumentationManager.getHealthInstrumentations().stream()
filter(instrumentation -> !instrumentation.isStarted()). .filter(instrumentation -> !instrumentation.isStarted())
forEach(instrumentation1 -> builder.withException(instrumentation1.getStartException())); .forEach(instrumentation1 -> builder
} .withException(instrumentation1.getStartException()));
}
} }

59
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java
View File

@ -17,38 +17,45 @@ import org.springframework.context.annotation.Configuration;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
@Configuration @Configuration
@EnableConfigurationProperties({RocketMQBinderConfigurationProperties.class, RocketMQExtendedBindingProperties.class}) @EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
RocketMQExtendedBindingProperties.class })
public class RocketMQBinderAutoConfiguration { public class RocketMQBinderAutoConfiguration {
private final RocketMQExtendedBindingProperties extendedBindingProperties; private final RocketMQExtendedBindingProperties extendedBindingProperties;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
@Autowired @Autowired
public RocketMQBinderAutoConfiguration(RocketMQExtendedBindingProperties extendedBindingProperties, public RocketMQBinderAutoConfiguration(
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) { RocketMQExtendedBindingProperties extendedBindingProperties,
this.extendedBindingProperties = extendedBindingProperties; RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties; this.extendedBindingProperties = extendedBindingProperties;
System.setProperty(ClientLogger.CLIENT_LOG_LEVEL, this.rocketBinderConfigurationProperties.getLogLevel()); this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
} System.setProperty(ClientLogger.CLIENT_LOG_LEVEL,
this.rocketBinderConfigurationProperties.getLogLevel());
}
@Bean @Bean
public RocketMQTopicProvisioner provisioningProvider() { public RocketMQTopicProvisioner provisioningProvider() {
return new RocketMQTopicProvisioner(); return new RocketMQTopicProvisioner();
} }
@Bean @Bean
public RocketMQMessageChannelBinder rocketMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider, public RocketMQMessageChannelBinder rocketMessageChannelBinder(
InstrumentationManager instrumentationManager, RocketMQTopicProvisioner provisioningProvider,
ConsumersManager consumersManager) { InstrumentationManager instrumentationManager,
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(consumersManager, extendedBindingProperties, ConsumersManager consumersManager) {
provisioningProvider, rocketBinderConfigurationProperties, instrumentationManager); RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
return binder; consumersManager, extendedBindingProperties, provisioningProvider,
} rocketBinderConfigurationProperties, instrumentationManager);
return binder;
}
@Bean @Bean
public ConsumersManager consumersManager(InstrumentationManager instrumentationManager) { public ConsumersManager consumersManager(
return new ConsumersManager(instrumentationManager, rocketBinderConfigurationProperties); InstrumentationManager instrumentationManager) {
} return new ConsumersManager(instrumentationManager,
rocketBinderConfigurationProperties);
}
} }

27
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderEndpointAutoConfiguration.java
View File

@ -15,19 +15,22 @@ import org.springframework.context.annotation.Configuration;
@AutoConfigureAfter(EndpointAutoConfiguration.class) @AutoConfigureAfter(EndpointAutoConfiguration.class)
public class RocketMQBinderEndpointAutoConfiguration { public class RocketMQBinderEndpointAutoConfiguration {
@Bean @Bean
public RocketMQBinderEndpoint rocketBinderEndpoint() { public RocketMQBinderEndpoint rocketBinderEndpoint() {
return new RocketMQBinderEndpoint(); return new RocketMQBinderEndpoint();
} }
@Bean @Bean
public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(InstrumentationManager instrumentationManager) { public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(
return new RocketMQBinderHealthIndicator(instrumentationManager); InstrumentationManager instrumentationManager) {
} return new RocketMQBinderHealthIndicator(instrumentationManager);
}
@Bean @Bean
public InstrumentationManager instrumentationManager(RocketMQBinderEndpoint rocketBinderEndpoint) { public InstrumentationManager instrumentationManager(
return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(), rocketBinderEndpoint.runtime()); RocketMQBinderEndpoint rocketBinderEndpoint) {
} return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(),
rocketBinderEndpoint.runtime());
}
} }

108
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/Acknowledgement.java
View File

@ -11,68 +11,72 @@ import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
*/ */
public class Acknowledgement { public class Acknowledgement {
/** /**
* for {@link ConsumeConcurrentlyContext} using * for {@link ConsumeConcurrentlyContext} using
*/ */
private ConsumeConcurrentlyStatus consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS; private ConsumeConcurrentlyStatus consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
/** /**
* Message consume retry strategy<br> * Message consume retry strategy<br>
* -1,no retry,put into DLQ directly<br> * -1,no retry,put into DLQ directly<br>
* 0,broker control retry frequency<br> * 0,broker control retry frequency<br>
* >0,client control retry frequency * >0,client control retry frequency
*/ */
private Integer consumeConcurrentlyDelayLevel = 0; private Integer consumeConcurrentlyDelayLevel = 0;
/** /**
* for {@link ConsumeOrderlyContext} using * for {@link ConsumeOrderlyContext} using
*/ */
private ConsumeOrderlyStatus consumeOrderlyStatus = ConsumeOrderlyStatus.SUCCESS; private ConsumeOrderlyStatus consumeOrderlyStatus = ConsumeOrderlyStatus.SUCCESS;
private Long consumeOrderlySuspendCurrentQueueTimeMill = -1L; private Long consumeOrderlySuspendCurrentQueueTimeMill = -1L;
public Acknowledgement setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus consumeConcurrentlyStatus) { public Acknowledgement setConsumeConcurrentlyStatus(
this.consumeConcurrentlyStatus = consumeConcurrentlyStatus; ConsumeConcurrentlyStatus consumeConcurrentlyStatus) {
return this; this.consumeConcurrentlyStatus = consumeConcurrentlyStatus;
} return this;
}
public ConsumeConcurrentlyStatus getConsumeConcurrentlyStatus() { public ConsumeConcurrentlyStatus getConsumeConcurrentlyStatus() {
return consumeConcurrentlyStatus; return consumeConcurrentlyStatus;
} }
public ConsumeOrderlyStatus getConsumeOrderlyStatus() { public ConsumeOrderlyStatus getConsumeOrderlyStatus() {
return consumeOrderlyStatus; return consumeOrderlyStatus;
} }
public Acknowledgement setConsumeOrderlyStatus(ConsumeOrderlyStatus consumeOrderlyStatus) { public Acknowledgement setConsumeOrderlyStatus(
this.consumeOrderlyStatus = consumeOrderlyStatus; ConsumeOrderlyStatus consumeOrderlyStatus) {
return this; this.consumeOrderlyStatus = consumeOrderlyStatus;
} return this;
}
public Integer getConsumeConcurrentlyDelayLevel() { public Integer getConsumeConcurrentlyDelayLevel() {
return consumeConcurrentlyDelayLevel; return consumeConcurrentlyDelayLevel;
} }
public void setConsumeConcurrentlyDelayLevel(Integer consumeConcurrentlyDelayLevel) { public void setConsumeConcurrentlyDelayLevel(Integer consumeConcurrentlyDelayLevel) {
this.consumeConcurrentlyDelayLevel = consumeConcurrentlyDelayLevel; this.consumeConcurrentlyDelayLevel = consumeConcurrentlyDelayLevel;
} }
public Long getConsumeOrderlySuspendCurrentQueueTimeMill() { public Long getConsumeOrderlySuspendCurrentQueueTimeMill() {
return consumeOrderlySuspendCurrentQueueTimeMill; return consumeOrderlySuspendCurrentQueueTimeMill;
} }
public void setConsumeOrderlySuspendCurrentQueueTimeMill(Long consumeOrderlySuspendCurrentQueueTimeMill) { public void setConsumeOrderlySuspendCurrentQueueTimeMill(
this.consumeOrderlySuspendCurrentQueueTimeMill = consumeOrderlySuspendCurrentQueueTimeMill; Long consumeOrderlySuspendCurrentQueueTimeMill) {
} this.consumeOrderlySuspendCurrentQueueTimeMill = consumeOrderlySuspendCurrentQueueTimeMill;
}
public static Acknowledgement buildOrderlyInstance() { public static Acknowledgement buildOrderlyInstance() {
Acknowledgement acknowledgement = new Acknowledgement(); Acknowledgement acknowledgement = new Acknowledgement();
acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUCCESS); acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUCCESS);
return acknowledgement; return acknowledgement;
} }
public static Acknowledgement buildConcurrentlyInstance() { public static Acknowledgement buildConcurrentlyInstance() {
Acknowledgement acknowledgement = new Acknowledgement(); Acknowledgement acknowledgement = new Acknowledgement();
acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS); acknowledgement
return acknowledgement; .setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS);
} return acknowledgement;
}
} }

143
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/ConsumersManager.java
View File

@ -22,85 +22,88 @@ import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsu
*/ */
public class ConsumersManager { public class ConsumersManager {
private final Logger logger = LoggerFactory.getLogger(this.getClass()); private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>(); private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
private final Map<String, Boolean> started = new HashMap<>(); private final Map<String, Boolean> started = new HashMap<>();
private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap = new HashMap<>();
= new HashMap<>(); private final InstrumentationManager instrumentationManager;
private final InstrumentationManager instrumentationManager; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
public ConsumersManager(InstrumentationManager instrumentationManager, public ConsumersManager(InstrumentationManager instrumentationManager,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) { RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
this.instrumentationManager = instrumentationManager; this.instrumentationManager = instrumentationManager;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties; this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
} }
public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group, String topic, public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group,
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) { String topic,
propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic), consumerProperties); ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
ConsumerGroupInstrumentation instrumentation = instrumentationManager.getConsumerGroupInstrumentation(group); propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic),
instrumentationManager.addHealthInstrumentation(instrumentation); consumerProperties);
ConsumerGroupInstrumentation instrumentation = instrumentationManager
.getConsumerGroupInstrumentation(group);
instrumentationManager.addHealthInstrumentation(instrumentation);
if (consumerGroups.containsKey(group)) { if (consumerGroups.containsKey(group)) {
return consumerGroups.get(group); return consumerGroups.get(group);
} }
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group); DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
consumer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr()); consumer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
consumerGroups.put(group, consumer); consumerGroups.put(group, consumer);
started.put(group, false); started.put(group, false);
consumer.setConsumeThreadMax(consumerProperties.getConcurrency()); consumer.setConsumeThreadMax(consumerProperties.getConcurrency());
consumer.setConsumeThreadMin(consumerProperties.getConcurrency()); consumer.setConsumeThreadMin(consumerProperties.getConcurrency());
if (consumerProperties.getExtension().getBroadcasting()) { if (consumerProperties.getExtension().getBroadcasting()) {
consumer.setMessageModel(MessageModel.BROADCASTING); consumer.setMessageModel(MessageModel.BROADCASTING);
} }
logger.info("RocketMQ consuming for SCS group {} created", group); logger.info("RocketMQ consuming for SCS group {} created", group);
return consumer; return consumer;
} }
public synchronized void startConsumers() throws MQClientException { public synchronized void startConsumers() throws MQClientException {
for (String group : getConsumerGroups()) { for (String group : getConsumerGroups()) {
start(group); start(group);
} }
} }
public synchronized void startConsumer(String group) throws MQClientException { public synchronized void startConsumer(String group) throws MQClientException {
start(group); start(group);
} }
public synchronized void stopConsumer(String group) { public synchronized void stopConsumer(String group) {
stop(group); stop(group);
} }
private void stop(String group) { private void stop(String group) {
if (consumerGroups.get(group) != null) { if (consumerGroups.get(group) != null) {
consumerGroups.get(group).shutdown(); consumerGroups.get(group).shutdown();
started.put(group, false); started.put(group, false);
} }
} }
private synchronized void start(String group) throws MQClientException { private synchronized void start(String group) throws MQClientException {
if (started.get(group)) { if (started.get(group)) {
return; return;
} }
ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager.getConsumerGroupInstrumentation( ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager
group); .getConsumerGroupInstrumentation(group);
instrumentationManager.addHealthInstrumentation(groupInstrumentation); instrumentationManager.addHealthInstrumentation(groupInstrumentation);
try { try {
consumerGroups.get(group).start(); consumerGroups.get(group).start();
started.put(group, true); started.put(group, true);
groupInstrumentation.markStartedSuccessfully(); groupInstrumentation.markStartedSuccessfully();
} catch (MQClientException e) { }
groupInstrumentation.markStartFailed(e); catch (MQClientException e) {
logger.error("RocketMQ Consumer hasn't been started. Caused by " + e.getErrorMessage(), e); groupInstrumentation.markStartFailed(e);
throw e; logger.error("RocketMQ Consumer hasn't been started. Caused by "
} + e.getErrorMessage(), e);
} throw e;
}
}
public synchronized Set<String> getConsumerGroups() { public synchronized Set<String> getConsumerGroups() {
return consumerGroups.keySet(); return consumerGroups.keySet();
} }
} }

376
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQInboundChannelAdapter.java
View File

@ -43,211 +43,249 @@ import org.springframework.util.StringUtils;
*/ */
public class RocketMQInboundChannelAdapter extends MessageProducerSupport { public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
private static final Logger logger = LoggerFactory.getLogger(RocketMQInboundChannelAdapter.class); private static final Logger logger = LoggerFactory
.getLogger(RocketMQInboundChannelAdapter.class);
private ConsumerInstrumentation consumerInstrumentation; private ConsumerInstrumentation consumerInstrumentation;
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties; private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
private final String destination; private final String destination;
private final String group; private final String group;
private final InstrumentationManager instrumentationManager; private final InstrumentationManager instrumentationManager;
private final ConsumersManager consumersManager; private final ConsumersManager consumersManager;
private RetryTemplate retryTemplate; private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback; private RecoveryCallback<? extends Object> recoveryCallback;
public RocketMQInboundChannelAdapter(ConsumersManager consumersManager, public RocketMQInboundChannelAdapter(ConsumersManager consumersManager,
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties, ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
String destination, String group, String destination, String group,
InstrumentationManager instrumentationManager) { InstrumentationManager instrumentationManager) {
this.consumersManager = consumersManager; this.consumersManager = consumersManager;
this.consumerProperties = consumerProperties; this.consumerProperties = consumerProperties;
this.destination = destination; this.destination = destination;
this.group = group; this.group = group;
this.instrumentationManager = instrumentationManager; this.instrumentationManager = instrumentationManager;
} }
@Override @Override
protected void doStart() { protected void doStart() {
if (!consumerProperties.getExtension().getEnabled()) { if (!consumerProperties.getExtension().getEnabled()) {
return; return;
} }
String tags = consumerProperties == null ? null : consumerProperties.getExtension().getTags(); String tags = consumerProperties == null ? null
Boolean isOrderly = consumerProperties == null ? false : consumerProperties.getExtension().getOrderly(); : consumerProperties.getExtension().getTags();
Boolean isOrderly = consumerProperties == null ? false
: consumerProperties.getExtension().getOrderly();
DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group, destination, consumerProperties); DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
destination, consumerProperties);
final CloudStreamMessageListener listener = isOrderly ? new CloudStreamMessageListenerOrderly( final CloudStreamMessageListener listener = isOrderly
instrumentationManager) ? new CloudStreamMessageListenerOrderly(instrumentationManager)
: new CloudStreamMessageListenerConcurrently(instrumentationManager); : new CloudStreamMessageListenerConcurrently(instrumentationManager);
if (retryTemplate != null) { if (retryTemplate != null) {
retryTemplate.registerListener(listener); retryTemplate.registerListener(listener);
} }
Set<String> tagsSet = tags == null ? new HashSet<>() : Arrays.stream(tags.split("\\|\\|")).map(String::trim) Set<String> tagsSet = tags == null ? new HashSet<>()
.collect(Collectors.toSet()); : Arrays.stream(tags.split("\\|\\|")).map(String::trim)
.collect(Collectors.toSet());
consumerInstrumentation = instrumentationManager.getConsumerInstrumentation(destination); consumerInstrumentation = instrumentationManager
instrumentationManager.addHealthInstrumentation(consumerInstrumentation); .getConsumerInstrumentation(destination);
instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
try { try {
if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) { if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) {
consumer.subscribe(destination, MessageSelector.bySql(consumerProperties.getExtension().getSql())); consumer.subscribe(destination, MessageSelector
} else { .bySql(consumerProperties.getExtension().getSql()));
consumer.subscribe(destination, String.join(" || ", tagsSet)); }
} else {
consumerInstrumentation.markStartedSuccessfully(); consumer.subscribe(destination, String.join(" || ", tagsSet));
} catch (MQClientException e) { }
consumerInstrumentation.markStartFailed(e); consumerInstrumentation.markStartedSuccessfully();
logger.error("RocketMQ Consumer hasn't been subscribed. Caused by " + e.getErrorMessage(), e); }
throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e); catch (MQClientException e) {
} consumerInstrumentation.markStartFailed(e);
logger.error("RocketMQ Consumer hasn't been subscribed. Caused by "
+ e.getErrorMessage(), e);
throw new RuntimeException("RocketMQ Consumer hasn't been subscribed.", e);
}
consumer.registerMessageListener(listener); consumer.registerMessageListener(listener);
try { try {
consumersManager.startConsumer(group); consumersManager.startConsumer(group);
} catch (MQClientException e) { }
logger.error("RocketMQ Consumer startup failed. Caused by " + e.getErrorMessage(), e); catch (MQClientException e) {
throw new RuntimeException("RocketMQ Consumer startup failed.", e); logger.error(
} "RocketMQ Consumer startup failed. Caused by " + e.getErrorMessage(),
} e);
throw new RuntimeException("RocketMQ Consumer startup failed.", e);
}
}
@Override @Override
protected void doStop() { protected void doStop() {
consumersManager.stopConsumer(group); consumersManager.stopConsumer(group);
} }
public void setRetryTemplate(RetryTemplate retryTemplate) { public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate; this.retryTemplate = retryTemplate;
} }
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) { public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback; this.recoveryCallback = recoveryCallback;
} }
protected class CloudStreamMessageListener implements MessageListener, RetryListener { protected class CloudStreamMessageListener implements MessageListener, RetryListener {
private final InstrumentationManager instrumentationManager; private final InstrumentationManager instrumentationManager;
CloudStreamMessageListener(InstrumentationManager instrumentationManager) { CloudStreamMessageListener(InstrumentationManager instrumentationManager) {
this.instrumentationManager = instrumentationManager; this.instrumentationManager = instrumentationManager;
} }
Acknowledgement consumeMessage(final List<MessageExt> msgs) { Acknowledgement consumeMessage(final List<MessageExt> msgs) {
boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null; boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
try { try {
if (enableRetry) { if (enableRetry) {
return RocketMQInboundChannelAdapter.this.retryTemplate.execute( return RocketMQInboundChannelAdapter.this.retryTemplate.execute(
(RetryCallback<Acknowledgement, Exception>)context -> doSendMsgs(msgs, context), (RetryCallback<Acknowledgement, Exception>) context -> doSendMsgs(
new RecoveryCallback<Acknowledgement>() { msgs, context),
@Override new RecoveryCallback<Acknowledgement>() {
public Acknowledgement recover(RetryContext context) throws Exception { @Override
RocketMQInboundChannelAdapter.this.recoveryCallback.recover(context); public Acknowledgement recover(RetryContext context)
if (ClassUtils.isAssignable(this.getClass(), MessageListenerConcurrently.class)) { throws Exception {
return Acknowledgement.buildConcurrentlyInstance(); RocketMQInboundChannelAdapter.this.recoveryCallback
} else { .recover(context);
return Acknowledgement.buildOrderlyInstance(); if (ClassUtils.isAssignable(this.getClass(),
} MessageListenerConcurrently.class)) {
} return Acknowledgement
}); .buildConcurrentlyInstance();
} else { }
Acknowledgement result = doSendMsgs(msgs, null); else {
instrumentationManager.getConsumerInstrumentation(RocketMQInboundChannelAdapter.this.destination) return Acknowledgement.buildOrderlyInstance();
.markConsumed(); }
return result; }
} });
} catch (Exception e) { }
logger.error("Rocket Message hasn't been processed successfully. Caused by ", e); else {
instrumentationManager.getConsumerInstrumentation(RocketMQInboundChannelAdapter.this.destination) Acknowledgement result = doSendMsgs(msgs, null);
.markConsumedFailure(); instrumentationManager
throw new RuntimeException("Rocket Message hasn't been processed successfully. Caused by ", e); .getConsumerInstrumentation(
} RocketMQInboundChannelAdapter.this.destination)
} .markConsumed();
return result;
}
}
catch (Exception e) {
logger.error(
"Rocket Message hasn't been processed successfully. Caused by ",
e);
instrumentationManager
.getConsumerInstrumentation(
RocketMQInboundChannelAdapter.this.destination)
.markConsumedFailure();
throw new RuntimeException(
"Rocket Message hasn't been processed successfully. Caused by ",
e);
}
}
private Acknowledgement doSendMsgs(final List<MessageExt> msgs, RetryContext context) { private Acknowledgement doSendMsgs(final List<MessageExt> msgs,
List<Acknowledgement> acknowledgements = new ArrayList<>(); RetryContext context) {
msgs.forEach(msg -> { List<Acknowledgement> acknowledgements = new ArrayList<>();
String retryInfo = context == null ? "" : "retryCount-" + String.valueOf(context.getRetryCount()) + "|"; msgs.forEach(msg -> {
logger.debug(retryInfo + "consuming msg:\n" + msg); String retryInfo = context == null ? ""
logger.debug(retryInfo + "message body:\n" + new String(msg.getBody())); : "retryCount-" + String.valueOf(context.getRetryCount()) + "|";
Acknowledgement acknowledgement = new Acknowledgement(); logger.debug(retryInfo + "consuming msg:\n" + msg);
Message<byte[]> toChannel = MessageBuilder.withPayload(msg.getBody()). logger.debug(retryInfo + "message body:\n" + new String(msg.getBody()));
setHeaders(new RocketMQMessageHeaderAccessor(). Acknowledgement acknowledgement = new Acknowledgement();
withAcknowledgment(acknowledgement). Message<byte[]> toChannel = MessageBuilder.withPayload(msg.getBody())
withTags(msg.getTags()). .setHeaders(new RocketMQMessageHeaderAccessor()
withKeys(msg.getKeys()). .withAcknowledgment(acknowledgement)
withFlag(msg.getFlag()). .withTags(msg.getTags()).withKeys(msg.getKeys())
withRocketMessage(msg) .withFlag(msg.getFlag()).withRocketMessage(msg))
).build(); .build();
acknowledgements.add(acknowledgement); acknowledgements.add(acknowledgement);
RocketMQInboundChannelAdapter.this.sendMessage(toChannel); RocketMQInboundChannelAdapter.this.sendMessage(toChannel);
}); });
return acknowledgements.get(0); return acknowledgements.get(0);
} }
@Override @Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) { public <T, E extends Throwable> boolean open(RetryContext context,
return true; RetryCallback<T, E> callback) {
} return true;
}
@Override @Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, public <T, E extends Throwable> void close(RetryContext context,
Throwable throwable) { RetryCallback<T, E> callback, Throwable throwable) {
if (throwable != null) { if (throwable != null) {
instrumentationManager.getConsumerInstrumentation( instrumentationManager
RocketMQInboundChannelAdapter.this.destination) .getConsumerInstrumentation(
.markConsumedFailure(); RocketMQInboundChannelAdapter.this.destination)
} else { .markConsumedFailure();
instrumentationManager.getConsumerInstrumentation( }
RocketMQInboundChannelAdapter.this.destination) else {
.markConsumed(); instrumentationManager
} .getConsumerInstrumentation(
} RocketMQInboundChannelAdapter.this.destination)
.markConsumed();
}
}
@Override @Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, public <T, E extends Throwable> void onError(RetryContext context,
Throwable throwable) { RetryCallback<T, E> callback, Throwable throwable) {
} }
} }
protected class CloudStreamMessageListenerConcurrently extends CloudStreamMessageListener implements protected class CloudStreamMessageListenerConcurrently
MessageListenerConcurrently { extends CloudStreamMessageListener implements MessageListenerConcurrently {
public CloudStreamMessageListenerConcurrently(InstrumentationManager instrumentationManager) { public CloudStreamMessageListenerConcurrently(
super(instrumentationManager); InstrumentationManager instrumentationManager) {
} super(instrumentationManager);
}
@Override @Override
public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs, public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
ConsumeConcurrentlyContext context) { ConsumeConcurrentlyContext context) {
Acknowledgement acknowledgement = consumeMessage(msgs); Acknowledgement acknowledgement = consumeMessage(msgs);
context.setDelayLevelWhenNextConsume(acknowledgement.getConsumeConcurrentlyDelayLevel()); context.setDelayLevelWhenNextConsume(
return acknowledgement.getConsumeConcurrentlyStatus(); acknowledgement.getConsumeConcurrentlyDelayLevel());
} return acknowledgement.getConsumeConcurrentlyStatus();
} }
}
protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener implements protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener
MessageListenerOrderly { implements MessageListenerOrderly {
public CloudStreamMessageListenerOrderly(InstrumentationManager instrumentationManager) { public CloudStreamMessageListenerOrderly(
super(instrumentationManager); InstrumentationManager instrumentationManager) {
} super(instrumentationManager);
}
@Override @Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
Acknowledgement acknowledgement = consumeMessage(msgs); ConsumeOrderlyContext context) {
context.setSuspendCurrentQueueTimeMillis((acknowledgement.getConsumeOrderlySuspendCurrentQueueTimeMill())); Acknowledgement acknowledgement = consumeMessage(msgs);
return acknowledgement.getConsumeOrderlyStatus(); context.setSuspendCurrentQueueTimeMillis(
} (acknowledgement.getConsumeOrderlySuspendCurrentQueueTimeMill()));
return acknowledgement.getConsumeOrderlyStatus();
}
} }
} }

180
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/integration/RocketMQMessageHandler.java
View File

@ -26,106 +26,120 @@ import org.springframework.messaging.MessagingException;
*/ */
public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle { public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
private DefaultMQProducer producer; private DefaultMQProducer producer;
private ProducerInstrumentation producerInstrumentation; private ProducerInstrumentation producerInstrumentation;
private final RocketMQProducerProperties producerProperties; private final RocketMQProducerProperties producerProperties;
private final String destination; private final String destination;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final InstrumentationManager instrumentationManager; private final InstrumentationManager instrumentationManager;
protected volatile boolean running = false; protected volatile boolean running = false;
public RocketMQMessageHandler(String destination, RocketMQProducerProperties producerProperties, public RocketMQMessageHandler(String destination,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties, RocketMQProducerProperties producerProperties,
InstrumentationManager instrumentationManager) { RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
this.destination = destination; InstrumentationManager instrumentationManager) {
this.producerProperties = producerProperties; this.destination = destination;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties; this.producerProperties = producerProperties;
this.instrumentationManager = instrumentationManager; this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
} this.instrumentationManager = instrumentationManager;
}
@Override @Override
public void start() { public void start() {
producer = new DefaultMQProducer(destination); producer = new DefaultMQProducer(destination);
producerInstrumentation = instrumentationManager.getProducerInstrumentation(destination); producerInstrumentation = instrumentationManager
instrumentationManager.addHealthInstrumentation(producerInstrumentation); .getProducerInstrumentation(destination);
instrumentationManager.addHealthInstrumentation(producerInstrumentation);
producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr()); producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
if (producerProperties.getMaxMessageSize() > 0) { if (producerProperties.getMaxMessageSize() > 0) {
producer.setMaxMessageSize(producerProperties.getMaxMessageSize()); producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
} }
try { try {
producer.start(); producer.start();
producerInstrumentation.markStartedSuccessfully(); producerInstrumentation.markStartedSuccessfully();
} catch (MQClientException e) { }
producerInstrumentation.markStartFailed(e); catch (MQClientException e) {
logger.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage()); producerInstrumentation.markStartFailed(e);
throw new MessagingException(e.getMessage(), e); logger.error(
} "RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
running = true; throw new MessagingException(e.getMessage(), e);
} }
running = true;
}
@Override @Override
public void stop() { public void stop() {
if (producer != null) { if (producer != null) {
producer.shutdown(); producer.shutdown();
} }
running = false; running = false;
} }
@Override @Override
public boolean isRunning() { public boolean isRunning() {
return running; return running;
} }
@Override @Override
protected void handleMessageInternal(org.springframework.messaging.Message<?> message) throws Exception { protected void handleMessageInternal(org.springframework.messaging.Message<?> message)
try { throws Exception {
Message toSend; try {
if (message.getPayload() instanceof byte[]) { Message toSend;
toSend = new Message(destination, (byte[])message.getPayload()); if (message.getPayload() instanceof byte[]) {
} else if (message.getPayload() instanceof String) { toSend = new Message(destination, (byte[]) message.getPayload());
toSend = new Message(destination, ((String)message.getPayload()).getBytes()); }
} else { else if (message.getPayload() instanceof String) {
throw new UnsupportedOperationException( toSend = new Message(destination,
"Payload class isn't supported: " + message.getPayload().getClass()); ((String) message.getPayload()).getBytes());
} }
RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message); else {
headerAccessor.setLeaveMutable(true); throw new UnsupportedOperationException("Payload class isn't supported: "
toSend.setDelayTimeLevel(headerAccessor.getDelayTimeLevel()); + message.getPayload().getClass());
toSend.setTags(headerAccessor.getTags()); }
toSend.setKeys(headerAccessor.getKeys()); RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(
toSend.setFlag(headerAccessor.getFlag()); message);
for (Map.Entry<String, String> entry : headerAccessor.getUserProperties().entrySet()) { headerAccessor.setLeaveMutable(true);
toSend.putUserProperty(entry.getKey(), entry.getValue()); toSend.setDelayTimeLevel(headerAccessor.getDelayTimeLevel());
} toSend.setTags(headerAccessor.getTags());
toSend.setKeys(headerAccessor.getKeys());
toSend.setFlag(headerAccessor.getFlag());
for (Map.Entry<String, String> entry : headerAccessor.getUserProperties()
.entrySet()) {
toSend.putUserProperty(entry.getKey(), entry.getValue());
}
SendResult sendRes = producer.send(toSend); SendResult sendRes = producer.send(toSend);
if (!sendRes.getSendStatus().equals(SendStatus.SEND_OK)) { if (!sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
throw new MQClientException("message hasn't been sent", null); throw new MQClientException("message hasn't been sent", null);
} }
if (message instanceof MutableMessage) { if (message instanceof MutableMessage) {
RocketMQMessageHeaderAccessor.putSendResult((MutableMessage)message, sendRes); RocketMQMessageHeaderAccessor.putSendResult((MutableMessage) message,
} sendRes);
instrumentationManager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP, }
Instant.now().toEpochMilli()); instrumentationManager.getRuntime().put(
producerInstrumentation.markSent(); RocketMQBinderConstants.LASTSEND_TIMESTAMP,
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException | Instant.now().toEpochMilli());
UnsupportedOperationException e) { producerInstrumentation.markSent();
producerInstrumentation.markSentFailure(); }
logger.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage()); catch (MQClientException | RemotingException | MQBrokerException
throw new MessagingException(e.getMessage(), e); | InterruptedException | UnsupportedOperationException e) {
} producerInstrumentation.markSentFailure();
logger.error(
"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
throw new MessagingException(e.getMessage(), e);
}
} }
} }

34
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerGroupInstrumentation.java
View File

@ -9,26 +9,26 @@ import com.codahale.metrics.MetricRegistry;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class ConsumerGroupInstrumentation extends Instrumentation { public class ConsumerGroupInstrumentation extends Instrumentation {
private MetricRegistry metricRegistry; private MetricRegistry metricRegistry;
private AtomicBoolean delayedStart = new AtomicBoolean(false); private AtomicBoolean delayedStart = new AtomicBoolean(false);
public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) { public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) {
super(name); super(name);
this.metricRegistry = metricRegistry; this.metricRegistry = metricRegistry;
} }
public void markDelayedStart() { public void markDelayedStart() {
delayedStart.set(true); delayedStart.set(true);
} }
@Override @Override
public boolean isUp() { public boolean isUp() {
return started.get() || delayedStart.get(); return started.get() || delayedStart.get();
} }
@Override @Override
public boolean isOutOfService() { public boolean isOutOfService() {
return !started.get() && startException == null && !delayedStart.get(); return !started.get() && startException == null && !delayedStart.get();
} }
} }

45
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ConsumerInstrumentation.java
View File

@ -1,37 +1,40 @@
package org.springframework.cloud.stream.binder.rocketmq.metrics; package org.springframework.cloud.stream.binder.rocketmq.metrics;
import static com.codahale.metrics.MetricRegistry.name;
import com.codahale.metrics.Counter; import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter; import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.MetricRegistry;
import static com.codahale.metrics.MetricRegistry.name;
/** /**
* @author juven.xuxb * @author juven.xuxb
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class ConsumerInstrumentation extends Instrumentation { public class ConsumerInstrumentation extends Instrumentation {
private final Counter totalConsumed; private final Counter totalConsumed;
private final Counter totalConsumedFailures; private final Counter totalConsumedFailures;
private final Meter consumedPerSecond; private final Meter consumedPerSecond;
private final Meter consumedFailuresPerSecond; private final Meter consumedFailuresPerSecond;
public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) { public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) {
super(baseMetricName); super(baseMetricName);
this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed")); this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed"));
this.consumedPerSecond = registry.meter(name(baseMetricName, "consumedPerSecond")); this.consumedPerSecond = registry
this.totalConsumedFailures = registry.counter(name(baseMetricName, "totalConsumedFailures")); .meter(name(baseMetricName, "consumedPerSecond"));
this.consumedFailuresPerSecond = registry.meter(name(baseMetricName, "consumedFailuresPerSecond")); this.totalConsumedFailures = registry
} .counter(name(baseMetricName, "totalConsumedFailures"));
this.consumedFailuresPerSecond = registry
.meter(name(baseMetricName, "consumedFailuresPerSecond"));
}
public void markConsumed() { public void markConsumed() {
totalConsumed.inc(); totalConsumed.inc();
consumedPerSecond.mark(); consumedPerSecond.mark();
} }
public void markConsumedFailure() { public void markConsumedFailure() {
totalConsumedFailures.inc(); totalConsumedFailures.inc();
consumedFailuresPerSecond.mark(); consumedFailuresPerSecond.mark();
} }
} }

62
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/Instrumentation.java
View File

@ -7,44 +7,44 @@ import java.util.concurrent.atomic.AtomicBoolean;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class Instrumentation { public class Instrumentation {
private final String name; private final String name;
protected final AtomicBoolean started = new AtomicBoolean(false); protected final AtomicBoolean started = new AtomicBoolean(false);
protected Exception startException = null; protected Exception startException = null;
Instrumentation(String name) { Instrumentation(String name) {
this.name = name; this.name = name;
} }
public boolean isDown() { public boolean isDown() {
return startException != null; return startException != null;
} }
public boolean isUp() { public boolean isUp() {
return started.get(); return started.get();
} }
public boolean isOutOfService() { public boolean isOutOfService() {
return !started.get() && startException == null; return !started.get() && startException == null;
} }
public void markStartedSuccessfully() { public void markStartedSuccessfully() {
started.set(true); started.set(true);
} }
public void markStartFailed(Exception e) { public void markStartFailed(Exception e) {
started.set(false); started.set(false);
startException = e; startException = e;
} }
public String getName() { public String getName() {
return name; return name;
} }
public boolean isStarted() { public boolean isStarted() {
return started.get(); return started.get();
} }
public Exception getStartException() { public Exception getStartException() {
return startException; return startException;
} }
} }

73
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/InstrumentationManager.java
View File

@ -12,46 +12,51 @@ import com.codahale.metrics.MetricRegistry;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class InstrumentationManager { public class InstrumentationManager {
private final MetricRegistry metricRegistry; private final MetricRegistry metricRegistry;
private final Map<String, Object> runtime; private final Map<String, Object> runtime;
private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>(); private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>();
private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>(); private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>();
private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>(); private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>();
private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>(); private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
public InstrumentationManager(MetricRegistry metricRegistry, Map<String, Object> runtime) { public InstrumentationManager(MetricRegistry metricRegistry,
this.metricRegistry = metricRegistry; Map<String, Object> runtime) {
this.runtime = runtime; this.metricRegistry = metricRegistry;
} this.runtime = runtime;
}
public ProducerInstrumentation getProducerInstrumentation(String destination) { public ProducerInstrumentation getProducerInstrumentation(String destination) {
String key = "scs-rocketmq.producer." + destination; String key = "scs-rocketmq.producer." + destination;
producerInstrumentations.putIfAbsent(key, new ProducerInstrumentation(metricRegistry, key)); producerInstrumentations.putIfAbsent(key,
return producerInstrumentations.get(key); new ProducerInstrumentation(metricRegistry, key));
} return producerInstrumentations.get(key);
}
public ConsumerInstrumentation getConsumerInstrumentation(String destination) { public ConsumerInstrumentation getConsumerInstrumentation(String destination) {
String key = "scs-rocketmq.consumer." + destination; String key = "scs-rocketmq.consumer." + destination;
consumeInstrumentations.putIfAbsent(key, new ConsumerInstrumentation(metricRegistry, key)); consumeInstrumentations.putIfAbsent(key,
return consumeInstrumentations.get(key); new ConsumerInstrumentation(metricRegistry, key));
} return consumeInstrumentations.get(key);
}
public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) { public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) {
String key = "scs-rocketmq.consumerGroup." + group; String key = "scs-rocketmq.consumerGroup." + group;
consumerGroupsInstrumentations.putIfAbsent(key, new ConsumerGroupInstrumentation(metricRegistry, key)); consumerGroupsInstrumentations.putIfAbsent(key,
return consumerGroupsInstrumentations.get(key); new ConsumerGroupInstrumentation(metricRegistry, key));
} return consumerGroupsInstrumentations.get(key);
}
public Set<Instrumentation> getHealthInstrumentations() { public Set<Instrumentation> getHealthInstrumentations() {
return healthInstrumentations.entrySet().stream().map(Map.Entry::getValue).collect(Collectors.toSet()); return healthInstrumentations.entrySet().stream().map(Map.Entry::getValue)
} .collect(Collectors.toSet());
}
public void addHealthInstrumentation(Instrumentation instrumentation) { public void addHealthInstrumentation(Instrumentation instrumentation) {
healthInstrumentations.put(instrumentation.getName(), instrumentation); healthInstrumentations.put(instrumentation.getName(), instrumentation);
} }
public Map<String, Object> getRuntime() { public Map<String, Object> getRuntime() {
return runtime; return runtime;
} }
} }

44
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/metrics/ProducerInstrumentation.java
View File

@ -1,37 +1,39 @@
package org.springframework.cloud.stream.binder.rocketmq.metrics; package org.springframework.cloud.stream.binder.rocketmq.metrics;
import static com.codahale.metrics.MetricRegistry.name;
import com.codahale.metrics.Counter; import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter; import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.MetricRegistry;
import static com.codahale.metrics.MetricRegistry.name;
/** /**
* @author juven.xuxb * @author juven.xuxb
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class ProducerInstrumentation extends Instrumentation { public class ProducerInstrumentation extends Instrumentation {
private final Counter totalSent; private final Counter totalSent;
private final Counter totalSentFailures; private final Counter totalSentFailures;
private final Meter sentPerSecond; private final Meter sentPerSecond;
private final Meter sentFailuresPerSecond; private final Meter sentFailuresPerSecond;
public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) { public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) {
super(baseMetricName); super(baseMetricName);
this.totalSent = registry.counter(name(baseMetricName, "totalSent")); this.totalSent = registry.counter(name(baseMetricName, "totalSent"));
this.totalSentFailures = registry.counter(name(baseMetricName, "totalSentFailures")); this.totalSentFailures = registry
this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond")); .counter(name(baseMetricName, "totalSentFailures"));
this.sentFailuresPerSecond = registry.meter(name(baseMetricName, "sentFailuresPerSecond")); this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond"));
} this.sentFailuresPerSecond = registry
.meter(name(baseMetricName, "sentFailuresPerSecond"));
}
public void markSent() { public void markSent() {
totalSent.inc(); totalSent.inc();
sentPerSecond.mark(); sentPerSecond.mark();
} }
public void markSentFailure() { public void markSentFailure() {
totalSentFailures.inc(); totalSentFailures.inc();
sentFailuresPerSecond.mark(); sentFailuresPerSecond.mark();
} }
} }

28
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBinderConfigurationProperties.java
View File

@ -9,24 +9,24 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder") @ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
public class RocketMQBinderConfigurationProperties { public class RocketMQBinderConfigurationProperties {
private String namesrvAddr; private String namesrvAddr;
private String logLevel = "ERROR"; private String logLevel = "ERROR";
public String getNamesrvAddr() { public String getNamesrvAddr() {
return namesrvAddr; return namesrvAddr;
} }
public void setNamesrvAddr(String namesrvAddr) { public void setNamesrvAddr(String namesrvAddr) {
this.namesrvAddr = namesrvAddr; this.namesrvAddr = namesrvAddr;
} }
public String getLogLevel() { public String getLogLevel() {
return logLevel; return logLevel;
} }
public void setLogLevel(String logLevel) { public void setLogLevel(String logLevel) {
this.logLevel = logLevel; this.logLevel = logLevel;
} }
} }

28
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java
View File

@ -6,23 +6,23 @@ package org.springframework.cloud.stream.binder.rocketmq.properties;
*/ */
public class RocketMQBindingProperties { public class RocketMQBindingProperties {
private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties(); private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties();
private RocketMQProducerProperties producer = new RocketMQProducerProperties(); private RocketMQProducerProperties producer = new RocketMQProducerProperties();
public RocketMQConsumerProperties getConsumer() { public RocketMQConsumerProperties getConsumer() {
return consumer; return consumer;
} }
public void setConsumer(RocketMQConsumerProperties consumer) { public void setConsumer(RocketMQConsumerProperties consumer) {
this.consumer = consumer; this.consumer = consumer;
} }
public RocketMQProducerProperties getProducer() { public RocketMQProducerProperties getProducer() {
return producer; return producer;
} }
public void setProducer(RocketMQProducerProperties producer) { public void setProducer(RocketMQProducerProperties producer) {
this.producer = producer; this.producer = producer;
} }
} }

6
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQConsumerProperties.java
View File

@ -19,9 +19,9 @@ public class RocketMQConsumerProperties {
private String tags; private String tags;
/** /**
* {@link MQPushConsumer#subscribe(String, MessageSelector)} * {@link MQPushConsumer#subscribe(String, MessageSelector)}
* {@link MessageSelector#bySql(String)} * {@link MessageSelector#bySql(String)}
*/ */
private String sql; private String sql;
/** /**

94
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java
View File

@ -12,53 +12,59 @@ import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
*/ */
@ConfigurationProperties("spring.cloud.stream.rocketmq") @ConfigurationProperties("spring.cloud.stream.rocketmq")
public class RocketMQExtendedBindingProperties implements public class RocketMQExtendedBindingProperties implements
ExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties> { ExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties> {
private Map<String, RocketMQBindingProperties> bindings = new HashMap<>(); private Map<String, RocketMQBindingProperties> bindings = new HashMap<>();
public Map<String, RocketMQBindingProperties> getBindings() { public Map<String, RocketMQBindingProperties> getBindings() {
return this.bindings; return this.bindings;
} }
public void setBindings(Map<String, RocketMQBindingProperties> bindings) { public void setBindings(Map<String, RocketMQBindingProperties> bindings) {
this.bindings = bindings; this.bindings = bindings;
} }
@Override @Override
public synchronized RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) { public synchronized RocketMQConsumerProperties getExtendedConsumerProperties(
if (bindings.containsKey(channelName)) { String channelName) {
if (bindings.get(channelName).getConsumer() != null) { if (bindings.containsKey(channelName)) {
return bindings.get(channelName).getConsumer(); if (bindings.get(channelName).getConsumer() != null) {
} else { return bindings.get(channelName).getConsumer();
RocketMQConsumerProperties properties = new RocketMQConsumerProperties(); }
this.bindings.get(channelName).setConsumer(properties); else {
return properties; RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
} this.bindings.get(channelName).setConsumer(properties);
} else { return properties;
RocketMQConsumerProperties properties = new RocketMQConsumerProperties(); }
RocketMQBindingProperties rbp = new RocketMQBindingProperties(); }
rbp.setConsumer(properties); else {
bindings.put(channelName, rbp); RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
return properties; RocketMQBindingProperties rbp = new RocketMQBindingProperties();
} rbp.setConsumer(properties);
} bindings.put(channelName, rbp);
return properties;
}
}
@Override @Override
public synchronized RocketMQProducerProperties getExtendedProducerProperties(String channelName) { public synchronized RocketMQProducerProperties getExtendedProducerProperties(
if (bindings.containsKey(channelName)) { String channelName) {
if (bindings.get(channelName).getProducer() != null) { if (bindings.containsKey(channelName)) {
return bindings.get(channelName).getProducer(); if (bindings.get(channelName).getProducer() != null) {
} else { return bindings.get(channelName).getProducer();
RocketMQProducerProperties properties = new RocketMQProducerProperties(); }
this.bindings.get(channelName).setProducer(properties); else {
return properties; RocketMQProducerProperties properties = new RocketMQProducerProperties();
} this.bindings.get(channelName).setProducer(properties);
} else { return properties;
RocketMQProducerProperties properties = new RocketMQProducerProperties(); }
RocketMQBindingProperties rbp = new RocketMQBindingProperties(); }
rbp.setProducer(properties); else {
bindings.put(channelName, rbp); RocketMQProducerProperties properties = new RocketMQProducerProperties();
return properties; RocketMQBindingProperties rbp = new RocketMQBindingProperties();
} rbp.setProducer(properties);
} bindings.put(channelName, rbp);
return properties;
}
}
} }

35
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQProducerProperties.java
View File

@ -8,28 +8,27 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
*/ */
public class RocketMQProducerProperties { public class RocketMQProducerProperties {
private Boolean enabled = true; private Boolean enabled = true;
/** /**
* Maximum allowed message size in bytes * Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
* {@link DefaultMQProducer#maxMessageSize} */
*/ private Integer maxMessageSize = 0;
private Integer maxMessageSize = 0;
public Boolean getEnabled() { public Boolean getEnabled() {
return enabled; return enabled;
} }
public void setEnabled(Boolean enabled) { public void setEnabled(Boolean enabled) {
this.enabled = enabled; this.enabled = enabled;
} }
public Integer getMaxMessageSize() { public Integer getMaxMessageSize() {
return maxMessageSize; return maxMessageSize;
} }
public void setMaxMessageSize(Integer maxMessageSize) { public void setMaxMessageSize(Integer maxMessageSize) {
this.maxMessageSize = maxMessageSize; this.maxMessageSize = maxMessageSize;
} }
} }

104
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/provisioning/RocketMQTopicProvisioner.java
View File

@ -17,73 +17,71 @@ import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
* @author Timur Valiev * @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a> * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/ */
public class RocketMQTopicProvisioner public class RocketMQTopicProvisioner implements
implements ProvisioningProvider<ExtendedConsumerProperties<RocketMQConsumerProperties>, ExtendedProducerProperties<RocketMQProducerProperties>> {
ProvisioningProvider<ExtendedConsumerProperties<RocketMQConsumerProperties>,
ExtendedProducerProperties<RocketMQProducerProperties>> {
private static final Logger logger = LoggerFactory.getLogger(RocketMQTopicProvisioner.class); private static final Logger logger = LoggerFactory
.getLogger(RocketMQTopicProvisioner.class);
@Override @Override
public ProducerDestination provisionProducerDestination(String name, public ProducerDestination provisionProducerDestination(String name,
ExtendedProducerProperties<RocketMQProducerProperties> ExtendedProducerProperties<RocketMQProducerProperties> properties)
properties) throws ProvisioningException {
throws ProvisioningException { checkTopic(name);
checkTopic(name); return new RocketProducerDestination(name);
return new RocketProducerDestination(name); }
}
@Override @Override
public ConsumerDestination provisionConsumerDestination(String name, String group, public ConsumerDestination provisionConsumerDestination(String name, String group,
ExtendedConsumerProperties<RocketMQConsumerProperties> ExtendedConsumerProperties<RocketMQConsumerProperties> properties)
properties) throws ProvisioningException {
throws ProvisioningException { checkTopic(name);
checkTopic(name); return new RocketConsumerDestination(name);
return new RocketConsumerDestination(name); }
}
private void checkTopic(String topic) { private void checkTopic(String topic) {
try { try {
Validators.checkTopic(topic); Validators.checkTopic(topic);
} catch (MQClientException e) { }
logger.error("topic check error: " + topic, e); catch (MQClientException e) {
throw new AssertionError(e); // Can't happen logger.error("topic check error: " + topic, e);
} throw new AssertionError(e); // Can't happen
} }
}
private static final class RocketProducerDestination implements ProducerDestination { private static final class RocketProducerDestination implements ProducerDestination {
private final String producerDestinationName; private final String producerDestinationName;
RocketProducerDestination(String destinationName) { RocketProducerDestination(String destinationName) {
this.producerDestinationName = destinationName; this.producerDestinationName = destinationName;
} }
@Override @Override
public String getName() { public String getName() {
return producerDestinationName; return producerDestinationName;
} }
@Override @Override
public String getNameForPartition(int partition) { public String getNameForPartition(int partition) {
return producerDestinationName; return producerDestinationName;
} }
} }
private static final class RocketConsumerDestination implements ConsumerDestination { private static final class RocketConsumerDestination implements ConsumerDestination {
private final String consumerDestinationName; private final String consumerDestinationName;
RocketConsumerDestination(String consumerDestinationName) { RocketConsumerDestination(String consumerDestinationName) {
this.consumerDestinationName = consumerDestinationName; this.consumerDestinationName = consumerDestinationName;
} }
@Override @Override
public String getName() { public String getName() {
return this.consumerDestinationName; return this.consumerDestinationName;
} }
} }
} }