Merge pull request #112 from fangjian0423/master
Add RocketMQ binder and refactor Sentinel ReadableDataSourcepull/117/head
xiaojing
6 years ago
committed by
GitHub
commit
c34ecfffd8
@ -0,0 +1,52 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-alibaba-examples</artifactId>
|
||||
<version>0.2.1.BUILD-SNAPSHOT</version>
|
||||
<relativePath>../../pom.xml</relativePath>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
|
||||
<artifactId>rocketmq-example</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<description>Example demonstrating how to use rocketmq</description>
|
||||
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-deploy-plugin</artifactId>
|
||||
<version>${maven-deploy-plugin.version}</version>
|
||||
<configuration>
|
||||
<skip>true</skip>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@ -0,0 +1,39 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class Foo {
|
||||
|
||||
private int id;
|
||||
private String tag;
|
||||
|
||||
public Foo() {
|
||||
}
|
||||
|
||||
public Foo(int id, String tag) {
|
||||
this.id = id;
|
||||
this.tag = tag;
|
||||
}
|
||||
|
||||
public int getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(int id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getTag() {
|
||||
return tag;
|
||||
}
|
||||
|
||||
public void setTag(String tag) {
|
||||
this.tag = tag;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Foo{" + "id=" + id + ", tag='" + tag + '\'' + '}';
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import org.springframework.cloud.stream.annotation.StreamListener;
|
||||
import org.springframework.messaging.handler.annotation.Payload;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Service
|
||||
public class ReceiveService {
|
||||
|
||||
@StreamListener("input1")
|
||||
public void receiveInput1(String receiveMsg) {
|
||||
System.out.println("input1 receive: " + receiveMsg);
|
||||
}
|
||||
|
||||
@StreamListener("input2")
|
||||
public void receiveInput2(String receiveMsg) {
|
||||
System.out.println("input2 receive: " + receiveMsg);
|
||||
}
|
||||
|
||||
@StreamListener("input3")
|
||||
public void receiveInput3(@Payload Foo foo) {
|
||||
System.out.println("input3 receive: " + foo);
|
||||
}
|
||||
|
||||
@StreamListener("input1")
|
||||
public void receiveInput1Again(String receiveMsg) {
|
||||
System.out.println("input1 receive again: " + receiveMsg);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,65 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySink;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.Input;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableBinding({ Source.class, MySink.class })
|
||||
public class RocketMQApplication {
|
||||
|
||||
public interface MySink {
|
||||
|
||||
@Input("input1")
|
||||
SubscribableChannel input1();
|
||||
|
||||
@Input("input2")
|
||||
SubscribableChannel input2();
|
||||
|
||||
@Input("input3")
|
||||
SubscribableChannel input3();
|
||||
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(RocketMQApplication.class, args);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CustomRunner customRunner() {
|
||||
return new CustomRunner();
|
||||
}
|
||||
|
||||
public static class CustomRunner implements CommandLineRunner {
|
||||
@Autowired
|
||||
private SenderService senderService;
|
||||
|
||||
@Override
|
||||
public void run(String... args) throws Exception {
|
||||
int count = 5;
|
||||
for (int index = 1; index <= count; index++) {
|
||||
String msgContent = "msg-" + index;
|
||||
if (index % 3 == 0) {
|
||||
senderService.send(msgContent);
|
||||
}
|
||||
else if (index % 3 == 1) {
|
||||
senderService.sendWithTags(msgContent, "tagStr");
|
||||
}
|
||||
else {
|
||||
senderService.sendObject(new Foo(index, "foo"), "tagObj");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
package org.springframework.cloud.alibaba.cloud.examples;
|
||||
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.rocketmq.common.message.MessageConst;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.util.MimeTypeUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Service
|
||||
public class SenderService {
|
||||
|
||||
@Autowired
|
||||
private Source source;
|
||||
|
||||
public void send(String msg) throws Exception {
|
||||
source.output().send(MessageBuilder.withPayload(msg).build());
|
||||
}
|
||||
|
||||
public <T> void sendWithTags(T msg, String tag) throws Exception {
|
||||
Message message = MessageBuilder.createMessage(msg,
|
||||
new MessageHeaders(Stream.of(tag).collect(Collectors
|
||||
.toMap(str -> MessageConst.PROPERTY_TAGS, String::toString))));
|
||||
source.output().send(message);
|
||||
}
|
||||
|
||||
public <T> void sendObject(T msg, String tag) throws Exception {
|
||||
Message message = MessageBuilder.withPayload(msg)
|
||||
.setHeader(MessageConst.PROPERTY_TAGS, tag)
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
|
||||
.build();
|
||||
source.output().send(message);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,31 @@
|
||||
spring.cloud.stream.default-binder=rocketmq
|
||||
|
||||
spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876
|
||||
|
||||
spring.cloud.stream.bindings.output.destination=test-topic
|
||||
spring.cloud.stream.bindings.output.content-type=application/json
|
||||
|
||||
spring.cloud.stream.bindings.input1.destination=test-topic
|
||||
spring.cloud.stream.bindings.input1.content-type=application/json
|
||||
spring.cloud.stream.bindings.input1.group=test-group1
|
||||
spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true
|
||||
spring.cloud.stream.bindings.input1.consumer.maxAttempts=1
|
||||
|
||||
spring.cloud.stream.bindings.input2.destination=test-topic
|
||||
spring.cloud.stream.bindings.input2.content-type=application/json
|
||||
spring.cloud.stream.bindings.input2.group=test-group2
|
||||
spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false
|
||||
spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr
|
||||
spring.cloud.stream.bindings.input2.consumer.concurrency=20
|
||||
spring.cloud.stream.bindings.input2.consumer.maxAttempts=1
|
||||
|
||||
spring.cloud.stream.bindings.input3.destination=test-topic
|
||||
spring.cloud.stream.bindings.input3.content-type=application/json
|
||||
spring.cloud.stream.bindings.input3.group=test-group3
|
||||
spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj
|
||||
spring.cloud.stream.bindings.input3.consumer.concurrency=20
|
||||
spring.cloud.stream.bindings.input3.consumer.maxAttempts=1
|
||||
|
||||
server.port=28081
|
||||
|
||||
management.endpoints.web.exposure.include=*
|
||||
2
spring-cloud-alibaba-examples/sentinel-example/sentinel-core-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/JsonFlowRuleListParser.java → spring-cloud-alibaba-examples/sentinel-example/sentinel-core-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/JsonFlowRuleListConverter.java
2
spring-cloud-alibaba-examples/sentinel-example/sentinel-core-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/JsonFlowRuleListParser.java → spring-cloud-alibaba-examples/sentinel-example/sentinel-core-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/JsonFlowRuleListConverter.java
@ -0,0 +1,16 @@
|
||||
[
|
||||
{
|
||||
"resource": "abc0",
|
||||
"count": 20.0,
|
||||
"grade": 0,
|
||||
"passCount": 0,
|
||||
"timeWindow": 10
|
||||
},
|
||||
{
|
||||
"resource": "abc1",
|
||||
"count": 15.0,
|
||||
"grade": 0,
|
||||
"passCount": 0,
|
||||
"timeWindow": 10
|
||||
}
|
||||
]
|
||||
@ -0,0 +1,26 @@
|
||||
[
|
||||
{
|
||||
"resource": "resource",
|
||||
"controlBehavior": 0,
|
||||
"count": 1,
|
||||
"grade": 1,
|
||||
"limitApp": "default",
|
||||
"strategy": 0
|
||||
},
|
||||
{
|
||||
"resource": "p",
|
||||
"controlBehavior": 0,
|
||||
"count": 1,
|
||||
"grade": 1,
|
||||
"limitApp": "default",
|
||||
"strategy": 0
|
||||
},
|
||||
{
|
||||
"resource": "abc",
|
||||
"controlBehavior": 0,
|
||||
"count": 1,
|
||||
"grade": 1,
|
||||
"limitApp": "default",
|
||||
"strategy": 0
|
||||
}
|
||||
]
|
||||
@ -0,0 +1,21 @@
|
||||
<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<Rules>
|
||||
<FlowRule>
|
||||
<resource>resource</resource>
|
||||
<controlBehavior>0</controlBehavior>
|
||||
<count>1</count>
|
||||
<grade>1</grade>
|
||||
<limitApp>default</limitApp>
|
||||
<strategy>0</strategy>
|
||||
</FlowRule>
|
||||
<FlowRule>
|
||||
<resource>test</resource>
|
||||
<controlBehavior>0</controlBehavior>
|
||||
<count>1</count>
|
||||
<grade>1</grade>
|
||||
<limitApp>default</limitApp>
|
||||
<strategy>0</strategy>
|
||||
</FlowRule>
|
||||
</Rules>
|
||||
|
||||
|
||||
@ -1,152 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.alibaba.sentinel.datasource;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
|
||||
import org.springframework.core.io.support.PropertiesLoaderUtils;
|
||||
import org.springframework.core.io.support.ResourcePatternResolver;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
import static org.springframework.core.io.support.ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX;
|
||||
|
||||
/**
|
||||
* {@link ReadableDataSource} Loader
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class DataSourceLoader {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(DataSourceLoader.class);
|
||||
|
||||
private final static String PROPERTIES_RESOURCE_LOCATION = "META-INF/sentinel-datasource.properties";
|
||||
|
||||
private final static String ALL_PROPERTIES_RESOURCES_LOCATION = CLASSPATH_ALL_URL_PREFIX
|
||||
+ PROPERTIES_RESOURCE_LOCATION;
|
||||
|
||||
private final static ConcurrentMap<String, Class<? extends ReadableDataSource>> dataSourceClassesCache
|
||||
= new ConcurrentHashMap<String, Class<? extends ReadableDataSource>>(
|
||||
4);
|
||||
|
||||
static void loadAllDataSourceClassesCache() {
|
||||
Map<String, Class<? extends ReadableDataSource>> dataSourceClassesMap = loadAllDataSourceClassesCache(
|
||||
ALL_PROPERTIES_RESOURCES_LOCATION);
|
||||
|
||||
dataSourceClassesCache.putAll(dataSourceClassesMap);
|
||||
}
|
||||
|
||||
static Map<String, Class<? extends ReadableDataSource>> loadAllDataSourceClassesCache(
|
||||
String resourcesLocation) {
|
||||
|
||||
Map<String, Class<? extends ReadableDataSource>> dataSourcesMap
|
||||
= new HashMap<String, Class<? extends ReadableDataSource>>(
|
||||
4);
|
||||
|
||||
ClassLoader classLoader = DataSourceLoader.class.getClassLoader();
|
||||
|
||||
ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
|
||||
|
||||
try {
|
||||
|
||||
Resource[] resources = resolver.getResources(resourcesLocation);
|
||||
|
||||
for (Resource resource : resources) {
|
||||
if (resource.exists()) {
|
||||
Properties properties = PropertiesLoaderUtils
|
||||
.loadProperties(resource);
|
||||
for (Map.Entry<Object, Object> entry : properties.entrySet()) {
|
||||
|
||||
String type = (String)entry.getKey();
|
||||
String className = (String)entry.getValue();
|
||||
|
||||
if (!ClassUtils.isPresent(className, classLoader)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug(
|
||||
"Sentinel DataSource implementation [ type : "
|
||||
+ type + ": , class : " + className
|
||||
+ " , url : " + resource.getURL()
|
||||
+ "] was not present in current classpath , "
|
||||
+ "thus loading will be ignored , please add dependency if required !");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
Assert.isTrue(!dataSourcesMap.containsKey(type),
|
||||
"The duplicated type[" + type
|
||||
+ "] of SentinelDataSource were found in "
|
||||
+ "resource [" + resource.getURL() + "]");
|
||||
|
||||
Class<?> dataSourceClass = ClassUtils.resolveClassName(className,
|
||||
classLoader);
|
||||
Assert.isAssignable(ReadableDataSource.class, dataSourceClass);
|
||||
|
||||
dataSourcesMap.put(type,
|
||||
(Class<? extends ReadableDataSource>)dataSourceClass);
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Sentinel DataSource implementation [ type : "
|
||||
+ type + ": , class : " + className
|
||||
+ "] was loaded.");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} catch (IOException e) {
|
||||
if (logger.isErrorEnabled()) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
return dataSourcesMap;
|
||||
}
|
||||
|
||||
public static Class<? extends ReadableDataSource> loadClass(String type)
|
||||
throws IllegalArgumentException {
|
||||
|
||||
Class<? extends ReadableDataSource> dataSourceClass = dataSourceClassesCache.get(type);
|
||||
|
||||
if (dataSourceClass == null) {
|
||||
if (dataSourceClassesCache.isEmpty()) {
|
||||
loadAllDataSourceClassesCache();
|
||||
dataSourceClass = dataSourceClassesCache.get(type);
|
||||
}
|
||||
}
|
||||
|
||||
if (dataSourceClass == null) {
|
||||
throw new IllegalArgumentException(
|
||||
"Sentinel DataSource implementation [ type : " + type
|
||||
+ " ] can't be found!");
|
||||
}
|
||||
|
||||
return dataSourceClass;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,272 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.alibaba.sentinel.datasource;
|
||||
|
||||
import java.beans.PropertyDescriptor;
|
||||
import java.lang.reflect.Field;
|
||||
import java.lang.reflect.Modifier;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.Converter;
|
||||
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
|
||||
import com.alibaba.csp.sentinel.property.SentinelProperty;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRule;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRuleManager;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.PropertyValues;
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.config.InstantiationAwareBeanPostProcessorAdapter;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.beans.factory.support.MergedBeanDefinitionPostProcessor;
|
||||
import org.springframework.beans.factory.support.RootBeanDefinition;
|
||||
import org.springframework.boot.context.event.ApplicationStartedEvent;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.annotation.SentinelDataSource;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.util.PropertySourcesUtils;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.core.env.ConfigurableEnvironment;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import static org.springframework.core.annotation.AnnotationUtils.getAnnotation;
|
||||
|
||||
/**
|
||||
* {@link SentinelDataSource @SentinelDataSource} Post Processor
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see ReadableDataSource
|
||||
* @see SentinelDataSource
|
||||
*/
|
||||
public class SentinelDataSourcePostProcessor
|
||||
extends InstantiationAwareBeanPostProcessorAdapter
|
||||
implements MergedBeanDefinitionPostProcessor {
|
||||
|
||||
private static final Logger logger = LoggerFactory
|
||||
.getLogger(SentinelDataSourcePostProcessor.class);
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
@Autowired
|
||||
private ConfigurableEnvironment environment;
|
||||
|
||||
private final Map<String, List<SentinelDataSourceField>> dataSourceFieldCache = new ConcurrentHashMap<>(
|
||||
64);
|
||||
|
||||
@Override
|
||||
public void postProcessMergedBeanDefinition(RootBeanDefinition beanDefinition,
|
||||
Class<?> beanType, String beanName) {
|
||||
// find all fields using by @SentinelDataSource annotation
|
||||
ReflectionUtils.doWithFields(beanType, new ReflectionUtils.FieldCallback() {
|
||||
@Override
|
||||
public void doWith(Field field)
|
||||
throws IllegalArgumentException, IllegalAccessException {
|
||||
SentinelDataSource annotation = getAnnotation(field,
|
||||
SentinelDataSource.class);
|
||||
if (annotation != null) {
|
||||
if (Modifier.isStatic(field.getModifiers())) {
|
||||
if (logger.isWarnEnabled()) {
|
||||
logger.warn(
|
||||
"@SentinelDataSource annotation is not supported on static fields: "
|
||||
+ field);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (dataSourceFieldCache.containsKey(beanName)) {
|
||||
dataSourceFieldCache.get(beanName)
|
||||
.add(new SentinelDataSourceField(annotation, field));
|
||||
} else {
|
||||
List<SentinelDataSourceField> list = new ArrayList<>();
|
||||
list.add(new SentinelDataSourceField(annotation, field));
|
||||
dataSourceFieldCache.put(beanName, list);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public PropertyValues postProcessPropertyValues(PropertyValues pvs,
|
||||
PropertyDescriptor[] pds, Object bean, String beanName)
|
||||
throws BeanCreationException {
|
||||
if (dataSourceFieldCache.containsKey(beanName)) {
|
||||
List<SentinelDataSourceField> sentinelDataSourceFields = dataSourceFieldCache
|
||||
.get(beanName);
|
||||
sentinelDataSourceFields.forEach(sentinelDataSourceField -> {
|
||||
try {
|
||||
// construct DataSource field annotated by @SentinelDataSource
|
||||
Field field = sentinelDataSourceField.getField();
|
||||
ReflectionUtils.makeAccessible(field);
|
||||
String dataSourceBeanName = constructDataSource(
|
||||
sentinelDataSourceField.getSentinelDataSource());
|
||||
field.set(bean, applicationContext.getBean(dataSourceBeanName));
|
||||
} catch (IllegalAccessException e) {
|
||||
e.printStackTrace();
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
});
|
||||
}
|
||||
return pvs;
|
||||
}
|
||||
|
||||
private String constructDataSource(SentinelDataSource annotation) {
|
||||
String prefix = annotation.value();
|
||||
if (StringUtils.isEmpty(prefix)) {
|
||||
prefix = SentinelDataSourceConstants.PROPERTY_DATASOURCE_PREFIX;
|
||||
}
|
||||
Map<String, Object> propertyMap = PropertySourcesUtils
|
||||
.getSubProperties(environment.getPropertySources(), prefix);
|
||||
String alias = propertyMap.get("type").toString();
|
||||
Class dataSourceClass = DataSourceLoader.loadClass(alias);
|
||||
|
||||
String beanName = StringUtils.isEmpty(annotation.name())
|
||||
? StringUtils.uncapitalize(dataSourceClass.getSimpleName()) + "_" + prefix
|
||||
: annotation.name();
|
||||
if (applicationContext.containsBean(beanName)) {
|
||||
return beanName;
|
||||
}
|
||||
|
||||
Class targetClass = null;
|
||||
// if alias exists in SentinelDataSourceRegistry, wired properties into
|
||||
// FactoryBean
|
||||
if (SentinelDataSourceRegistry.checkFactoryBean(alias)) {
|
||||
targetClass = SentinelDataSourceRegistry.getFactoryBean(alias);
|
||||
} else {
|
||||
// if alias not exists in SentinelDataSourceRegistry, wired properties into
|
||||
// raw class
|
||||
targetClass = dataSourceClass;
|
||||
}
|
||||
|
||||
registerDataSource(beanName, targetClass, propertyMap);
|
||||
|
||||
return beanName;
|
||||
}
|
||||
|
||||
private void registerDataSource(String beanName, Class targetClass,
|
||||
Map<String, Object> propertyMap) {
|
||||
BeanDefinitionBuilder builder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(targetClass);
|
||||
for (String propertyName : propertyMap.keySet()) {
|
||||
Field field = ReflectionUtils.findField(targetClass, propertyName);
|
||||
if (field != null) {
|
||||
if (field.getType().isAssignableFrom(Converter.class)) {
|
||||
// Converter get from ApplicationContext
|
||||
builder.addPropertyReference(propertyName,
|
||||
propertyMap.get(propertyName).toString());
|
||||
} else {
|
||||
// wired properties
|
||||
builder.addPropertyValue(propertyName, propertyMap.get(propertyName));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory)applicationContext
|
||||
.getAutowireCapableBeanFactory();
|
||||
beanFactory.registerBeanDefinition(beanName, builder.getBeanDefinition());
|
||||
}
|
||||
|
||||
@EventListener(classes = ApplicationStartedEvent.class)
|
||||
public void appStartedListener(ApplicationStartedEvent event) throws Exception {
|
||||
logger.info("[Sentinel Starter] Start to find ReadableDataSource");
|
||||
Map<String, ReadableDataSource> dataSourceMap = event.getApplicationContext().getBeansOfType(
|
||||
ReadableDataSource.class);
|
||||
if (dataSourceMap.size() == 1) {
|
||||
logger.info("[Sentinel Starter] There exists only one ReadableDataSource named {}, start to load rules",
|
||||
dataSourceMap.keySet().iterator().next());
|
||||
ReadableDataSource dataSource = dataSourceMap.values().iterator().next();
|
||||
Object ruleConfig = dataSource.loadConfig();
|
||||
SentinelProperty sentinelProperty = dataSource.getProperty();
|
||||
Integer rulesNum;
|
||||
if ((rulesNum = checkRuleType(ruleConfig, FlowRule.class)) > 0) {
|
||||
FlowRuleManager.register2Property(sentinelProperty);
|
||||
logger.info("[Sentinel Starter] load {} flow rules", rulesNum);
|
||||
}
|
||||
if ((rulesNum = checkRuleType(ruleConfig, DegradeRule.class)) > 0) {
|
||||
DegradeRuleManager.register2Property(sentinelProperty);
|
||||
logger.info("[Sentinel Starter] load {} degrade rules", rulesNum);
|
||||
}
|
||||
if ((rulesNum = checkRuleType(ruleConfig, SystemRule.class)) > 0) {
|
||||
SystemRuleManager.register2Property(sentinelProperty);
|
||||
logger.info("[Sentinel Starter] load {} system rules", rulesNum);
|
||||
}
|
||||
if ((rulesNum = checkRuleType(ruleConfig, AuthorityRule.class)) > 0) {
|
||||
AuthorityRuleManager.register2Property(sentinelProperty);
|
||||
logger.info("[Sentinel Starter] load {} authority rules", rulesNum);
|
||||
}
|
||||
} else if (dataSourceMap.size() > 1) {
|
||||
logger.warn(
|
||||
"[Sentinel Starter] There exists more than one ReadableDataSource, can not choose which one to load");
|
||||
} else {
|
||||
logger.warn(
|
||||
"[Sentinel Starter] No ReadableDataSource exists");
|
||||
}
|
||||
}
|
||||
|
||||
private Integer checkRuleType(Object ruleConfig, Class type) {
|
||||
if (ruleConfig.getClass() == type) {
|
||||
return 1;
|
||||
} else if (ruleConfig instanceof List) {
|
||||
List ruleList = (List)ruleConfig;
|
||||
if (ruleList.stream().filter(rule -> rule.getClass() == type).toArray().length == ruleList.size()) {
|
||||
return ruleList.size();
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
class SentinelDataSourceField {
|
||||
private SentinelDataSource sentinelDataSource;
|
||||
private Field field;
|
||||
|
||||
public SentinelDataSourceField(SentinelDataSource sentinelDataSource,
|
||||
Field field) {
|
||||
this.sentinelDataSource = sentinelDataSource;
|
||||
this.field = field;
|
||||
}
|
||||
|
||||
public SentinelDataSource getSentinelDataSource() {
|
||||
return sentinelDataSource;
|
||||
}
|
||||
|
||||
public void setSentinelDataSource(SentinelDataSource sentinelDataSource) {
|
||||
this.sentinelDataSource = sentinelDataSource;
|
||||
}
|
||||
|
||||
public Field getField() {
|
||||
return field;
|
||||
}
|
||||
|
||||
public void setField(Field field) {
|
||||
this.field = field;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,68 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.alibaba.sentinel.datasource;
|
||||
|
||||
import java.util.HashMap;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
|
||||
|
||||
import org.springframework.beans.factory.FactoryBean;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.ApolloDataSourceFactoryBean;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.FileRefreshableDataSourceFactoryBean;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.NacosDataSourceFactoryBean;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.ZookeeperDataSourceFactoryBean;
|
||||
|
||||
/**
|
||||
* Registry to save DataSource FactoryBean
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see ReadableDataSource
|
||||
* @see FileRefreshableDataSourceFactoryBean
|
||||
* @see ZookeeperDataSourceFactoryBean
|
||||
* @see NacosDataSourceFactoryBean
|
||||
* @see ApolloDataSourceFactoryBean
|
||||
*/
|
||||
public class SentinelDataSourceRegistry {
|
||||
|
||||
private static HashMap<String, Class<? extends FactoryBean>> cache = new HashMap<>(
|
||||
32);
|
||||
|
||||
static {
|
||||
SentinelDataSourceRegistry.registerFactoryBean("file",
|
||||
FileRefreshableDataSourceFactoryBean.class);
|
||||
SentinelDataSourceRegistry.registerFactoryBean("zk",
|
||||
ZookeeperDataSourceFactoryBean.class);
|
||||
SentinelDataSourceRegistry.registerFactoryBean("nacos",
|
||||
NacosDataSourceFactoryBean.class);
|
||||
SentinelDataSourceRegistry.registerFactoryBean("apollo",
|
||||
ApolloDataSourceFactoryBean.class);
|
||||
}
|
||||
|
||||
public static synchronized void registerFactoryBean(String alias,
|
||||
Class<? extends FactoryBean> clazz) {
|
||||
cache.put(alias, clazz);
|
||||
}
|
||||
|
||||
public static Class<? extends FactoryBean> getFactoryBean(String alias) {
|
||||
return cache.get(alias);
|
||||
}
|
||||
|
||||
public static boolean checkFactoryBean(String alias) {
|
||||
return cache.containsKey(alias);
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,50 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.alibaba.sentinel.datasource.annotation;
|
||||
|
||||
import java.lang.annotation.Documented;
|
||||
import java.lang.annotation.ElementType;
|
||||
import java.lang.annotation.Retention;
|
||||
import java.lang.annotation.RetentionPolicy;
|
||||
import java.lang.annotation.Target;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
|
||||
|
||||
import org.springframework.core.annotation.AliasFor;
|
||||
|
||||
/**
|
||||
* An annotation to inject {@link ReadableDataSource} instance
|
||||
* into a Spring Bean. The Properties of DataSource bean get from config files with
|
||||
* specific prefix.
|
||||
*
|
||||
* <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see ReadableDataSource
|
||||
*/
|
||||
@Target({ElementType.FIELD})
|
||||
@Retention(RetentionPolicy.RUNTIME)
|
||||
@Documented
|
||||
public @interface SentinelDataSource {
|
||||
|
||||
@AliasFor("prefix")
|
||||
String value() default "";
|
||||
|
||||
@AliasFor("value")
|
||||
String prefix() default "";
|
||||
|
||||
String name() default ""; // spring bean name
|
||||
|
||||
}
|
||||
@ -0,0 +1,41 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
|
||||
/**
|
||||
* Abstract class Using by {@link DataSourcePropertiesConfiguration}
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class AbstractDataSourceProperties {
|
||||
|
||||
private String dataType = "json";
|
||||
private String converterClass;
|
||||
@JsonIgnore
|
||||
private final String factoryBeanName;
|
||||
|
||||
public AbstractDataSourceProperties(String factoryBeanName) {
|
||||
this.factoryBeanName = factoryBeanName;
|
||||
}
|
||||
|
||||
public String getDataType() {
|
||||
return dataType;
|
||||
}
|
||||
|
||||
public void setDataType(String dataType) {
|
||||
this.dataType = dataType;
|
||||
}
|
||||
|
||||
public String getConverterClass() {
|
||||
return converterClass;
|
||||
}
|
||||
|
||||
public void setConverterClass(String converterClass) {
|
||||
this.converterClass = converterClass;
|
||||
}
|
||||
|
||||
public String getFactoryBeanName() {
|
||||
return factoryBeanName;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.ApolloDataSourceFactoryBean;
|
||||
|
||||
/**
|
||||
* Apollo Properties class Using by {@link DataSourcePropertiesConfiguration} and
|
||||
* {@link ApolloDataSourceFactoryBean}
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ApolloDataSourceProperties extends AbstractDataSourceProperties {
|
||||
|
||||
private String namespaceName;
|
||||
private String flowRulesKey;
|
||||
private String defaultFlowRuleValue;
|
||||
|
||||
public ApolloDataSourceProperties() {
|
||||
super(ApolloDataSourceFactoryBean.class.getName());
|
||||
}
|
||||
|
||||
public String getNamespaceName() {
|
||||
return namespaceName;
|
||||
}
|
||||
|
||||
public void setNamespaceName(String namespaceName) {
|
||||
this.namespaceName = namespaceName;
|
||||
}
|
||||
|
||||
public String getFlowRulesKey() {
|
||||
return flowRulesKey;
|
||||
}
|
||||
|
||||
public void setFlowRulesKey(String flowRulesKey) {
|
||||
this.flowRulesKey = flowRulesKey;
|
||||
}
|
||||
|
||||
public String getDefaultFlowRuleValue() {
|
||||
return defaultFlowRuleValue;
|
||||
}
|
||||
|
||||
public void setDefaultFlowRuleValue(String defaultFlowRuleValue) {
|
||||
this.defaultFlowRuleValue = defaultFlowRuleValue;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,79 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
|
||||
/**
|
||||
* Using By ConfigurationProperties.
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see NacosDataSourceProperties
|
||||
* @see ApolloDataSourceProperties
|
||||
* @see ZookeeperDataSourceProperties
|
||||
* @see FileDataSourceProperties
|
||||
*/
|
||||
public class DataSourcePropertiesConfiguration {
|
||||
|
||||
private FileDataSourceProperties file;
|
||||
|
||||
private NacosDataSourceProperties nacos;
|
||||
|
||||
private ZookeeperDataSourceProperties zk;
|
||||
|
||||
private ApolloDataSourceProperties apollo;
|
||||
|
||||
public FileDataSourceProperties getFile() {
|
||||
return file;
|
||||
}
|
||||
|
||||
public void setFile(FileDataSourceProperties file) {
|
||||
this.file = file;
|
||||
}
|
||||
|
||||
public NacosDataSourceProperties getNacos() {
|
||||
return nacos;
|
||||
}
|
||||
|
||||
public void setNacos(NacosDataSourceProperties nacos) {
|
||||
this.nacos = nacos;
|
||||
}
|
||||
|
||||
public ZookeeperDataSourceProperties getZk() {
|
||||
return zk;
|
||||
}
|
||||
|
||||
public void setZk(ZookeeperDataSourceProperties zk) {
|
||||
this.zk = zk;
|
||||
}
|
||||
|
||||
public ApolloDataSourceProperties getApollo() {
|
||||
return apollo;
|
||||
}
|
||||
|
||||
public void setApollo(ApolloDataSourceProperties apollo) {
|
||||
this.apollo = apollo;
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
public List<String> getInvalidField() {
|
||||
return Arrays.stream(this.getClass().getDeclaredFields()).map(field -> {
|
||||
try {
|
||||
if (!ObjectUtils.isEmpty(field.get(this))) {
|
||||
return field.getName();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
catch (IllegalAccessException e) {
|
||||
// won't happen
|
||||
}
|
||||
return null;
|
||||
}).filter(Objects::nonNull).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,53 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.FileRefreshableDataSourceFactoryBean;
|
||||
|
||||
/**
|
||||
* File Properties class Using by {@link DataSourcePropertiesConfiguration} and
|
||||
* {@link FileRefreshableDataSourceFactoryBean}
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class FileDataSourceProperties extends AbstractDataSourceProperties {
|
||||
|
||||
private String file;
|
||||
private String charset = "utf-8";
|
||||
private long recommendRefreshMs = 3000L;
|
||||
private int bufSize = 1024 * 1024;
|
||||
|
||||
public FileDataSourceProperties() {
|
||||
super(FileRefreshableDataSourceFactoryBean.class.getName());
|
||||
}
|
||||
|
||||
public String getFile() {
|
||||
return file;
|
||||
}
|
||||
|
||||
public void setFile(String file) {
|
||||
this.file = file;
|
||||
}
|
||||
|
||||
public String getCharset() {
|
||||
return charset;
|
||||
}
|
||||
|
||||
public void setCharset(String charset) {
|
||||
this.charset = charset;
|
||||
}
|
||||
|
||||
public long getRecommendRefreshMs() {
|
||||
return recommendRefreshMs;
|
||||
}
|
||||
|
||||
public void setRecommendRefreshMs(long recommendRefreshMs) {
|
||||
this.recommendRefreshMs = recommendRefreshMs;
|
||||
}
|
||||
|
||||
public int getBufSize() {
|
||||
return bufSize;
|
||||
}
|
||||
|
||||
public void setBufSize(int bufSize) {
|
||||
this.bufSize = bufSize;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.NacosDataSourceFactoryBean;
|
||||
|
||||
/**
|
||||
* Nacos Properties class Using by {@link DataSourcePropertiesConfiguration} and
|
||||
* {@link NacosDataSourceFactoryBean}
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class NacosDataSourceProperties extends AbstractDataSourceProperties {
|
||||
|
||||
private String serverAddr;
|
||||
private String groupId;
|
||||
private String dataId;
|
||||
|
||||
public NacosDataSourceProperties() {
|
||||
super(NacosDataSourceFactoryBean.class.getName());
|
||||
}
|
||||
|
||||
public String getServerAddr() {
|
||||
return serverAddr;
|
||||
}
|
||||
|
||||
public void setServerAddr(String serverAddr) {
|
||||
this.serverAddr = serverAddr;
|
||||
}
|
||||
|
||||
public String getGroupId() {
|
||||
return groupId;
|
||||
}
|
||||
|
||||
public void setGroupId(String groupId) {
|
||||
this.groupId = groupId;
|
||||
}
|
||||
|
||||
public String getDataId() {
|
||||
return dataId;
|
||||
}
|
||||
|
||||
public void setDataId(String dataId) {
|
||||
this.dataId = dataId;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,56 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.config;
|
||||
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.factorybean.ZookeeperDataSourceFactoryBean;
|
||||
|
||||
/**
|
||||
* Zookeeper Properties class Using by {@link DataSourcePropertiesConfiguration} and
|
||||
* {@link ZookeeperDataSourceFactoryBean}
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ZookeeperDataSourceProperties extends AbstractDataSourceProperties {
|
||||
|
||||
public ZookeeperDataSourceProperties() {
|
||||
super(ZookeeperDataSourceFactoryBean.class.getName());
|
||||
}
|
||||
|
||||
private String serverAddr;
|
||||
|
||||
private String path;
|
||||
|
||||
private String groupId;
|
||||
|
||||
private String dataId;
|
||||
|
||||
public String getServerAddr() {
|
||||
return serverAddr;
|
||||
}
|
||||
|
||||
public void setServerAddr(String serverAddr) {
|
||||
this.serverAddr = serverAddr;
|
||||
}
|
||||
|
||||
public String getPath() {
|
||||
return path;
|
||||
}
|
||||
|
||||
public void setPath(String path) {
|
||||
this.path = path;
|
||||
}
|
||||
|
||||
public String getGroupId() {
|
||||
return groupId;
|
||||
}
|
||||
|
||||
public void setGroupId(String groupId) {
|
||||
this.groupId = groupId;
|
||||
}
|
||||
|
||||
public String getDataId() {
|
||||
return dataId;
|
||||
}
|
||||
|
||||
public void setDataId(String dataId) {
|
||||
this.dataId = dataId;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,157 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.converter;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.Converter;
|
||||
import com.alibaba.csp.sentinel.slots.block.AbstractRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRule;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
/**
|
||||
* Convert sentinel rules for json array Using strict mode to parse json
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see FlowRule
|
||||
* @see DegradeRule
|
||||
* @see SystemRule
|
||||
* @see AuthorityRule
|
||||
* @see ObjectMapper
|
||||
*/
|
||||
public class JsonConverter implements Converter<String, List<AbstractRule>> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(JsonConverter.class);
|
||||
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public JsonConverter(ObjectMapper objectMapper) {
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<AbstractRule> convert(String source) {
|
||||
List<AbstractRule> ruleList = new ArrayList<>();
|
||||
if (StringUtils.isEmpty(source)) {
|
||||
logger.info(
|
||||
"Sentinel JsonConverter can not convert rules because source is empty");
|
||||
return ruleList;
|
||||
}
|
||||
try {
|
||||
List jsonArray = objectMapper.readValue(source,
|
||||
new TypeReference<List<HashMap>>() {
|
||||
});
|
||||
jsonArray.stream().forEach(obj -> {
|
||||
|
||||
String itemJson = null;
|
||||
try {
|
||||
itemJson = objectMapper.writeValueAsString(obj);
|
||||
}
|
||||
catch (JsonProcessingException e) {
|
||||
// won't be happen
|
||||
}
|
||||
|
||||
List<AbstractRule> rules = Arrays.asList(convertFlowRule(itemJson),
|
||||
convertDegradeRule(itemJson), convertSystemRule(itemJson),
|
||||
convertAuthorityRule(itemJson));
|
||||
|
||||
List<AbstractRule> convertRuleList = rules.stream()
|
||||
.filter(rule -> !ObjectUtils.isEmpty(rule))
|
||||
.collect(Collectors.toList());
|
||||
|
||||
if (convertRuleList.size() == 0) {
|
||||
logger.warn(
|
||||
"Sentinel JsonConverter can not convert {} to any rules, ignore",
|
||||
itemJson);
|
||||
}
|
||||
else if (convertRuleList.size() > 1) {
|
||||
logger.warn(
|
||||
"Sentinel JsonConverter convert {} and match multi rules, ignore",
|
||||
itemJson);
|
||||
}
|
||||
else {
|
||||
ruleList.add(convertRuleList.get(0));
|
||||
}
|
||||
|
||||
});
|
||||
if (jsonArray.size() != ruleList.size()) {
|
||||
logger.warn(
|
||||
"Sentinel JsonConverter Source list size is not equals to Target List, maybe a "
|
||||
+ "part of json is invalid. Source List: " + jsonArray
|
||||
+ ", Target List: " + ruleList);
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Sentinel JsonConverter convert error: " + e.getMessage());
|
||||
throw new RuntimeException(
|
||||
"Sentinel JsonConverter convert error: " + e.getMessage(), e);
|
||||
}
|
||||
logger.info("Sentinel JsonConverter convert {} rules: {}", ruleList.size(),
|
||||
ruleList);
|
||||
return ruleList;
|
||||
}
|
||||
|
||||
private FlowRule convertFlowRule(String json) {
|
||||
try {
|
||||
FlowRule rule = objectMapper.readValue(json, FlowRule.class);
|
||||
if (FlowRuleManager.isValidRule(rule)) {
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private DegradeRule convertDegradeRule(String json) {
|
||||
try {
|
||||
DegradeRule rule = objectMapper.readValue(json, DegradeRule.class);
|
||||
if (DegradeRuleManager.isValidRule(rule)) {
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private SystemRule convertSystemRule(String json) {
|
||||
SystemRule rule = null;
|
||||
try {
|
||||
rule = objectMapper.readValue(json, SystemRule.class);
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
|
||||
private AuthorityRule convertAuthorityRule(String json) {
|
||||
AuthorityRule rule = null;
|
||||
try {
|
||||
rule = objectMapper.readValue(json, AuthorityRule.class);
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,157 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.datasource.converter;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.Converter;
|
||||
import com.alibaba.csp.sentinel.slots.block.AbstractRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRule;
|
||||
|
||||
import com.fasterxml.jackson.core.JsonProcessingException;
|
||||
import com.fasterxml.jackson.core.type.TypeReference;
|
||||
import com.fasterxml.jackson.dataformat.xml.XmlMapper;
|
||||
|
||||
/**
|
||||
* Convert sentinel rules for xml array Using strict mode to parse xml
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see FlowRule
|
||||
* @see DegradeRule
|
||||
* @see SystemRule
|
||||
* @see AuthorityRule
|
||||
* @see XmlMapper
|
||||
*/
|
||||
public class XmlConverter implements Converter<String, List<AbstractRule>> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(XmlConverter.class);
|
||||
|
||||
private final XmlMapper xmlMapper;
|
||||
|
||||
public XmlConverter(XmlMapper xmlMapper) {
|
||||
this.xmlMapper = xmlMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<AbstractRule> convert(String source) {
|
||||
List<AbstractRule> ruleList = new ArrayList<>();
|
||||
if (StringUtils.isEmpty(source)) {
|
||||
logger.info(
|
||||
"Sentinel XmlConverter can not convert rules because source is empty");
|
||||
return ruleList;
|
||||
}
|
||||
try {
|
||||
List xmlArray = xmlMapper.readValue(source,
|
||||
new TypeReference<List<HashMap>>() {
|
||||
});
|
||||
xmlArray.stream().forEach(obj -> {
|
||||
|
||||
String itemXml = null;
|
||||
try {
|
||||
itemXml = xmlMapper.writeValueAsString(obj);
|
||||
}
|
||||
catch (JsonProcessingException e) {
|
||||
// won't be happen
|
||||
}
|
||||
|
||||
List<AbstractRule> rules = Arrays.asList(convertFlowRule(itemXml),
|
||||
convertDegradeRule(itemXml), convertSystemRule(itemXml),
|
||||
convertAuthorityRule(itemXml));
|
||||
|
||||
List<AbstractRule> convertRuleList = rules.stream()
|
||||
.filter(rule -> !ObjectUtils.isEmpty(rule))
|
||||
.collect(Collectors.toList());
|
||||
|
||||
if (convertRuleList.size() == 0) {
|
||||
logger.warn(
|
||||
"Sentinel XmlConverter can not convert {} to any rules, ignore",
|
||||
itemXml);
|
||||
}
|
||||
else if (convertRuleList.size() > 1) {
|
||||
logger.warn(
|
||||
"Sentinel XmlConverter convert {} and match multi rules, ignore",
|
||||
itemXml);
|
||||
}
|
||||
else {
|
||||
ruleList.add(convertRuleList.get(0));
|
||||
}
|
||||
|
||||
});
|
||||
if (xmlArray.size() != ruleList.size()) {
|
||||
logger.warn(
|
||||
"Sentinel XmlConverter Source list size is not equals to Target List, maybe a "
|
||||
+ "part of xml is invalid. Source List: " + xmlArray
|
||||
+ ", Target List: " + ruleList);
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Sentinel XmlConverter convert error: " + e.getMessage());
|
||||
throw new RuntimeException(
|
||||
"Sentinel XmlConverter convert error: " + e.getMessage(), e);
|
||||
}
|
||||
logger.info("Sentinel XmlConverter convert {} rules: {}", ruleList.size(),
|
||||
ruleList);
|
||||
return ruleList;
|
||||
}
|
||||
|
||||
private FlowRule convertFlowRule(String xml) {
|
||||
try {
|
||||
FlowRule rule = xmlMapper.readValue(xml, FlowRule.class);
|
||||
if (FlowRuleManager.isValidRule(rule)) {
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private DegradeRule convertDegradeRule(String xml) {
|
||||
try {
|
||||
DegradeRule rule = xmlMapper.readValue(xml, DegradeRule.class);
|
||||
if (DegradeRuleManager.isValidRule(rule)) {
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private SystemRule convertSystemRule(String xml) {
|
||||
SystemRule rule = null;
|
||||
try {
|
||||
rule = xmlMapper.readValue(xml, SystemRule.class);
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
|
||||
private AuthorityRule convertAuthorityRule(String xml) {
|
||||
AuthorityRule rule = null;
|
||||
try {
|
||||
rule = xmlMapper.readValue(xml, AuthorityRule.class);
|
||||
}
|
||||
catch (Exception e) {
|
||||
// ignore
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,75 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.alibaba.sentinel.datasource.util;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Properties;
|
||||
|
||||
import org.springframework.core.env.EnumerablePropertySource;
|
||||
import org.springframework.core.env.PropertySource;
|
||||
import org.springframework.core.env.PropertySources;
|
||||
|
||||
/**
|
||||
* {@link PropertySources} Utilities
|
||||
*
|
||||
* @author <a href="mailto:mercyblitz@gmail.com">Mercy</a>
|
||||
*/
|
||||
public abstract class PropertySourcesUtils {
|
||||
|
||||
/**
|
||||
* Get Sub {@link Properties}
|
||||
*
|
||||
* @param propertySources {@link PropertySource} Iterable
|
||||
* @param prefix the prefix of property name
|
||||
* @return Map
|
||||
* @see Properties
|
||||
*/
|
||||
public static Map<String, Object> getSubProperties(Iterable<PropertySource<?>> propertySources, String prefix) {
|
||||
|
||||
Map<String, Object> subProperties = new LinkedHashMap<String, Object>();
|
||||
|
||||
String normalizedPrefix = normalizePrefix(prefix);
|
||||
|
||||
for (PropertySource<?> source : propertySources) {
|
||||
if (source instanceof EnumerablePropertySource) {
|
||||
for (String name : ((EnumerablePropertySource<?>)source).getPropertyNames()) {
|
||||
if (!subProperties.containsKey(name) && name.startsWith(normalizedPrefix)) {
|
||||
String subName = name.substring(normalizedPrefix.length());
|
||||
if (!subProperties.containsKey(subName)) { // take first one
|
||||
Object value = source.getProperty(name);
|
||||
subProperties.put(subName, value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return subProperties;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize the prefix
|
||||
*
|
||||
* @param prefix the prefix
|
||||
* @return the prefix
|
||||
*/
|
||||
public static String normalizePrefix(String prefix) {
|
||||
return prefix.endsWith(".") ? prefix : prefix + ".";
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,315 @@
|
||||
package org.springframework.cloud.alibaba.sentinel.custom;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.lang.reflect.Field;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
|
||||
import org.springframework.boot.context.event.ApplicationStartedEvent;
|
||||
import org.springframework.cloud.alibaba.sentinel.SentinelProperties;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.config.AbstractDataSourceProperties;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.converter.JsonConverter;
|
||||
import org.springframework.cloud.alibaba.sentinel.datasource.converter.XmlConverter;
|
||||
import org.springframework.context.event.EventListener;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
import org.springframework.util.ResourceUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import com.alibaba.csp.sentinel.datasource.ReadableDataSource;
|
||||
import com.alibaba.csp.sentinel.property.SentinelProperty;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.authority.AuthorityRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.degrade.DegradeRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRule;
|
||||
import com.alibaba.csp.sentinel.slots.block.flow.FlowRuleManager;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRule;
|
||||
import com.alibaba.csp.sentinel.slots.system.SystemRuleManager;
|
||||
|
||||
/**
|
||||
* Sentinel {@link ReadableDataSource} Handler Handle the configurations of
|
||||
* 'spring.cloud.sentinel.datasource'
|
||||
*
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
* @see SentinelProperties#datasource
|
||||
* @see JsonConverter
|
||||
* @see XmlConverter
|
||||
*/
|
||||
public class SentinelDataSourceHandler {
|
||||
|
||||
private static final Logger logger = LoggerFactory
|
||||
.getLogger(SentinelDataSourceHandler.class);
|
||||
|
||||
private List<String> dataTypeList = Arrays.asList("json", "xml");
|
||||
|
||||
private List<Class> rulesList = Arrays.asList(FlowRule.class, DegradeRule.class,
|
||||
SystemRule.class, AuthorityRule.class);
|
||||
|
||||
private List<String> dataSourceBeanNameList = Collections
|
||||
.synchronizedList(new ArrayList<>());
|
||||
|
||||
private final String DATATYPE_FIELD = "dataType";
|
||||
private final String CUSTOM_DATATYPE = "custom";
|
||||
private final String CONVERTERCLASS_FIELD = "converterClass";
|
||||
|
||||
@Autowired
|
||||
private SentinelProperties sentinelProperties;
|
||||
|
||||
@EventListener(classes = ApplicationStartedEvent.class)
|
||||
public void buildDataSource(ApplicationStartedEvent event) throws Exception {
|
||||
|
||||
DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) event
|
||||
.getApplicationContext().getAutowireCapableBeanFactory();
|
||||
|
||||
sentinelProperties.getDatasource()
|
||||
.forEach((dataSourceName, dataSourceProperties) -> {
|
||||
if (dataSourceProperties.getInvalidField().size() != 1) {
|
||||
logger.error("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " multi datasource active and won't loaded: "
|
||||
+ dataSourceProperties.getInvalidField());
|
||||
return;
|
||||
}
|
||||
Optional.ofNullable(dataSourceProperties.getFile())
|
||||
.ifPresent(file -> {
|
||||
try {
|
||||
dataSourceProperties.getFile().setFile(ResourceUtils
|
||||
.getFile(StringUtils.trimAllWhitespace(
|
||||
dataSourceProperties.getFile()
|
||||
.getFile()))
|
||||
.getAbsolutePath());
|
||||
}
|
||||
catch (IOException e) {
|
||||
logger.error("[Sentinel Starter] DataSource "
|
||||
+ dataSourceName + " handle file error: "
|
||||
+ e.getMessage());
|
||||
throw new RuntimeException(
|
||||
"[Sentinel Starter] DataSource "
|
||||
+ dataSourceName
|
||||
+ " handle file error: "
|
||||
+ e.getMessage(),
|
||||
e);
|
||||
}
|
||||
registerBean(beanFactory, file,
|
||||
dataSourceName + "-sentinel-file-datasource");
|
||||
});
|
||||
Optional.ofNullable(dataSourceProperties.getNacos())
|
||||
.ifPresent(nacos -> {
|
||||
registerBean(beanFactory, nacos,
|
||||
dataSourceName + "-sentinel-nacos-datasource");
|
||||
});
|
||||
Optional.ofNullable(dataSourceProperties.getApollo())
|
||||
.ifPresent(apollo -> {
|
||||
registerBean(beanFactory, apollo,
|
||||
dataSourceName + "-sentinel-apollo-datasource");
|
||||
});
|
||||
Optional.ofNullable(dataSourceProperties.getZk()).ifPresent(zk -> {
|
||||
registerBean(beanFactory, zk,
|
||||
dataSourceName + "-sentinel-zk-datasource");
|
||||
});
|
||||
});
|
||||
|
||||
dataSourceBeanNameList.forEach(beanName -> {
|
||||
ReadableDataSource dataSource = beanFactory.getBean(beanName,
|
||||
ReadableDataSource.class);
|
||||
Object ruleConfig;
|
||||
try {
|
||||
logger.info("[Sentinel Starter] DataSource " + beanName
|
||||
+ " start to loadConfig");
|
||||
ruleConfig = dataSource.loadConfig();
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("[Sentinel Starter] DataSource " + beanName
|
||||
+ " loadConfig error: " + e.getMessage(), e);
|
||||
return;
|
||||
}
|
||||
SentinelProperty sentinelProperty = dataSource.getProperty();
|
||||
Class ruleType = getAndCheckRuleType(ruleConfig, beanName);
|
||||
if (ruleType != null) {
|
||||
if (ruleType == FlowRule.class) {
|
||||
FlowRuleManager.register2Property(sentinelProperty);
|
||||
}
|
||||
else if (ruleType == DegradeRule.class) {
|
||||
DegradeRuleManager.register2Property(sentinelProperty);
|
||||
}
|
||||
else if (ruleType == SystemRule.class) {
|
||||
SystemRuleManager.register2Property(sentinelProperty);
|
||||
}
|
||||
else {
|
||||
AuthorityRuleManager.register2Property(sentinelProperty);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void registerBean(DefaultListableBeanFactory beanFactory,
|
||||
final AbstractDataSourceProperties dataSourceProperties,
|
||||
String dataSourceName) {
|
||||
|
||||
Map<String, Object> propertyMap = Arrays
|
||||
.stream(dataSourceProperties.getClass().getDeclaredFields())
|
||||
.collect(HashMap::new, (m, v) -> {
|
||||
try {
|
||||
v.setAccessible(true);
|
||||
m.put(v.getName(), v.get(dataSourceProperties));
|
||||
}
|
||||
catch (IllegalAccessException e) {
|
||||
logger.error("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " field: " + v.getName() + " invoke error");
|
||||
throw new RuntimeException(
|
||||
"[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " field: " + v.getName() + " invoke error",
|
||||
e);
|
||||
}
|
||||
}, HashMap::putAll);
|
||||
propertyMap.put(CONVERTERCLASS_FIELD, dataSourceProperties.getConverterClass());
|
||||
propertyMap.put(DATATYPE_FIELD, dataSourceProperties.getDataType());
|
||||
|
||||
BeanDefinitionBuilder builder = BeanDefinitionBuilder
|
||||
.genericBeanDefinition(dataSourceProperties.getFactoryBeanName());
|
||||
|
||||
propertyMap.forEach((propertyName, propertyValue) -> {
|
||||
Field field = ReflectionUtils.findField(dataSourceProperties.getClass(),
|
||||
propertyName);
|
||||
if (field != null) {
|
||||
if (DATATYPE_FIELD.equals(propertyName)) {
|
||||
String dataType = StringUtils
|
||||
.trimAllWhitespace(propertyValue.toString());
|
||||
if (CUSTOM_DATATYPE.equals(dataType)) {
|
||||
try {
|
||||
if (StringUtils
|
||||
.isEmpty(dataSourceProperties.getConverterClass())) {
|
||||
throw new RuntimeException(
|
||||
"[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ "dataType is custom, please set converter-class "
|
||||
+ "property");
|
||||
}
|
||||
// construct custom Converter with 'converterClass'
|
||||
// configuration and register
|
||||
String customConvertBeanName = "sentinel-"
|
||||
+ dataSourceProperties.getConverterClass();
|
||||
if (!beanFactory.containsBean(customConvertBeanName)) {
|
||||
beanFactory.registerBeanDefinition(customConvertBeanName,
|
||||
BeanDefinitionBuilder
|
||||
.genericBeanDefinition(
|
||||
Class.forName(dataSourceProperties
|
||||
.getConverterClass()))
|
||||
.getBeanDefinition());
|
||||
}
|
||||
builder.addPropertyReference("converter",
|
||||
customConvertBeanName);
|
||||
}
|
||||
catch (ClassNotFoundException e) {
|
||||
logger.error("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " handle "
|
||||
+ dataSourceProperties.getClass().getSimpleName()
|
||||
+ " error, class name: "
|
||||
+ dataSourceProperties.getConverterClass());
|
||||
throw new RuntimeException(
|
||||
"[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " handle "
|
||||
+ dataSourceProperties.getClass()
|
||||
.getSimpleName()
|
||||
+ " error, class name: "
|
||||
+ dataSourceProperties.getConverterClass(),
|
||||
e);
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (!dataTypeList.contains(StringUtils
|
||||
.trimAllWhitespace(propertyValue.toString()))) {
|
||||
throw new RuntimeException("[Sentinel Starter] DataSource "
|
||||
+ dataSourceName + " dataType: " + propertyValue
|
||||
+ " is not support now. please using these types: "
|
||||
+ dataTypeList.toString());
|
||||
}
|
||||
// converter type now support xml or json.
|
||||
// The bean name of these converters wrapped by
|
||||
// 'sentinel-{converterType}-converter'
|
||||
builder.addPropertyReference("converter",
|
||||
"sentinel-" + propertyValue.toString() + "-converter");
|
||||
}
|
||||
}
|
||||
else if (CONVERTERCLASS_FIELD.equals(propertyName)) {
|
||||
return;
|
||||
}
|
||||
else {
|
||||
// wired properties
|
||||
builder.addPropertyValue(propertyName, propertyValue);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
beanFactory.registerBeanDefinition(dataSourceName, builder.getBeanDefinition());
|
||||
// init in Spring
|
||||
beanFactory.getBean(dataSourceName);
|
||||
dataSourceBeanNameList.add(dataSourceName);
|
||||
}
|
||||
|
||||
private Class getAndCheckRuleType(Object ruleConfig, String dataSourceName) {
|
||||
if (rulesList.contains(ruleConfig.getClass())) {
|
||||
logger.info("[Sentinel Starter] DataSource {} load {} {}", dataSourceName, 1,
|
||||
ruleConfig.getClass().getSimpleName());
|
||||
return ruleConfig.getClass();
|
||||
}
|
||||
else if (ruleConfig instanceof List) {
|
||||
List convertedRuleList = (List) ruleConfig;
|
||||
if (CollectionUtils.isEmpty(convertedRuleList)) {
|
||||
logger.warn("[Sentinel Starter] DataSource {} rule list is empty.",
|
||||
dataSourceName);
|
||||
return null;
|
||||
}
|
||||
if (convertedRuleList.stream()
|
||||
.allMatch(rule -> rulesList.contains(rule.getClass()))) {
|
||||
if (rulesList.contains(convertedRuleList.get(0).getClass())
|
||||
&& convertedRuleList.stream()
|
||||
.filter(rule -> rule.getClass() == convertedRuleList
|
||||
.get(0).getClass())
|
||||
.toArray().length == convertedRuleList.size()) {
|
||||
logger.info("[Sentinel Starter] DataSource {} load {} {}",
|
||||
dataSourceName, convertedRuleList.size(),
|
||||
convertedRuleList.get(0).getClass().getSimpleName());
|
||||
return convertedRuleList.get(0).getClass();
|
||||
}
|
||||
else {
|
||||
logger.warn(
|
||||
"[Sentinel Starter] DataSource {} all rules are not same rule type and it will not be used. "
|
||||
+ "Rule List: {}",
|
||||
dataSourceName, convertedRuleList.toString());
|
||||
}
|
||||
}
|
||||
else {
|
||||
List<Class> classList = (List<Class>) convertedRuleList.stream()
|
||||
.map(Object::getClass).collect(Collectors.toList());
|
||||
logger.error("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " rule class is invalid. Class List: " + classList);
|
||||
throw new RuntimeException(
|
||||
"[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " rule class is invalid. Class List: " + classList);
|
||||
}
|
||||
}
|
||||
else {
|
||||
logger.error("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " rule class is invalid. Class: " + ruleConfig.getClass());
|
||||
throw new RuntimeException("[Sentinel Starter] DataSource " + dataSourceName
|
||||
+ " rule class is invalid. Class: " + ruleConfig.getClass());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
public List<String> getDataSourceBeanNameList() {
|
||||
return dataSourceBeanNameList;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,20 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-alibaba</artifactId>
|
||||
<version>0.2.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<artifactId>spring-cloud-starter-stream-rocketmq</artifactId>
|
||||
<name>Spring Cloud Starter Stream RocketMQ</name>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
@ -0,0 +1,77 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-alibaba</artifactId>
|
||||
<version>0.2.1.BUILD-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream-binder-rocketmq</artifactId>
|
||||
<name>Spring Cloud Alibaba RocketMQ Binder</name>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-stream</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.dropwizard.metrics</groupId>
|
||||
<artifactId>metrics-core</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.rocketmq</groupId>
|
||||
<artifactId>rocketmq-client</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-autoconfigure</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-actuator</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-actuator-autoconfigure</artifactId>
|
||||
<scope>provided</scope>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
||||
@ -0,0 +1,26 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public interface RocketMQBinderConstants {
|
||||
|
||||
/**
|
||||
* Header key
|
||||
*/
|
||||
String ORIGINAL_ROCKET_MESSAGE = "ORIGINAL_ROCKETMQ_MESSAGE";
|
||||
|
||||
String ROCKET_FLAG = "ROCKETMQ_FLAG";
|
||||
|
||||
String ROCKET_SEND_RESULT = "ROCKETMQ_SEND_RESULT";
|
||||
|
||||
String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
|
||||
|
||||
/**
|
||||
* Instrumentation key
|
||||
*/
|
||||
String LASTSEND_TIMESTAMP = "lastSend.timestamp";
|
||||
|
||||
String ENDPOINT_ID = "rocketmq-binder";
|
||||
|
||||
}
|
||||
@ -0,0 +1,101 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.consuming.ConsumersManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQInboundChannelAdapter;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQMessageHandler;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner;
|
||||
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProducerDestination;
|
||||
import org.springframework.integration.core.MessageProducer;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageChannelBinder extends
|
||||
AbstractMessageChannelBinder<ExtendedConsumerProperties<RocketMQConsumerProperties>,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>, RocketMQTopicProvisioner>
|
||||
implements ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(RocketMQMessageChannelBinder.class);
|
||||
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
private final RocketMQTopicProvisioner rocketTopicProvisioner;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
private final ConsumersManager consumersManager;
|
||||
|
||||
public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
|
||||
RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQTopicProvisioner provisioningProvider,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
super(null, provisioningProvider);
|
||||
this.consumersManager = consumersManager;
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketTopicProvisioner = provisioningProvider;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>
|
||||
producerProperties,
|
||||
MessageChannel errorChannel) throws Exception {
|
||||
if (producerProperties.getExtension().getEnabled()) {
|
||||
return new RocketMQMessageHandler(destination.getName(), producerProperties.getExtension(),
|
||||
rocketBinderConfigurationProperties, instrumentationManager);
|
||||
} else {
|
||||
throw new RuntimeException(
|
||||
"Binding for channel " + destination.getName() + "has been disabled, message can't be delivered");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties>
|
||||
consumerProperties)
|
||||
throws Exception {
|
||||
if (group == null || "".equals(group)) {
|
||||
throw new RuntimeException("'group' must be configured for channel + " + destination.getName());
|
||||
}
|
||||
|
||||
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(consumersManager,
|
||||
consumerProperties, destination.getName(), group, instrumentationManager);
|
||||
|
||||
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, group,
|
||||
consumerProperties);
|
||||
if (consumerProperties.getMaxAttempts() > 1) {
|
||||
rocketInboundChannelAdapter.setRetryTemplate(buildRetryTemplate(consumerProperties));
|
||||
rocketInboundChannelAdapter.setRecoveryCallback(errorInfrastructure.getRecoverer());
|
||||
} else {
|
||||
rocketInboundChannelAdapter.setErrorChannel(errorInfrastructure.getErrorChannel());
|
||||
}
|
||||
|
||||
return rocketInboundChannelAdapter;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
return extendedBindingProperties.getExtendedProducerProperties(channelName);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,106 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.rocketmq.client.producer.SendResult;
|
||||
import org.apache.rocketmq.common.message.MessageConst;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.consuming.Acknowledgement;
|
||||
import org.springframework.integration.support.MutableMessage;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
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 <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageHeaderAccessor extends MessageHeaderAccessor {
|
||||
|
||||
public RocketMQMessageHeaderAccessor() {
|
||||
super();
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor(Message<?> message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
public Acknowledgement getAcknowledgement(Message message) {
|
||||
return message.getHeaders().get(ACKNOWLEDGEMENT_KEY, Acknowledgement.class);
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withAcknowledgment(Acknowledgement acknowledgment) {
|
||||
setHeader(ACKNOWLEDGEMENT_KEY, acknowledgment);
|
||||
return this;
|
||||
}
|
||||
|
||||
public String getTags() {
|
||||
return (String)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_TAGS, "");
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withTags(String tag) {
|
||||
setHeader(MessageConst.PROPERTY_TAGS, tag);
|
||||
return this;
|
||||
}
|
||||
|
||||
public String getKeys() {
|
||||
return (String)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_KEYS, "");
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withKeys(String keys) {
|
||||
setHeader(MessageConst.PROPERTY_KEYS, keys);
|
||||
return this;
|
||||
}
|
||||
|
||||
public MessageExt getRocketMessage() {
|
||||
return getMessageHeaders().get(ORIGINAL_ROCKET_MESSAGE, MessageExt.class);
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withRocketMessage(MessageExt message) {
|
||||
setHeader(ORIGINAL_ROCKET_MESSAGE, message);
|
||||
return this;
|
||||
}
|
||||
|
||||
public Integer getDelayTimeLevel() {
|
||||
return (Integer)getMessageHeaders().getOrDefault(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 0);
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
|
||||
setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
|
||||
return this;
|
||||
}
|
||||
|
||||
public Integer getFlag() {
|
||||
return (Integer)getMessageHeaders().getOrDefault(ROCKET_FLAG, 0);
|
||||
}
|
||||
|
||||
public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
|
||||
setHeader(ROCKET_FLAG, delayTimeLevel);
|
||||
return this;
|
||||
}
|
||||
|
||||
public SendResult getSendResult() {
|
||||
return getMessageHeaders().get(ROCKET_SEND_RESULT, SendResult.class);
|
||||
}
|
||||
|
||||
public static void putSendResult(MutableMessage message, SendResult sendResult) {
|
||||
message.getHeaders().put(ROCKET_SEND_RESULT, sendResult);
|
||||
}
|
||||
|
||||
public Map<String, String> getUserProperties() {
|
||||
Map<String, String> result = new HashMap<>();
|
||||
for (Map.Entry<String, Object> entry : this.toMap().entrySet()) {
|
||||
if (entry.getValue() instanceof String && !MessageConst.STRING_HASH_SET.contains(entry.getKey()) && !entry
|
||||
.getKey().equals(MessageHeaders.CONTENT_TYPE)) {
|
||||
result.put(entry.getKey(), (String)entry.getValue());
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,39 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.actuator;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
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.ReadOperation;
|
||||
|
||||
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ENDPOINT_ID;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Endpoint(id = ENDPOINT_ID)
|
||||
public class RocketMQBinderEndpoint {
|
||||
|
||||
private MetricRegistry metricRegistry = new MetricRegistry();
|
||||
private Map<String, Object> runtime = new ConcurrentHashMap<>();
|
||||
|
||||
@ReadOperation
|
||||
public Map<String, Object> invoke() {
|
||||
Map<String, Object> result = new HashMap<>();
|
||||
result.put("metrics", metricRegistry().getMetrics());
|
||||
result.put("runtime", runtime());
|
||||
return result;
|
||||
}
|
||||
|
||||
public MetricRegistry metricRegistry() {
|
||||
return metricRegistry;
|
||||
}
|
||||
|
||||
public Map<String, Object> runtime() {
|
||||
return runtime;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.actuator;
|
||||
|
||||
import org.springframework.boot.actuate.health.AbstractHealthIndicator;
|
||||
import org.springframework.boot.actuate.health.Health;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
public RocketMQBinderHealthIndicator(InstrumentationManager instrumentationManager) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doHealthCheck(Health.Builder builder) throws Exception {
|
||||
if (instrumentationManager.getHealthInstrumentations().stream().
|
||||
allMatch(Instrumentation::isUp)) {
|
||||
builder.up();
|
||||
return;
|
||||
}
|
||||
if (instrumentationManager.getHealthInstrumentations().stream().
|
||||
allMatch(Instrumentation::isOutOfService)) {
|
||||
builder.outOfService();
|
||||
return;
|
||||
}
|
||||
builder.down();
|
||||
instrumentationManager.getHealthInstrumentations().stream().
|
||||
filter(instrumentation -> !instrumentation.isStarted()).
|
||||
forEach(instrumentation1 -> builder.withException(instrumentation1.getStartException()));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,54 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.config;
|
||||
|
||||
import org.apache.rocketmq.client.log.ClientLogger;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.consuming.ConsumersManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.provisioning.RocketMQTopicProvisioner;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({RocketMQBinderConfigurationProperties.class, RocketMQExtendedBindingProperties.class})
|
||||
public class RocketMQBinderAutoConfiguration {
|
||||
|
||||
private final RocketMQExtendedBindingProperties extendedBindingProperties;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
@Autowired
|
||||
public RocketMQBinderAutoConfiguration(RocketMQExtendedBindingProperties extendedBindingProperties,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.extendedBindingProperties = extendedBindingProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
System.setProperty(ClientLogger.CLIENT_LOG_LEVEL, this.rocketBinderConfigurationProperties.getLogLevel());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RocketMQTopicProvisioner provisioningProvider() {
|
||||
return new RocketMQTopicProvisioner();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RocketMQMessageChannelBinder rocketMessageChannelBinder(RocketMQTopicProvisioner provisioningProvider,
|
||||
InstrumentationManager instrumentationManager,
|
||||
ConsumersManager consumersManager) {
|
||||
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(consumersManager, extendedBindingProperties,
|
||||
provisioningProvider, rocketBinderConfigurationProperties, instrumentationManager);
|
||||
return binder;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ConsumersManager consumersManager(InstrumentationManager instrumentationManager) {
|
||||
return new ConsumersManager(instrumentationManager, rocketBinderConfigurationProperties);
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,33 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.config;
|
||||
|
||||
import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderEndpoint;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.actuator.RocketMQBinderHealthIndicator;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@Configuration
|
||||
@AutoConfigureAfter(EndpointAutoConfiguration.class)
|
||||
public class RocketMQBinderEndpointAutoConfiguration {
|
||||
|
||||
@Bean
|
||||
public RocketMQBinderEndpoint rocketBinderEndpoint() {
|
||||
return new RocketMQBinderEndpoint();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public RocketMQBinderHealthIndicator rocketBinderHealthIndicator(InstrumentationManager instrumentationManager) {
|
||||
return new RocketMQBinderHealthIndicator(instrumentationManager);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public InstrumentationManager instrumentationManager(RocketMQBinderEndpoint rocketBinderEndpoint) {
|
||||
return new InstrumentationManager(rocketBinderEndpoint.metricRegistry(), rocketBinderEndpoint.runtime());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,78 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.consuming;
|
||||
|
||||
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;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class Acknowledgement {
|
||||
|
||||
/**
|
||||
* for {@link ConsumeConcurrentlyContext} using
|
||||
*/
|
||||
private ConsumeConcurrentlyStatus consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
||||
/**
|
||||
* 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 Integer consumeConcurrentlyDelayLevel = 0;
|
||||
|
||||
/**
|
||||
* for {@link ConsumeOrderlyContext} using
|
||||
*/
|
||||
private ConsumeOrderlyStatus consumeOrderlyStatus = ConsumeOrderlyStatus.SUCCESS;
|
||||
private Long consumeOrderlySuspendCurrentQueueTimeMill = -1L;
|
||||
|
||||
public Acknowledgement setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus consumeConcurrentlyStatus) {
|
||||
this.consumeConcurrentlyStatus = consumeConcurrentlyStatus;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ConsumeConcurrentlyStatus getConsumeConcurrentlyStatus() {
|
||||
return consumeConcurrentlyStatus;
|
||||
}
|
||||
|
||||
public ConsumeOrderlyStatus getConsumeOrderlyStatus() {
|
||||
return consumeOrderlyStatus;
|
||||
}
|
||||
|
||||
public Acknowledgement setConsumeOrderlyStatus(ConsumeOrderlyStatus consumeOrderlyStatus) {
|
||||
this.consumeOrderlyStatus = consumeOrderlyStatus;
|
||||
return this;
|
||||
}
|
||||
|
||||
public Integer getConsumeConcurrentlyDelayLevel() {
|
||||
return consumeConcurrentlyDelayLevel;
|
||||
}
|
||||
|
||||
public void setConsumeConcurrentlyDelayLevel(Integer consumeConcurrentlyDelayLevel) {
|
||||
this.consumeConcurrentlyDelayLevel = consumeConcurrentlyDelayLevel;
|
||||
}
|
||||
|
||||
public Long getConsumeOrderlySuspendCurrentQueueTimeMill() {
|
||||
return consumeOrderlySuspendCurrentQueueTimeMill;
|
||||
}
|
||||
|
||||
public void setConsumeOrderlySuspendCurrentQueueTimeMill(Long consumeOrderlySuspendCurrentQueueTimeMill) {
|
||||
this.consumeOrderlySuspendCurrentQueueTimeMill = consumeOrderlySuspendCurrentQueueTimeMill;
|
||||
}
|
||||
|
||||
public static Acknowledgement buildOrderlyInstance() {
|
||||
Acknowledgement acknowledgement = new Acknowledgement();
|
||||
acknowledgement.setConsumeOrderlyStatus(ConsumeOrderlyStatus.SUCCESS);
|
||||
return acknowledgement;
|
||||
}
|
||||
|
||||
public static Acknowledgement buildConcurrentlyInstance() {
|
||||
Acknowledgement acknowledgement = new Acknowledgement();
|
||||
acknowledgement.setConsumeConcurrentlyStatus(ConsumeConcurrentlyStatus.CONSUME_SUCCESS);
|
||||
return acknowledgement;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,106 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.consuming;
|
||||
|
||||
import java.util.AbstractMap;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.ConsumerGroupInstrumentation;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ConsumersManager {
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(this.getClass());
|
||||
|
||||
private final Map<String, DefaultMQPushConsumer> consumerGroups = new HashMap<>();
|
||||
private final Map<String, Boolean> started = new HashMap<>();
|
||||
private final Map<Map.Entry<String, String>, ExtendedConsumerProperties<RocketMQConsumerProperties>> propertiesMap
|
||||
= new HashMap<>();
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
public ConsumersManager(InstrumentationManager instrumentationManager,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
}
|
||||
|
||||
public synchronized DefaultMQPushConsumer getOrCreateConsumer(String group, String topic,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
|
||||
propertiesMap.put(new AbstractMap.SimpleEntry<>(group, topic), consumerProperties);
|
||||
ConsumerGroupInstrumentation instrumentation = instrumentationManager.getConsumerGroupInstrumentation(group);
|
||||
instrumentationManager.addHealthInstrumentation(instrumentation);
|
||||
|
||||
if (consumerGroups.containsKey(group)) {
|
||||
return consumerGroups.get(group);
|
||||
}
|
||||
|
||||
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
|
||||
consumer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
|
||||
consumerGroups.put(group, consumer);
|
||||
started.put(group, false);
|
||||
consumer.setConsumeThreadMax(consumerProperties.getConcurrency());
|
||||
consumer.setConsumeThreadMin(consumerProperties.getConcurrency());
|
||||
if (consumerProperties.getExtension().getBroadcasting()) {
|
||||
consumer.setMessageModel(MessageModel.BROADCASTING);
|
||||
}
|
||||
logger.info("RocketMQ consuming for SCS group {} created", group);
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public synchronized void startConsumers() throws MQClientException {
|
||||
for (String group : getConsumerGroups()) {
|
||||
start(group);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void startConsumer(String group) throws MQClientException {
|
||||
start(group);
|
||||
}
|
||||
|
||||
public synchronized void stopConsumer(String group) {
|
||||
stop(group);
|
||||
}
|
||||
|
||||
private void stop(String group) {
|
||||
if (consumerGroups.get(group) != null) {
|
||||
consumerGroups.get(group).shutdown();
|
||||
started.put(group, false);
|
||||
}
|
||||
}
|
||||
|
||||
private synchronized void start(String group) throws MQClientException {
|
||||
if (started.get(group)) {
|
||||
return;
|
||||
}
|
||||
ConsumerGroupInstrumentation groupInstrumentation = instrumentationManager.getConsumerGroupInstrumentation(
|
||||
group);
|
||||
instrumentationManager.addHealthInstrumentation(groupInstrumentation);
|
||||
try {
|
||||
consumerGroups.get(group).start();
|
||||
started.put(group, true);
|
||||
groupInstrumentation.markStartedSuccessfully();
|
||||
} catch (MQClientException e) {
|
||||
groupInstrumentation.markStartFailed(e);
|
||||
logger.error("RocketMQ Consumer hasn't been started. Caused by " + e.getErrorMessage(), e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized Set<String> getConsumerGroups() {
|
||||
return consumerGroups.keySet();
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,253 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.integration;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.apache.commons.lang3.ClassUtils;
|
||||
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.MessageListener;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.common.message.MessageExt;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageHeaderAccessor;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.consuming.Acknowledgement;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.consuming.ConsumersManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.ConsumerInstrumentation;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import org.springframework.integration.endpoint.MessageProducerSupport;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
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.StringUtils;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(RocketMQInboundChannelAdapter.class);
|
||||
|
||||
private ConsumerInstrumentation consumerInstrumentation;
|
||||
|
||||
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
|
||||
|
||||
private final String destination;
|
||||
|
||||
private final String group;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
private final ConsumersManager consumersManager;
|
||||
|
||||
private RetryTemplate retryTemplate;
|
||||
|
||||
private RecoveryCallback<? extends Object> recoveryCallback;
|
||||
|
||||
public RocketMQInboundChannelAdapter(ConsumersManager consumersManager,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
|
||||
String destination, String group,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
this.consumersManager = consumersManager;
|
||||
this.consumerProperties = consumerProperties;
|
||||
this.destination = destination;
|
||||
this.group = group;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
if (!consumerProperties.getExtension().getEnabled()) {
|
||||
return;
|
||||
}
|
||||
|
||||
String tags = consumerProperties == null ? null : consumerProperties.getExtension().getTags();
|
||||
Boolean isOrderly = consumerProperties == null ? false : consumerProperties.getExtension().getOrderly();
|
||||
|
||||
DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group, destination, consumerProperties);
|
||||
|
||||
final CloudStreamMessageListener listener = isOrderly ? new CloudStreamMessageListenerOrderly(
|
||||
instrumentationManager)
|
||||
: new CloudStreamMessageListenerConcurrently(instrumentationManager);
|
||||
|
||||
if (retryTemplate != null) {
|
||||
retryTemplate.registerListener(listener);
|
||||
}
|
||||
|
||||
Set<String> tagsSet = tags == null ? new HashSet<>() : Arrays.stream(tags.split("\\|\\|")).map(String::trim)
|
||||
.collect(Collectors.toSet());
|
||||
|
||||
consumerInstrumentation = instrumentationManager.getConsumerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(consumerInstrumentation);
|
||||
|
||||
try {
|
||||
if (!StringUtils.isEmpty(consumerProperties.getExtension().getSql())) {
|
||||
consumer.subscribe(destination, MessageSelector.bySql(consumerProperties.getExtension().getSql()));
|
||||
} else {
|
||||
consumer.subscribe(destination, String.join(" || ", tagsSet));
|
||||
}
|
||||
consumerInstrumentation.markStartedSuccessfully();
|
||||
} 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);
|
||||
|
||||
try {
|
||||
consumersManager.startConsumer(group);
|
||||
} catch (MQClientException e) {
|
||||
logger.error("RocketMQ Consumer startup failed. Caused by " + e.getErrorMessage(), e);
|
||||
throw new RuntimeException("RocketMQ Consumer startup failed.", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
consumersManager.stopConsumer(group);
|
||||
}
|
||||
|
||||
public void setRetryTemplate(RetryTemplate retryTemplate) {
|
||||
this.retryTemplate = retryTemplate;
|
||||
}
|
||||
|
||||
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
|
||||
this.recoveryCallback = recoveryCallback;
|
||||
}
|
||||
|
||||
protected class CloudStreamMessageListener implements MessageListener, RetryListener {
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
CloudStreamMessageListener(InstrumentationManager instrumentationManager) {
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
Acknowledgement consumeMessage(final List<MessageExt> msgs) {
|
||||
boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
|
||||
try {
|
||||
if (enableRetry) {
|
||||
return RocketMQInboundChannelAdapter.this.retryTemplate.execute(
|
||||
(RetryCallback<Acknowledgement, Exception>)context -> doSendMsgs(msgs, context),
|
||||
new RecoveryCallback<Acknowledgement>() {
|
||||
@Override
|
||||
public Acknowledgement recover(RetryContext context) throws Exception {
|
||||
RocketMQInboundChannelAdapter.this.recoveryCallback.recover(context);
|
||||
if (ClassUtils.isAssignable(this.getClass(), MessageListenerConcurrently.class)) {
|
||||
return Acknowledgement.buildConcurrentlyInstance();
|
||||
} else {
|
||||
return Acknowledgement.buildOrderlyInstance();
|
||||
}
|
||||
}
|
||||
});
|
||||
} else {
|
||||
Acknowledgement result = doSendMsgs(msgs, null);
|
||||
instrumentationManager.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) {
|
||||
List<Acknowledgement> acknowledgements = new ArrayList<>();
|
||||
msgs.forEach(msg -> {
|
||||
String retryInfo = context == null ? "" : "retryCount-" + String.valueOf(context.getRetryCount()) + "|";
|
||||
logger.debug(retryInfo + "consuming msg:\n" + msg);
|
||||
logger.debug(retryInfo + "message body:\n" + new String(msg.getBody()));
|
||||
Acknowledgement acknowledgement = new Acknowledgement();
|
||||
Message<byte[]> toChannel = MessageBuilder.withPayload(msg.getBody()).
|
||||
setHeaders(new RocketMQMessageHeaderAccessor().
|
||||
withAcknowledgment(acknowledgement).
|
||||
withTags(msg.getTags()).
|
||||
withKeys(msg.getKeys()).
|
||||
withFlag(msg.getFlag()).
|
||||
withRocketMessage(msg)
|
||||
).build();
|
||||
acknowledgements.add(acknowledgement);
|
||||
RocketMQInboundChannelAdapter.this.sendMessage(toChannel);
|
||||
});
|
||||
return acknowledgements.get(0);
|
||||
}
|
||||
|
||||
@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) {
|
||||
if (throwable != null) {
|
||||
instrumentationManager.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumedFailure();
|
||||
} else {
|
||||
instrumentationManager.getConsumerInstrumentation(
|
||||
RocketMQInboundChannelAdapter.this.destination)
|
||||
.markConsumed();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
|
||||
Throwable throwable) {
|
||||
}
|
||||
}
|
||||
|
||||
protected class CloudStreamMessageListenerConcurrently extends CloudStreamMessageListener implements
|
||||
MessageListenerConcurrently {
|
||||
|
||||
public CloudStreamMessageListenerConcurrently(InstrumentationManager instrumentationManager) {
|
||||
super(instrumentationManager);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumeConcurrentlyStatus consumeMessage(final List<MessageExt> msgs,
|
||||
ConsumeConcurrentlyContext context) {
|
||||
Acknowledgement acknowledgement = consumeMessage(msgs);
|
||||
context.setDelayLevelWhenNextConsume(acknowledgement.getConsumeConcurrentlyDelayLevel());
|
||||
return acknowledgement.getConsumeConcurrentlyStatus();
|
||||
}
|
||||
}
|
||||
|
||||
protected class CloudStreamMessageListenerOrderly extends CloudStreamMessageListener implements
|
||||
MessageListenerOrderly {
|
||||
|
||||
public CloudStreamMessageListenerOrderly(InstrumentationManager instrumentationManager) {
|
||||
super(instrumentationManager);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
|
||||
Acknowledgement acknowledgement = consumeMessage(msgs);
|
||||
context.setSuspendCurrentQueueTimeMillis((acknowledgement.getConsumeOrderlySuspendCurrentQueueTimeMill()));
|
||||
return acknowledgement.getConsumeOrderlyStatus();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,131 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.integration;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.rocketmq.client.exception.MQBrokerException;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.apache.rocketmq.client.producer.DefaultMQProducer;
|
||||
import org.apache.rocketmq.client.producer.SendResult;
|
||||
import org.apache.rocketmq.client.producer.SendStatus;
|
||||
import org.apache.rocketmq.common.message.Message;
|
||||
import org.apache.rocketmq.remoting.exception.RemotingException;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageHeaderAccessor;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.metrics.ProducerInstrumentation;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
|
||||
import org.springframework.context.Lifecycle;
|
||||
import org.springframework.integration.handler.AbstractMessageHandler;
|
||||
import org.springframework.integration.support.MutableMessage;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
|
||||
|
||||
private DefaultMQProducer producer;
|
||||
|
||||
private ProducerInstrumentation producerInstrumentation;
|
||||
|
||||
private final RocketMQProducerProperties producerProperties;
|
||||
|
||||
private final String destination;
|
||||
|
||||
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
|
||||
|
||||
private final InstrumentationManager instrumentationManager;
|
||||
|
||||
protected volatile boolean running = false;
|
||||
|
||||
public RocketMQMessageHandler(String destination, RocketMQProducerProperties producerProperties,
|
||||
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
|
||||
InstrumentationManager instrumentationManager) {
|
||||
this.destination = destination;
|
||||
this.producerProperties = producerProperties;
|
||||
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
|
||||
this.instrumentationManager = instrumentationManager;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start() {
|
||||
producer = new DefaultMQProducer(destination);
|
||||
|
||||
producerInstrumentation = instrumentationManager.getProducerInstrumentation(destination);
|
||||
instrumentationManager.addHealthInstrumentation(producerInstrumentation);
|
||||
|
||||
producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
|
||||
|
||||
if (producerProperties.getMaxMessageSize() > 0) {
|
||||
producer.setMaxMessageSize(producerProperties.getMaxMessageSize());
|
||||
}
|
||||
|
||||
try {
|
||||
producer.start();
|
||||
producerInstrumentation.markStartedSuccessfully();
|
||||
} catch (MQClientException e) {
|
||||
producerInstrumentation.markStartFailed(e);
|
||||
logger.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
|
||||
throw new MessagingException(e.getMessage(), e);
|
||||
}
|
||||
running = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (producer != null) {
|
||||
producer.shutdown();
|
||||
}
|
||||
running = false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isRunning() {
|
||||
return running;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void handleMessageInternal(org.springframework.messaging.Message<?> message) throws Exception {
|
||||
try {
|
||||
Message toSend;
|
||||
if (message.getPayload() instanceof byte[]) {
|
||||
toSend = new Message(destination, (byte[])message.getPayload());
|
||||
} else if (message.getPayload() instanceof String) {
|
||||
toSend = new Message(destination, ((String)message.getPayload()).getBytes());
|
||||
} else {
|
||||
throw new UnsupportedOperationException(
|
||||
"Payload class isn't supported: " + message.getPayload().getClass());
|
||||
}
|
||||
RocketMQMessageHeaderAccessor headerAccessor = new RocketMQMessageHeaderAccessor(message);
|
||||
headerAccessor.setLeaveMutable(true);
|
||||
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);
|
||||
|
||||
if (!sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
|
||||
throw new MQClientException("message hasn't been sent", null);
|
||||
}
|
||||
if (message instanceof MutableMessage) {
|
||||
RocketMQMessageHeaderAccessor.putSendResult((MutableMessage)message, sendRes);
|
||||
}
|
||||
instrumentationManager.getRuntime().put(RocketMQBinderConstants.LASTSEND_TIMESTAMP,
|
||||
Instant.now().toEpochMilli());
|
||||
producerInstrumentation.markSent();
|
||||
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException |
|
||||
UnsupportedOperationException e) {
|
||||
producerInstrumentation.markSentFailure();
|
||||
logger.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
|
||||
throw new MessagingException(e.getMessage(), e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,34 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ConsumerGroupInstrumentation extends Instrumentation {
|
||||
private MetricRegistry metricRegistry;
|
||||
|
||||
private AtomicBoolean delayedStart = new AtomicBoolean(false);
|
||||
|
||||
public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) {
|
||||
super(name);
|
||||
this.metricRegistry = metricRegistry;
|
||||
}
|
||||
|
||||
public void markDelayedStart() {
|
||||
delayedStart.set(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isUp() {
|
||||
return started.get() || delayedStart.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOutOfService() {
|
||||
return !started.get() && startException == null && !delayedStart.get();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import com.codahale.metrics.Counter;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
/**
|
||||
* @author juven.xuxb
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ConsumerInstrumentation extends Instrumentation {
|
||||
|
||||
private final Counter totalConsumed;
|
||||
private final Counter totalConsumedFailures;
|
||||
private final Meter consumedPerSecond;
|
||||
private final Meter consumedFailuresPerSecond;
|
||||
|
||||
public ConsumerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
this.totalConsumed = registry.counter(name(baseMetricName, "totalConsumed"));
|
||||
this.consumedPerSecond = registry.meter(name(baseMetricName, "consumedPerSecond"));
|
||||
this.totalConsumedFailures = registry.counter(name(baseMetricName, "totalConsumedFailures"));
|
||||
this.consumedFailuresPerSecond = registry.meter(name(baseMetricName, "consumedFailuresPerSecond"));
|
||||
}
|
||||
|
||||
public void markConsumed() {
|
||||
totalConsumed.inc();
|
||||
consumedPerSecond.mark();
|
||||
}
|
||||
|
||||
public void markConsumedFailure() {
|
||||
totalConsumedFailures.inc();
|
||||
consumedFailuresPerSecond.mark();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,50 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class Instrumentation {
|
||||
private final String name;
|
||||
protected final AtomicBoolean started = new AtomicBoolean(false);
|
||||
protected Exception startException = null;
|
||||
|
||||
Instrumentation(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public boolean isDown() {
|
||||
return startException != null;
|
||||
}
|
||||
|
||||
public boolean isUp() {
|
||||
return started.get();
|
||||
}
|
||||
|
||||
public boolean isOutOfService() {
|
||||
return !started.get() && startException == null;
|
||||
}
|
||||
|
||||
public void markStartedSuccessfully() {
|
||||
started.set(true);
|
||||
}
|
||||
|
||||
public void markStartFailed(Exception e) {
|
||||
started.set(false);
|
||||
startException = e;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public boolean isStarted() {
|
||||
return started.get();
|
||||
}
|
||||
|
||||
public Exception getStartException() {
|
||||
return startException;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,57 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class InstrumentationManager {
|
||||
private final MetricRegistry metricRegistry;
|
||||
private final Map<String, Object> runtime;
|
||||
private final Map<String, ProducerInstrumentation> producerInstrumentations = new HashMap<>();
|
||||
private final Map<String, ConsumerInstrumentation> consumeInstrumentations = new HashMap<>();
|
||||
private final Map<String, ConsumerGroupInstrumentation> consumerGroupsInstrumentations = new HashMap<>();
|
||||
|
||||
private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
|
||||
|
||||
public InstrumentationManager(MetricRegistry metricRegistry, Map<String, Object> runtime) {
|
||||
this.metricRegistry = metricRegistry;
|
||||
this.runtime = runtime;
|
||||
}
|
||||
|
||||
public ProducerInstrumentation getProducerInstrumentation(String destination) {
|
||||
String key = "scs-rocketmq.producer." + destination;
|
||||
producerInstrumentations.putIfAbsent(key, new ProducerInstrumentation(metricRegistry, key));
|
||||
return producerInstrumentations.get(key);
|
||||
}
|
||||
|
||||
public ConsumerInstrumentation getConsumerInstrumentation(String destination) {
|
||||
String key = "scs-rocketmq.consumer." + destination;
|
||||
consumeInstrumentations.putIfAbsent(key, new ConsumerInstrumentation(metricRegistry, key));
|
||||
return consumeInstrumentations.get(key);
|
||||
}
|
||||
|
||||
public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) {
|
||||
String key = "scs-rocketmq.consumerGroup." + group;
|
||||
consumerGroupsInstrumentations.putIfAbsent(key, new ConsumerGroupInstrumentation(metricRegistry, key));
|
||||
return consumerGroupsInstrumentations.get(key);
|
||||
}
|
||||
|
||||
public Set<Instrumentation> getHealthInstrumentations() {
|
||||
return healthInstrumentations.entrySet().stream().map(Map.Entry::getValue).collect(Collectors.toSet());
|
||||
}
|
||||
|
||||
public void addHealthInstrumentation(Instrumentation instrumentation) {
|
||||
healthInstrumentations.put(instrumentation.getName(), instrumentation);
|
||||
}
|
||||
|
||||
public Map<String, Object> getRuntime() {
|
||||
return runtime;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.metrics;
|
||||
|
||||
import com.codahale.metrics.Counter;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
|
||||
import static com.codahale.metrics.MetricRegistry.name;
|
||||
|
||||
/**
|
||||
* @author juven.xuxb
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class ProducerInstrumentation extends Instrumentation {
|
||||
|
||||
private final Counter totalSent;
|
||||
private final Counter totalSentFailures;
|
||||
private final Meter sentPerSecond;
|
||||
private final Meter sentFailuresPerSecond;
|
||||
|
||||
public ProducerInstrumentation(MetricRegistry registry, String baseMetricName) {
|
||||
super(baseMetricName);
|
||||
this.totalSent = registry.counter(name(baseMetricName, "totalSent"));
|
||||
this.totalSentFailures = registry.counter(name(baseMetricName, "totalSentFailures"));
|
||||
this.sentPerSecond = registry.meter(name(baseMetricName, "sentPerSecond"));
|
||||
this.sentFailuresPerSecond = registry.meter(name(baseMetricName, "sentFailuresPerSecond"));
|
||||
}
|
||||
|
||||
public void markSent() {
|
||||
totalSent.inc();
|
||||
sentPerSecond.mark();
|
||||
}
|
||||
|
||||
public void markSentFailure() {
|
||||
totalSentFailures.inc();
|
||||
sentFailuresPerSecond.mark();
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,32 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
|
||||
public class RocketMQBinderConfigurationProperties {
|
||||
|
||||
private String namesrvAddr;
|
||||
|
||||
private String logLevel = "ERROR";
|
||||
|
||||
public String getNamesrvAddr() {
|
||||
return namesrvAddr;
|
||||
}
|
||||
|
||||
public void setNamesrvAddr(String namesrvAddr) {
|
||||
this.namesrvAddr = namesrvAddr;
|
||||
}
|
||||
|
||||
public String getLogLevel() {
|
||||
return logLevel;
|
||||
}
|
||||
|
||||
public void setLogLevel(String logLevel) {
|
||||
this.logLevel = logLevel;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,28 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQBindingProperties {
|
||||
|
||||
private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties();
|
||||
|
||||
private RocketMQProducerProperties producer = new RocketMQProducerProperties();
|
||||
|
||||
public RocketMQConsumerProperties getConsumer() {
|
||||
return consumer;
|
||||
}
|
||||
|
||||
public void setConsumer(RocketMQConsumerProperties consumer) {
|
||||
this.consumer = consumer;
|
||||
}
|
||||
|
||||
public RocketMQProducerProperties getProducer() {
|
||||
return producer;
|
||||
}
|
||||
|
||||
public void setProducer(RocketMQProducerProperties producer) {
|
||||
this.producer = producer;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,79 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
import org.apache.rocketmq.client.consumer.MQPushConsumer;
|
||||
import org.apache.rocketmq.client.consumer.MessageSelector;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
|
||||
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
|
||||
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQConsumerProperties {
|
||||
|
||||
/**
|
||||
* using '||' to split tag
|
||||
* {@link MQPushConsumer#subscribe(String, String)}
|
||||
*/
|
||||
private String tags;
|
||||
|
||||
/**
|
||||
* {@link MQPushConsumer#subscribe(String, MessageSelector)}
|
||||
* {@link MessageSelector#bySql(String)}
|
||||
*/
|
||||
private String sql;
|
||||
|
||||
/**
|
||||
* {@link MessageModel#BROADCASTING}
|
||||
*/
|
||||
private Boolean broadcasting = false;
|
||||
|
||||
/**
|
||||
* if orderly is true, using {@link MessageListenerOrderly}
|
||||
* else if orderly if false, using {@link MessageListenerConcurrently}
|
||||
*/
|
||||
private Boolean orderly = false;
|
||||
|
||||
private Boolean enabled = true;
|
||||
|
||||
public String getTags() {
|
||||
return tags;
|
||||
}
|
||||
|
||||
public void setTags(String tags) {
|
||||
this.tags = tags;
|
||||
}
|
||||
|
||||
public String getSql() {
|
||||
return sql;
|
||||
}
|
||||
|
||||
public void setSql(String sql) {
|
||||
this.sql = sql;
|
||||
}
|
||||
|
||||
public Boolean getOrderly() {
|
||||
return orderly;
|
||||
}
|
||||
|
||||
public void setOrderly(Boolean orderly) {
|
||||
this.orderly = orderly;
|
||||
}
|
||||
|
||||
public Boolean getEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(Boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
public Boolean getBroadcasting() {
|
||||
return broadcasting;
|
||||
}
|
||||
|
||||
public void setBroadcasting(Boolean broadcasting) {
|
||||
this.broadcasting = broadcasting;
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,64 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
@ConfigurationProperties("spring.cloud.stream.rocketmq")
|
||||
public class RocketMQExtendedBindingProperties implements
|
||||
ExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties> {
|
||||
|
||||
private Map<String, RocketMQBindingProperties> bindings = new HashMap<>();
|
||||
|
||||
public Map<String, RocketMQBindingProperties> getBindings() {
|
||||
return this.bindings;
|
||||
}
|
||||
|
||||
public void setBindings(Map<String, RocketMQBindingProperties> bindings) {
|
||||
this.bindings = bindings;
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
|
||||
if (bindings.containsKey(channelName)) {
|
||||
if (bindings.get(channelName).getConsumer() != null) {
|
||||
return bindings.get(channelName).getConsumer();
|
||||
} else {
|
||||
RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
|
||||
this.bindings.get(channelName).setConsumer(properties);
|
||||
return properties;
|
||||
}
|
||||
} else {
|
||||
RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
|
||||
RocketMQBindingProperties rbp = new RocketMQBindingProperties();
|
||||
rbp.setConsumer(properties);
|
||||
bindings.put(channelName, rbp);
|
||||
return properties;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized RocketMQProducerProperties getExtendedProducerProperties(String channelName) {
|
||||
if (bindings.containsKey(channelName)) {
|
||||
if (bindings.get(channelName).getProducer() != null) {
|
||||
return bindings.get(channelName).getProducer();
|
||||
} else {
|
||||
RocketMQProducerProperties properties = new RocketMQProducerProperties();
|
||||
this.bindings.get(channelName).setProducer(properties);
|
||||
return properties;
|
||||
}
|
||||
} else {
|
||||
RocketMQProducerProperties properties = new RocketMQProducerProperties();
|
||||
RocketMQBindingProperties rbp = new RocketMQBindingProperties();
|
||||
rbp.setProducer(properties);
|
||||
bindings.put(channelName, rbp);
|
||||
return properties;
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.properties;
|
||||
|
||||
import org.apache.rocketmq.client.producer.DefaultMQProducer;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQProducerProperties {
|
||||
|
||||
private Boolean enabled = true;
|
||||
|
||||
/**
|
||||
* Maximum allowed message size in bytes
|
||||
* {@link DefaultMQProducer#maxMessageSize}
|
||||
*/
|
||||
private Integer maxMessageSize = 0;
|
||||
|
||||
public Boolean getEnabled() {
|
||||
return enabled;
|
||||
}
|
||||
|
||||
public void setEnabled(Boolean enabled) {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
public Integer getMaxMessageSize() {
|
||||
return maxMessageSize;
|
||||
}
|
||||
|
||||
public void setMaxMessageSize(Integer maxMessageSize) {
|
||||
this.maxMessageSize = maxMessageSize;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,89 @@
|
||||
package org.springframework.cloud.stream.binder.rocketmq.provisioning;
|
||||
|
||||
import org.apache.rocketmq.client.Validators;
|
||||
import org.apache.rocketmq.client.exception.MQClientException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQProducerProperties;
|
||||
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProducerDestination;
|
||||
import org.springframework.cloud.stream.provisioning.ProvisioningException;
|
||||
import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
|
||||
|
||||
/**
|
||||
* @author Timur Valiev
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQTopicProvisioner
|
||||
implements
|
||||
ProvisioningProvider<ExtendedConsumerProperties<RocketMQConsumerProperties>,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>> {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(RocketMQTopicProvisioner.class);
|
||||
|
||||
@Override
|
||||
public ProducerDestination provisionProducerDestination(String name,
|
||||
ExtendedProducerProperties<RocketMQProducerProperties>
|
||||
properties)
|
||||
throws ProvisioningException {
|
||||
checkTopic(name);
|
||||
return new RocketProducerDestination(name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ConsumerDestination provisionConsumerDestination(String name, String group,
|
||||
ExtendedConsumerProperties<RocketMQConsumerProperties>
|
||||
properties)
|
||||
throws ProvisioningException {
|
||||
checkTopic(name);
|
||||
return new RocketConsumerDestination(name);
|
||||
}
|
||||
|
||||
private void checkTopic(String topic) {
|
||||
try {
|
||||
Validators.checkTopic(topic);
|
||||
} catch (MQClientException e) {
|
||||
logger.error("topic check error: " + topic, e);
|
||||
throw new AssertionError(e); // Can't happen
|
||||
}
|
||||
}
|
||||
|
||||
private static final class RocketProducerDestination implements ProducerDestination {
|
||||
|
||||
private final String producerDestinationName;
|
||||
|
||||
RocketProducerDestination(String destinationName) {
|
||||
this.producerDestinationName = destinationName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return producerDestinationName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getNameForPartition(int partition) {
|
||||
return producerDestinationName;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static final class RocketConsumerDestination implements ConsumerDestination {
|
||||
|
||||
private final String consumerDestinationName;
|
||||
|
||||
RocketConsumerDestination(String consumerDestinationName) {
|
||||
this.consumerDestinationName = consumerDestinationName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return this.consumerDestinationName;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1 @@
|
||||
rocketmq:org.springframework.cloud.stream.binder.rocketmq.config.RocketMQBinderAutoConfiguration
|
||||
@ -0,0 +1,2 @@
|
||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||
org.springframework.cloud.stream.binder.rocketmq.config.RocketMQBinderEndpointAutoConfiguration
|
||||
@ -0,0 +1,72 @@
|
||||
/*
|
||||
* Copyright (C) 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
|
||||
*
|
||||
* http://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 org.springframework.cloud.stream.binder.rocketmq;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.config.RocketMQBinderAutoConfiguration;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.config.RocketMQBinderEndpointAutoConfiguration;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
|
||||
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQExtendedBindingProperties;
|
||||
|
||||
/**
|
||||
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
|
||||
*/
|
||||
public class RocketMQAutoConfigurationTests {
|
||||
|
||||
private ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withConfiguration(
|
||||
AutoConfigurations.of(RocketMQBinderEndpointAutoConfiguration.class,
|
||||
RocketMQBinderAutoConfiguration.class))
|
||||
.withPropertyValues(
|
||||
"spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876",
|
||||
"spring.cloud.stream.bindings.output.destination=TopicOrderTest",
|
||||
"spring.cloud.stream.bindings.output.content-type=application/json",
|
||||
"spring.cloud.stream.bindings.input1.destination=TopicOrderTest",
|
||||
"spring.cloud.stream.bindings.input1.content-type=application/json",
|
||||
"spring.cloud.stream.bindings.input1.group=test-group1",
|
||||
"spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true",
|
||||
"spring.cloud.stream.bindings.input1.consumer.maxAttempts=1",
|
||||
"spring.cloud.stream.bindings.input2.destination=TopicOrderTest",
|
||||
"spring.cloud.stream.bindings.input2.content-type=application/json",
|
||||
"spring.cloud.stream.bindings.input2.group=test-group2",
|
||||
"spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false",
|
||||
"spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tag1");
|
||||
|
||||
@Test
|
||||
public void testProperties() {
|
||||
this.contextRunner.run(context -> {
|
||||
RocketMQBinderConfigurationProperties binderConfigurationProperties = context
|
||||
.getBean(RocketMQBinderConfigurationProperties.class);
|
||||
assertThat(binderConfigurationProperties.getNamesrvAddr())
|
||||
.isEqualTo("127.0.0.1:9876");
|
||||
RocketMQExtendedBindingProperties bindingProperties = context
|
||||
.getBean(RocketMQExtendedBindingProperties.class);
|
||||
assertThat(
|
||||
bindingProperties.getExtendedConsumerProperties("input2").getTags())
|
||||
.isEqualTo("tag1");
|
||||
assertThat(bindingProperties.getExtendedConsumerProperties("input2")
|
||||
.getOrderly()).isFalse();
|
||||
assertThat(bindingProperties.getExtendedConsumerProperties("input1")
|
||||
.getOrderly()).isTrue();
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue