Upgrade RocketMQ to 4.9.2.

fix when 'spring-boot-starter-actuator' doesn't exist, throw ClassNotFount.
add example.
optimize pom.
pull/2406/head
Freeman Lau 3 years ago
parent 35e0d7a1a7
commit 842ea176c0

@ -90,7 +90,7 @@
<curator.version>4.0.1</curator.version> <curator.version>4.0.1</curator.version>
<!-- Apache RocketMQ --> <!-- Apache RocketMQ -->
<rocketmq.version>4.6.1</rocketmq.version> <rocketmq.version>4.9.2</rocketmq.version>
<!-- Maven Plugin Versions --> <!-- Maven Plugin Versions -->
<maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version> <maven-compiler-plugin.version>3.8.1</maven-compiler-plugin.version>

@ -22,6 +22,10 @@
<groupId>com.alibaba.cloud</groupId> <groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId> <artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
</dependency> </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId> <artifactId>spring-boot-starter-json</artifactId>

@ -61,6 +61,8 @@ public class RocketMQComprehensiveApplication {
return flux -> flux.map(user -> { return flux -> flux.map(user -> {
user.setId(String.valueOf( user.setId(String.valueOf(
Long.parseLong(user.getId()) * Long.parseLong(user.getId()))); Long.parseLong(user.getId()) * Long.parseLong(user.getId())));
user.setName("not freeman");
user.getMeta().put("hobbies", Arrays.asList("programming"));
return user; return user;
}); });
} }

@ -5,24 +5,32 @@ spring:
name: rocketmq-comprehensive-example name: rocketmq-comprehensive-example
cloud: cloud:
stream: stream:
function:
definition: producer;consumer;processor
rocketmq: rocketmq:
binder: binder:
name-server: 127.0.0.1:9876 name-server: 127.0.0.1:9876
function: bindings:
definition: producer;processor;consumer # TODO producer must have a group, need optimization !!!
producer-out-0:
producer:
group: output_1
processor-out-0:
producer:
group: output_2
bindings: bindings:
producer-out-0: producer-out-0:
destination: num destination: num
group: producer_group
processor-in-0:
destination: num
group: processor_group
processor-out-0: processor-out-0:
destination: square destination: square
processor-in-0:
destination: num
group: processor_group group: processor_group
consumer-in-0: consumer-in-0:
destination: square destination: square
group: consumer_group group: consumer_group
logging: logging:
level: level:
org.apache.rocketmq: debug org.springframework.context.support: debug

@ -24,24 +24,6 @@
<optional>true</optional> <optional>true</optional>
</dependency> </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-autoconfigure</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-actuator</artifactId>
<optional>true</optional>
</dependency>
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-actuator-autoconfigure</artifactId> <artifactId>spring-boot-actuator-autoconfigure</artifactId>
@ -56,18 +38,13 @@
<groupId>org.apache.rocketmq</groupId> <groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-acl</artifactId> <artifactId>rocketmq-acl</artifactId>
</dependency> </dependency>
<!-- Testing -->
<dependency> <dependency>
<groupId>org.springframework.boot</groupId> <groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId> <artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies> </dependencies>
</project> </project>

@ -48,7 +48,7 @@ public class ExtendedBindingHandlerMappingsProviderConfiguration {
} }
@Bean @Bean
public RocketMQConfigBeanPostProcessor rocketMQConfigBeanPostProcessor() { public static RocketMQConfigBeanPostProcessor rocketMQConfigBeanPostProcessor() {
return new RocketMQConfigBeanPostProcessor(); return new RocketMQConfigBeanPostProcessor();
} }

@ -24,6 +24,7 @@ import com.alibaba.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvis
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator; import org.springframework.boot.actuate.autoconfigure.health.ConditionalOnEnabledHealthIndicator;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Bean;
@ -46,13 +47,6 @@ public class RocketMQBinderAutoConfiguration {
@Autowired @Autowired
private RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties; private RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
@Bean
@ConditionalOnEnabledHealthIndicator("rocketmq")
@ConditionalOnClass(name = "org.springframework.boot.actuate.health.HealthIndicator")
public RocketMQBinderHealthIndicator rocketMQBinderHealthIndicator() {
return new RocketMQBinderHealthIndicator();
}
@Bean @Bean
public RocketMQTopicProvisioner rocketMQTopicProvisioner() { public RocketMQTopicProvisioner rocketMQTopicProvisioner() {
return new RocketMQTopicProvisioner(); return new RocketMQTopicProvisioner();
@ -65,4 +59,16 @@ public class RocketMQBinderAutoConfiguration {
extendedBindingProperties, provisioningProvider); extendedBindingProperties, provisioningProvider);
} }
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(HealthIndicator.class)
@ConditionalOnEnabledHealthIndicator("rocketmq")
static class KafkaBinderHealthIndicatorConfiguration {
@Bean
public RocketMQBinderHealthIndicator rocketMQBinderHealthIndicator() {
return new RocketMQBinderHealthIndicator();
}
}
} }

@ -33,7 +33,7 @@ import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils; import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils; import org.springframework.util.ObjectUtils;
/** /**
* @author zkzlx * @author zkzlx
@ -138,13 +138,13 @@ public final class RocketMQMessageConverterSupport {
if (Objects.nonNull(headers) && !headers.isEmpty()) { if (Objects.nonNull(headers) && !headers.isEmpty()) {
Object tag = headers.getOrDefault(Headers.TAGS, Object tag = headers.getOrDefault(Headers.TAGS,
headers.get(toRocketHeaderKey(Headers.TAGS))); headers.get(toRocketHeaderKey(Headers.TAGS)));
if (StringUtils.hasLength(tag.toString())) { if (!ObjectUtils.isEmpty(tag)) {
rocketMsg.setTags(String.valueOf(tag)); rocketMsg.setTags(String.valueOf(tag));
} }
Object keys = headers.getOrDefault(Headers.KEYS, Object keys = headers.getOrDefault(Headers.KEYS,
headers.get(toRocketHeaderKey(Headers.KEYS))); headers.get(toRocketHeaderKey(Headers.KEYS)));
if (StringUtils.hasLength(keys.toString())) { if (!ObjectUtils.isEmpty(keys)) {
rocketMsg.setKeys(keys.toString()); rocketMsg.setKeys(keys.toString());
} }
Object flagObj = headers.getOrDefault(Headers.FLAG, Object flagObj = headers.getOrDefault(Headers.FLAG,

@ -19,7 +19,7 @@ package com.alibaba.cloud.stream.binder.rocketmq;
import com.alibaba.cloud.stream.binder.rocketmq.autoconfigurate.RocketMQBinderAutoConfiguration; import com.alibaba.cloud.stream.binder.rocketmq.autoconfigurate.RocketMQBinderAutoConfiguration;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties; import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties; import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
import org.junit.Test; import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.boot.test.context.runner.ApplicationContextRunner;

Loading…
Cancel
Save