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

Merge pull request #2 from spring-cloud-incubator/1.x

Browse Source 
1.x
This commit is contained in:
jackBao 2019-03-15 15:42:15 +08:00 committed by GitHub
commit 043e879522
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
58 changed files with 1594 additions and 1457 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cloud-alibaba-dependencies/pom.xml
View File

@ -25,7 +25,7 @@
<aliyun.sdk.version>4.0.1</aliyun.sdk.version>
<alicloud.context.version>1.0.5</alicloud.context.version>
<aliyun.sdk.edas.version>2.16.0</aliyun.sdk.edas.version>
<rocketmq.version>4.3.1</rocketmq.version>
<rocketmq.starter.version>2.0.2</rocketmq.starter.version>
<schedulerX.client.version>2.1.6</schedulerX.client.version>
<aliyun.java.sdk.dysmsapi>1.1.0</aliyun.java.sdk.dysmsapi>
<aliyun.sdk.mns>1.1.8</aliyun.sdk.mns>
@ -93,8 +93,8 @@
<!--Apache RocketMQ-->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>${rocketmq.version}</version>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>${rocketmq.starter.version}</version>
</dependency>
<!--Sentinel-->

3
spring-cloud-alibaba-examples/pom.xml
View File

@ -31,7 +31,8 @@
<module>fescar-example/storage-service</module>
<module>fescar-example/account-service</module>
<module>acm-example/acm-local-example</module>
<module>rocketmq-example</module>
<module>rocketmq-example/rocketmq-consume-example</module>
<module>rocketmq-example/rocketmq-produce-example</module>
<module>env-extension</module>
<module>sms-example</module>
<module>spring-cloud-bus-rocketmq-example</module>

13
spring-cloud-alibaba-examples/rocketmq-example/pom.xml → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-consume-example/pom.xml
View File

@ -6,13 +6,14 @@
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-alibaba-examples</artifactId>
<version>0.1.2.BUILD-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>rocketmq-example</artifactId>
<artifactId>rocketmq-consume-example</artifactId>
<packaging>jar</packaging>
<description>Example demonstrating how to use rocketmq</description>
<description>Example demonstrating how to use rocketmq consume</description>
<dependencies>
@ -20,18 +21,17 @@
<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>
<dependency>
<groupId>io.dropwizard.metrics</groupId>
<artifactId>metrics-core</artifactId>
</dependency>
</dependencies>
<build>
@ -52,4 +52,3 @@
</build>
</project>

14
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-consume-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java
View File

@ -24,15 +24,15 @@ public class Foo {
this.id = id;
}
public String getBar() {
return bar;
}
public String getBar() {
return bar;
}
public void setBar(String bar) {
this.bar = bar;
}
public void setBar(String bar) {
this.bar = bar;
}
@Override
@Override
public String toString() {
return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}';
}

5
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-consume-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java
View File

@ -25,11 +25,6 @@ public class ReceiveService {
System.out.println("input3 receive: " + foo);
}
@StreamListener("input1")
public void receiveInput1Again(String receiveMsg) {
System.out.println("input1 receive again: " + receiveMsg);
}
@StreamListener("input4")
public void receiveTransactionalMsg(String transactionMsg) {
System.out.println("input4 receive transaction msg: " + transactionMsg);

36
spring-cloud-alibaba-examples/rocketmq-example/rocketmq-consume-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQConsumerApplication.java Normal file
View File

@ -0,0 +1,36 @@
package org.springframework.cloud.alibaba.cloud.examples;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.alibaba.cloud.examples.RocketMQConsumerApplication.MySink;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@SpringBootApplication
@EnableBinding({ MySink.class })
public class RocketMQConsumerApplication {
public interface MySink {
@Input("input1")
SubscribableChannel input1();
@Input("input2")
SubscribableChannel input2();
@Input("input3")
SubscribableChannel input3();
@Input("input4")
SubscribableChannel input4();
}
public static void main(String[] args) {
SpringApplication.run(RocketMQConsumerApplication.class, args);
}
}

25
spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-consume-example/src/main/resources/application.properties
View File

@ -1,21 +1,9 @@
spring.cloud.stream.default-binder=rocketmq
spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876
spring.cloud.stream.bindings.output1.destination=test-topic
spring.cloud.stream.bindings.output1.content-type=application/json
spring.cloud.stream.bindings.output2.destination=TransactionTopic
spring.cloud.stream.bindings.output2.content-type=application/json
spring.cloud.stream.rocketmq.bindings.output2.producer.transactional=true
spring.cloud.stream.rocketmq.bindings.output2.producer.executer=org.springframework.cloud.alibaba.cloud.examples.MyTransactionExecuter
spring.cloud.stream.rocketmq.bindings.output2.producer.transaction-check-listener=org.springframework.cloud.alibaba.cloud.examples.MyTransactionCheckListener
spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876
spring.cloud.stream.bindings.input1.destination=test-topic
spring.cloud.stream.bindings.input1.content-type=text/plain
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=text/plain
@ -30,14 +18,15 @@ 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
spring.cloud.stream.bindings.input4.destination=TransactionTopic
spring.cloud.stream.bindings.input4.content-type=text/plain
spring.cloud.stream.bindings.input4.group=transaction-group
spring.cloud.stream.bindings.input4.consumer.concurrency=210
spring.cloud.stream.bindings.input4.consumer.concurrency=5
spring.application.name=rocketmq-example
spring.application.name=rocketmq-consume-example
server.port=28081
management.security.enabled=false
server.port=28082
management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always

54
spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/pom.xml Normal file
View File

@ -0,0 +1,54 @@
<?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.1.2.BUILD-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>rocketmq-produce-example</artifactId>
<packaging>jar</packaging>
<description>Example demonstrating how to use rocketmq produce</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>

39
spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java Normal file
View File

@ -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 bar;
public Foo() {
}
public Foo(int id, String bar) {
this.id = id;
this.bar = bar;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getBar() {
return bar;
}
public void setBar(String bar) {
this.bar = bar;
}
@Override
public String toString() {
return "Foo{" + "id=" + id + ", bar='" + bar + '\'' + '}';
}
}

35
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQProduceApplication.java
View File

@ -4,37 +4,18 @@ 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.alibaba.cloud.examples.RocketMQApplication.MySource;
import org.springframework.cloud.alibaba.cloud.examples.RocketMQProduceApplication.MySource;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@SpringBootApplication
@EnableBinding({ MySource.class, MySink.class })
public class RocketMQApplication {
public interface MySink {
@Input("input1")
SubscribableChannel input1();
@Input("input2")
SubscribableChannel input2();
@Input("input3")
SubscribableChannel input3();
@Input("input4")
SubscribableChannel input4();
}
@EnableBinding({ MySource.class })
public class RocketMQProduceApplication {
public interface MySource {
@Output("output1")
@ -45,7 +26,7 @@ public class RocketMQApplication {
}
public static void main(String[] args) {
SpringApplication.run(RocketMQApplication.class, args);
SpringApplication.run(RocketMQProduceApplication.class, args);
}
@Bean
@ -87,13 +68,13 @@ public class RocketMQApplication {
@Override
public void run(String... args) throws Exception {
// COMMIT_MESSAGE message
senderService.sendTransactionalMsg("transactional-msg1", false);
senderService.sendTransactionalMsg("transactional-msg1", 1);
// ROLLBACK_MESSAGE message
senderService.sendTransactionalMsg("transactional-msg2", true);
senderService.sendTransactionalMsg("transactional-msg2", 2);
// ROLLBACK_MESSAGE message
senderService.sendTransactionalMsg("transactional-msg3", true);
senderService.sendTransactionalMsg("transactional-msg3", 3);
// COMMIT_MESSAGE message
senderService.sendTransactionalMsg("transactional-msg4", false);
senderService.sendTransactionalMsg("transactional-msg4", 4);
}
}

20
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java → spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java
View File

@ -2,10 +2,13 @@ package org.springframework.cloud.alibaba.cloud.examples;
import java.util.HashMap;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySource;
import org.springframework.cloud.alibaba.cloud.examples.RocketMQProduceApplication.MySource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
@ -26,10 +29,10 @@ public class SenderService {
}
public <T> void sendWithTags(T msg, String tag) throws Exception {
Map<String, Object> map = new HashMap<>();
map.put(MessageConst.PROPERTY_TAGS, tag);
Message message = MessageBuilder.createMessage(msg, new MessageHeaders(map));
source.output1().send(message);
Map<String, Object> map = new HashMap<>();
map.put(MessageConst.PROPERTY_TAGS, tag);
Message message = MessageBuilder.createMessage(msg, new MessageHeaders(map));
source.output1().send(message);
}
public <T> void sendObject(T msg, String tag) throws Exception {
@ -40,12 +43,11 @@ public class SenderService {
source.output1().send(message);
}
public <T> void sendTransactionalMsg(T msg, boolean error) throws Exception {
public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
MessageBuilder builder = MessageBuilder.withPayload(msg)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
if (error) {
builder.setHeader("test", "1");
}
builder.setHeader("test", String.valueOf(num));
builder.setHeader(RocketMQHeaders.TAGS, "binder");
Message message = builder.build();
source.output2().send(message);
}

54
spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/TransactionListenerImpl.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.cloud.examples;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@RocketMQTransactionListener(txProducerGroup = "myTxProducerGroup", corePoolSize = 5, maximumPoolSize = 10)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
Object num = msg.getHeaders().get("test");
if ("1".equals(num)) {
System.out.println(
"executer: " + new String((byte[]) msg.getPayload()) + " unknown");
return RocketMQLocalTransactionState.UNKNOWN;
}
else if ("2".equals(num)) {
System.out.println(
"executer: " + new String((byte[]) msg.getPayload()) + " rollback");
return RocketMQLocalTransactionState.ROLLBACK;
}
System.out.println(
"executer: " + new String((byte[]) msg.getPayload()) + " commit");
return RocketMQLocalTransactionState.COMMIT;
}
@Override
public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
System.out.println("check: " + new String((byte[]) msg.getPayload()));
return RocketMQLocalTransactionState.COMMIT;
}
}

20
spring-cloud-alibaba-examples/rocketmq-example/rocketmq-produce-example/src/main/resources/application.properties Normal file
View File

@ -0,0 +1,20 @@
logging.level.org.springframework.cloud.stream.binder.rocketmq=DEBUG
spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876
spring.cloud.stream.bindings.output1.destination=test-topic
spring.cloud.stream.bindings.output1.content-type=application/json
spring.cloud.stream.rocketmq.bindings.output1.producer.group=binder-group
spring.cloud.stream.rocketmq.bindings.output1.producer.sync=true
spring.cloud.stream.bindings.output2.destination=TransactionTopic
spring.cloud.stream.bindings.output2.content-type=application/json
spring.cloud.stream.rocketmq.bindings.output2.producer.transactional=true
spring.cloud.stream.rocketmq.bindings.output2.producer.group=myTxProducerGroup
spring.application.name=rocketmq-produce-example
server.port=28081
management.endpoints.web.exposure.include=*
management.endpoint.health.show-details=always

18
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/MyTransactionCheckListener.java
View File

@ -1,18 +0,0 @@
package org.springframework.cloud.alibaba.cloud.examples;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class MyTransactionCheckListener implements TransactionCheckListener {
@Override
public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
System.out.println("TransactionCheckListener: " + new String(msg.getBody()));
return LocalTransactionState.COMMIT_MESSAGE;
}
}

20
spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/MyTransactionExecuter.java
View File

@ -1,20 +0,0 @@
package org.springframework.cloud.alibaba.cloud.examples;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.common.message.Message;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class MyTransactionExecuter implements LocalTransactionExecuter {
@Override
public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
if ("1".equals(msg.getUserProperty("test"))) {
System.out.println(new String(msg.getBody()) + " rollback");
return LocalTransactionState.ROLLBACK_MESSAGE;
}
System.out.println(new String(msg.getBody()) + " commit");
return LocalTransactionState.COMMIT_MESSAGE;
}
}

2
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/NacosConfigAutoConfiguration.java
View File

@ -17,6 +17,7 @@
package org.springframework.cloud.alibaba.nacos;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.alibaba.nacos.refresh.NacosContextRefresher;
import org.springframework.cloud.alibaba.nacos.refresh.NacosRefreshHistory;
import org.springframework.cloud.alibaba.nacos.refresh.NacosRefreshProperties;
@ -28,6 +29,7 @@ import org.springframework.context.annotation.Configuration;
* @author juven.xuxb
*/
@Configuration
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public class NacosConfigAutoConfiguration {
@Bean

2
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/NacosConfigBootstrapConfiguration.java
View File

@ -26,6 +26,7 @@ import org.springframework.context.annotation.Configuration;
* @author xiaojing
*/
@Configuration
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public class NacosConfigBootstrapConfiguration {
@Bean
@ -35,7 +36,6 @@ public class NacosConfigBootstrapConfiguration {
}
@Bean
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public NacosPropertySourceLocator nacosPropertySourceLocator(
NacosConfigProperties nacosConfigProperties) {
return new NacosPropertySourceLocator(nacosConfigProperties);

61
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/NacosConfigProperties.java
View File

@ -18,6 +18,7 @@ package org.springframework.cloud.alibaba.nacos;
import com.alibaba.nacos.api.NacosFactory;
import com.alibaba.nacos.api.config.ConfigService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -30,7 +31,14 @@ import java.util.List;
import java.util.Objects;
import java.util.Properties;
import static com.alibaba.nacos.api.PropertyKeyConst.*;
import static com.alibaba.nacos.api.PropertyKeyConst.ACCESS_KEY;
import static com.alibaba.nacos.api.PropertyKeyConst.CLUSTER_NAME;
import static com.alibaba.nacos.api.PropertyKeyConst.CONTEXT_PATH;
import static com.alibaba.nacos.api.PropertyKeyConst.ENCODE;
import static com.alibaba.nacos.api.PropertyKeyConst.ENDPOINT;
import static com.alibaba.nacos.api.PropertyKeyConst.NAMESPACE;
import static com.alibaba.nacos.api.PropertyKeyConst.SECRET_KEY;
import static com.alibaba.nacos.api.PropertyKeyConst.SERVER_ADDR;
/**
* nacos properties
@ -47,11 +55,6 @@ public class NacosConfigProperties {
private static final Logger log = LoggerFactory
.getLogger(NacosConfigProperties.class);
/**
* whether to enable nacos config.
*/
private boolean enabled = true;
/**
* nacos config server address
*/
@ -114,8 +117,6 @@ public class NacosConfigProperties {
private String name;
private String[] activeProfiles;
/**
* the dataids for configurable multiple shared configurations , multiple separated by
* commas .
@ -134,24 +135,8 @@ public class NacosConfigProperties {
private ConfigService configService;
@Autowired
private Environment environment;
@PostConstruct
public void init() {
this.activeProfiles = environment.getActiveProfiles();
}
// todo sts support
public boolean isEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String getServerAddr() {
return serverAddr;
}
@ -252,10 +237,6 @@ public class NacosConfigProperties {
return name;
}
public String[] getActiveProfiles() {
return activeProfiles;
}
public String getSharedDataids() {
return sharedDataids;
}
@ -284,10 +265,6 @@ public class NacosConfigProperties {
this.name = name;
}
public void setActiveProfiles(String[] activeProfiles) {
this.activeProfiles = activeProfiles;
}
public static class Config {
/**
* the data id of extended configuration
@ -329,17 +306,15 @@ public class NacosConfigProperties {
@Override
public String toString() {
return "NacosConfigProperties{" + "enabled=" + enabled + ", serverAddr='"
+ serverAddr + '\'' + ", encode='" + encode + '\'' + ", group='" + group
+ '\'' + ", prefix='" + prefix + '\'' + ", fileExtension='"
+ fileExtension + '\'' + ", timeout=" + timeout + ", endpoint='"
+ endpoint + '\'' + ", namespace='" + namespace + '\'' + ", accessKey='"
+ accessKey + '\'' + ", secretKey='" + secretKey + '\''
+ ", contextPath='" + contextPath + '\'' + ", clusterName='" + clusterName
+ '\'' + ", name='" + name + '\'' + ", activeProfiles="
+ Arrays.toString(activeProfiles) + ", sharedDataids='" + sharedDataids
+ '\'' + ", refreshableDataids='" + refreshableDataids + '\''
+ ", extConfig=" + extConfig + '}';
return "NacosConfigProperties{" + "serverAddr='" + serverAddr + '\''
+ ", encode='" + encode + '\'' + ", group='" + group + '\'' + ", prefix='"
+ prefix + '\'' + ", fileExtension='" + fileExtension + '\''
+ ", timeout=" + timeout + ", endpoint='" + endpoint + '\''
+ ", namespace='" + namespace + '\'' + ", accessKey='" + accessKey + '\''
+ ", secretKey='" + secretKey + '\'' + ", contextPath='" + contextPath
+ '\'' + ", clusterName='" + clusterName + '\'' + ", name='" + name + '\''
+ ", sharedDataids='" + sharedDataids + '\'' + ", refreshableDataids='"
+ refreshableDataids + '\'' + ", extConfig=" + extConfig + '}';
}
public ConfigService configServiceInstance() {

7
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/client/NacosPropertySourceBuilder.java
View File

@ -18,6 +18,7 @@ package org.springframework.cloud.alibaba.nacos.client;
import com.alibaba.nacos.api.config.ConfigService;
import com.alibaba.nacos.api.exception.NacosException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
@ -26,7 +27,11 @@ import org.springframework.core.io.ByteArrayResource;
import org.springframework.util.StringUtils;
import java.io.StringReader;
import java.util.*;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
/**
* @author xiaojing

19
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/client/NacosPropertySourceLocator.java
View File

@ -17,6 +17,7 @@
package org.springframework.cloud.alibaba.nacos.client;
import com.alibaba.nacos.api.config.ConfigService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.alibaba.nacos.NacosConfigProperties;
@ -70,7 +71,6 @@ public class NacosPropertySourceLocator implements PropertySourceLocator {
timeout);
String name = nacosConfigProperties.getName();
String nacosGroup = nacosConfigProperties.getGroup();
String dataIdPrefix = nacosConfigProperties.getPrefix();
if (StringUtils.isEmpty(dataIdPrefix)) {
dataIdPrefix = name;
@ -80,17 +80,12 @@ public class NacosPropertySourceLocator implements PropertySourceLocator {
dataIdPrefix = env.getProperty("spring.application.name");
}
List<String> profiles = Arrays.asList(env.getActiveProfiles());
nacosConfigProperties.setActiveProfiles(profiles.toArray(new String[0]));
String fileExtension = nacosConfigProperties.getFileExtension();
CompositePropertySource composite = new CompositePropertySource(
NACOS_PROPERTY_SOURCE_NAME);
loadSharedConfiguration(composite);
loadExtConfiguration(composite);
loadApplicationConfiguration(composite, nacosGroup, dataIdPrefix, fileExtension);
loadApplicationConfiguration(composite, dataIdPrefix, nacosConfigProperties, env);
return composite;
}
@ -151,11 +146,15 @@ public class NacosPropertySourceLocator implements PropertySourceLocator {
}
private void loadApplicationConfiguration(
CompositePropertySource compositePropertySource, String nacosGroup,
String dataIdPrefix, String fileExtension) {
CompositePropertySource compositePropertySource, String dataIdPrefix,
NacosConfigProperties properties, Environment environment) {
String fileExtension = properties.getFileExtension();
String nacosGroup = properties.getGroup();
loadNacosDataIfPresent(compositePropertySource,
dataIdPrefix + DOT + fileExtension, nacosGroup, fileExtension, true);
for (String profile : nacosConfigProperties.getActiveProfiles()) {
for (String profile : environment.getActiveProfiles()) {
String dataId = dataIdPrefix + SEP1 + profile + DOT + fileExtension;
loadNacosDataIfPresent(compositePropertySource, dataId, nacosGroup,
fileExtension, true);

2
spring-cloud-alibaba-nacos-config/src/main/java/org/springframework/cloud/alibaba/nacos/endpoint/NacosConfigEndpointAutoConfiguration.java
View File

@ -19,6 +19,7 @@ package org.springframework.cloud.alibaba.nacos.endpoint;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.cloud.alibaba.nacos.NacosConfigProperties;
import org.springframework.cloud.alibaba.nacos.refresh.NacosRefreshHistory;
@ -29,6 +30,7 @@ import org.springframework.context.annotation.Bean;
*/
@ConditionalOnWebApplication
@ConditionalOnClass(name = "org.springframework.boot.actuate.endpoint.AbstractEndpoint")
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public class NacosConfigEndpointAutoConfiguration {
@Autowired

6
spring-cloud-alibaba-nacos-config/src/test/java/org/springframework/cloud/alibaba/nacos/NacosConfigurationTests.java
View File

@ -163,7 +163,6 @@ public class NacosConfigurationTests {
checkoutNacosConfigFileExtension();
checkoutNacosConfigTimeout();
checkoutNacosConfigEncode();
checkoutNacosConfigProfiles();
checkoutEndpoint();
checkoutDataLoad();
@ -239,11 +238,6 @@ public class NacosConfigurationTests {
Assert.assertEquals(environment.getProperty("user.age"), "12");
}
private void checkoutNacosConfigProfiles() {
assertEquals("NacosConfigProperties' profiles is wrong",
new String[] { "dev", "test" }, properties.getActiveProfiles());
}
private void checkoutEndpoint() throws Exception {
NacosConfigEndpoint nacosConfigEndpoint = new NacosConfigEndpoint(properties,
refreshHistory);

6
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/NacosDiscoveryProperties.java
View File

@ -45,7 +45,7 @@ import static com.alibaba.nacos.api.PropertyKeyConst.*;
@ConfigurationProperties("spring.cloud.nacos.discovery")
public class NacosDiscoveryProperties {
private static final Logger LOGGER = LoggerFactory
private static final Logger log = LoggerFactory
.getLogger(NacosDiscoveryProperties.class);
/**
@ -155,7 +155,7 @@ public class NacosDiscoveryProperties {
serverAddr = Objects.toString(serverAddr, "");
if (serverAddr.lastIndexOf("/") != -1) {
serverAddr.substring(0, serverAddr.length() - 1);
serverAddr = serverAddr.substring(0, serverAddr.length() - 1);
}
endpoint = Objects.toString(endpoint, "");
namespace = Objects.toString(namespace, "");
@ -401,7 +401,7 @@ public class NacosDiscoveryProperties {
namingService = NacosFactory.createNamingService(properties);
}
catch (Exception e) {
LOGGER.error("create naming service error!properties={},e=,", this, e);
log.error("create naming service error!properties={},e=,", this, e);
return null;
}
return namingService;

14
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/discovery/NacosDiscoveryClient.java
View File

@ -34,8 +34,7 @@ import java.util.*;
*/
public class NacosDiscoveryClient implements DiscoveryClient {
private static final Logger LOGGER = LoggerFactory
.getLogger(NacosDiscoveryClient.class);
private static final Logger log = LoggerFactory.getLogger(NacosDiscoveryClient.class);
public static final String DESCRIPTION = "Spring Cloud Nacos Discovery Client";
private NacosDiscoveryProperties discoveryProperties;
@ -78,11 +77,12 @@ public class NacosDiscoveryClient implements DiscoveryClient {
nacosServiceInstance.setHost(instance.getIp());
nacosServiceInstance.setPort(instance.getPort());
nacosServiceInstance.setServiceId(serviceId);
Map<String, String> metadata = new HashMap<>();
metadata.put("instanceId", instance.getInstanceId());
metadata.put("weight", instance.getWeight() + "");
metadata.put("healthy", instance.isHealthy() + "");
metadata.put("cluster", instance.getClusterName() + "");
metadata.put("nacos.instanceId", instance.getInstanceId());
metadata.put("nacos.weight", instance.getWeight() + "");
metadata.put("nacos.healthy", instance.isHealthy() + "");
metadata.put("nacos.cluster", instance.getClusterName() + "");
metadata.putAll(instance.getMetadata());
nacosServiceInstance.setMetadata(metadata);
@ -111,7 +111,7 @@ public class NacosDiscoveryClient implements DiscoveryClient {
return services.getData();
}
catch (Exception e) {
LOGGER.error("get service name from nacos server fail,", e);
log.error("get service name from nacos server fail,", e);
return Collections.emptyList();
}
}

4
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/endpoint/NacosDiscoveryEndpoint.java
View File

@ -35,7 +35,7 @@ import org.springframework.cloud.alibaba.nacos.NacosDiscoveryProperties;
*/
public class NacosDiscoveryEndpoint extends AbstractEndpoint<Map<String, Object>> {
private static final Logger LOGGER = LoggerFactory
private static final Logger log = LoggerFactory
.getLogger(NacosDiscoveryEndpoint.class);
private NacosDiscoveryProperties nacosDiscoveryProperties;
@ -60,7 +60,7 @@ public class NacosDiscoveryEndpoint extends AbstractEndpoint<Map<String, Object>
subscribe = namingService.getSubscribeServices();
}
catch (Exception e) {
LOGGER.error("get subscribe services from nacos fail,", e);
log.error("get subscribe services from nacos fail,", e);
}
result.put("subscribe", subscribe);
return result;

2
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/endpoint/NacosDiscoveryEndpointAutoConfiguration.java
View File

@ -18,6 +18,7 @@ package org.springframework.cloud.alibaba.nacos.endpoint;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.alibaba.nacos.ConditionalOnNacosDiscoveryEnabled;
import org.springframework.cloud.alibaba.nacos.NacosDiscoveryProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@ -27,6 +28,7 @@ import org.springframework.context.annotation.Configuration;
*/
@Configuration
@ConditionalOnClass(name = "org.springframework.boot.actuate.endpoint.AbstractEndpoint")
@ConditionalOnNacosDiscoveryEnabled
public class NacosDiscoveryEndpointAutoConfiguration {
@Bean

4
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/registry/NacosAutoServiceRegistration.java
View File

@ -31,7 +31,7 @@ import org.springframework.util.StringUtils;
*/
public class NacosAutoServiceRegistration
extends AbstractAutoServiceRegistration<Registration> {
private static final Logger LOGGER = LoggerFactory
private static final Logger log = LoggerFactory
.getLogger(NacosAutoServiceRegistration.class);
private NacosRegistration registration;
@ -65,7 +65,7 @@ public class NacosAutoServiceRegistration
@Override
protected void register() {
if (!this.registration.getNacosDiscoveryProperties().isRegisterEnabled()) {
LOGGER.debug("Registration disabled.");
log.debug("Registration disabled.");
return;
}
if (this.registration.getPort() < 0) {

16
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/registry/NacosServiceRegistry.java
View File

@ -31,7 +31,7 @@ import org.springframework.util.StringUtils;
*/
public class NacosServiceRegistry implements ServiceRegistry<Registration> {
private static Logger logger = LoggerFactory.getLogger(NacosServiceRegistry.class);
private static final Logger log = LoggerFactory.getLogger(NacosServiceRegistry.class);
private final NacosDiscoveryProperties nacosDiscoveryProperties;
@ -46,7 +46,7 @@ public class NacosServiceRegistry implements ServiceRegistry<Registration> {
public void register(Registration registration) {
if (StringUtils.isEmpty(registration.getServiceId())) {
logger.info("No service to register for nacos client...");
log.warn("No service to register for nacos client...");
return;
}
@ -61,11 +61,11 @@ public class NacosServiceRegistry implements ServiceRegistry<Registration> {
try {
namingService.registerInstance(serviceId, instance);
logger.info("nacos registry, {} {}:{} register finished", serviceId,
log.info("nacos registry, {} {}:{} register finished", serviceId,
instance.getIp(), instance.getPort());
}
catch (Exception e) {
logger.error("nacos registry, {} register failed...{},", serviceId,
log.error("nacos registry, {} register failed...{},", serviceId,
registration.toString(), e);
}
}
@ -73,10 +73,10 @@ public class NacosServiceRegistry implements ServiceRegistry<Registration> {
@Override
public void deregister(Registration registration) {
logger.info("De-registering from Nacos Server now...");
log.info("De-registering from Nacos Server now...");
if (StringUtils.isEmpty(registration.getServiceId())) {
logger.info("No dom to de-register for nacos client...");
log.warn("No dom to de-register for nacos client...");
return;
}
@ -88,11 +88,11 @@ public class NacosServiceRegistry implements ServiceRegistry<Registration> {
registration.getPort(), nacosDiscoveryProperties.getClusterName());
}
catch (Exception e) {
logger.error("ERR_NACOS_DEREGISTER, de-register failed...{},",
log.error("ERR_NACOS_DEREGISTER, de-register failed...{},",
registration.toString(), e);
}
logger.info("De-registration finished.");
log.info("De-registration finished.");
}
@Override

2
spring-cloud-alibaba-nacos-discovery/src/main/java/org/springframework/cloud/alibaba/nacos/ribbon/RibbonNacosAutoConfiguration.java
View File

@ -19,6 +19,7 @@ package org.springframework.cloud.alibaba.nacos.ribbon;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.alibaba.nacos.ConditionalOnNacosDiscoveryEnabled;
import org.springframework.cloud.netflix.ribbon.RibbonAutoConfiguration;
import org.springframework.cloud.netflix.ribbon.RibbonClients;
import org.springframework.cloud.netflix.ribbon.SpringClientFactory;
@ -32,6 +33,7 @@ import org.springframework.context.annotation.Configuration;
@EnableConfigurationProperties
@ConditionalOnBean(SpringClientFactory.class)
@ConditionalOnRibbonNacos
@ConditionalOnNacosDiscoveryEnabled
@AutoConfigureAfter(RibbonAutoConfiguration.class)
@RibbonClients(defaultConfiguration = NacosRibbonClientConfiguration.class)
public class RibbonNacosAutoConfiguration {

6
spring-cloud-alibaba-nacos-discovery/src/main/resources/META-INF/additional-spring-configuration-metadata.json
View File

@ -5,6 +5,12 @@
"defaultValue": "${spring.application.name}",
"description": "the service name to register, default value is ${spring.application.name}."
},
{
"name": "spring.cloud.nacos.discovery.enabled",
"type": "java.lang.Boolean",
"defaultValue": true,
"description": "enable nacos discovery or not."
},
{
"name": "spring.cloud.nacos.discovery.namingLoadCacheAtStart",
"type": "java.lang.Boolean",

1
spring-cloud-alibaba-sentinel/src/main/java/org/springframework/cloud/alibaba/sentinel/custom/SentinelAutoConfiguration.java
View File

@ -166,6 +166,7 @@ public class SentinelAutoConfiguration {
@Bean
@ConditionalOnMissingBean
@ConditionalOnClass(name = "org.springframework.web.client.RestTemplate")
@ConditionalOnProperty(name = "resttemplate.sentinel.enabled", havingValue = "true", matchIfMissing = true)
public SentinelBeanPostProcessor sentinelBeanPostProcessor(
ApplicationContext applicationContext) {
return new SentinelBeanPostProcessor(applicationContext);

6
spring-cloud-alibaba-sentinel/src/main/resources/META-INF/additional-spring-configuration-metadata.json
View File

@ -6,6 +6,12 @@
"defaultValue": true,
"description": "enable or disable sentinel auto configure."
},
{
"name": "resttemplate.sentinel.enabled",
"type": "java.lang.Boolean",
"defaultValue": true,
"description": "enable or disable @SentinelRestTemplate."
},
{
"name": "spring.cloud.sentinel.eager",
"type": "java.lang.Boolean",

20
spring-cloud-starter-alibaba/spring-cloud-starter-bus-rocketmq/src/main/java/org/springframework/cloud/bus/rocketmq/env/RocketMQBusEnvironmentPostProcessor.java vendored
View File

@ -16,19 +16,20 @@
*/
package org.springframework.cloud.bus.rocketmq.env;
import static org.springframework.cloud.bus.SpringCloudBusClient.INPUT;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.env.EnvironmentPostProcessor;
import org.springframework.cloud.bus.BusEnvironmentPostProcessor;
import org.springframework.cloud.bus.SpringCloudBusClient;
import org.springframework.core.Ordered;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
import java.util.HashMap;
import java.util.Map;
/**
* The lowest precedence {@link EnvironmentPostProcessor} configures default RocketMQ Bus
* Properties that will be appended into {@link SpringApplication#defaultProperties}
@ -64,9 +65,16 @@ public class RocketMQBusEnvironmentPostProcessor
private void configureDefaultProperties(Map<String, Object> source) {
// Required Properties
String groupBindingPropertyName = createBindingPropertyName(
SpringCloudBusClient.INPUT, "group");
String groupBindingPropertyName = createBindingPropertyName(INPUT, "group");
String broadcastingPropertyName = createRocketMQPropertyName(INPUT,
"broadcasting");
source.put(groupBindingPropertyName, "rocketmq-bus-group");
source.put(broadcastingPropertyName, "true");
}
private String createRocketMQPropertyName(String channel, String propertyName) {
return "spring.cloud.stream.rocketmq.bindings." + INPUT + ".consumer."
+ propertyName;
}
private String createBindingPropertyName(String channel, String propertyName) {

21
spring-cloud-stream-binder-rocketmq/pom.xml
View File

@ -20,43 +20,32 @@
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
</dependency>
<dependency>
<groupId>io.dropwizard.metrics</groupId>
<artifactId>metrics-core</artifactId>
<scope>provided</scope>
<optional>true</optional>
</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.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-actuator</artifactId>
<scope>provided</scope>
<optional>true</optional>
</dependency>

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

@ -21,43 +21,16 @@ package org.springframework.cloud.stream.binder.rocketmq;
*/
public interface RocketMQBinderConstants {
String ENDPOINT_ID = "rocketmq_binder";
/**
* Header key
*/
String ORIGINAL_ROCKET_MESSAGE = "ORIGINAL_ROCKETMQ_MESSAGE";
String ROCKET_FLAG = "ROCKETMQ_FLAG";
String ROCKET_SEND_RESULT = "ROCKETMQ_SEND_RESULT";
String ROCKET_TRANSACTIONAL_ARG = "ROCKETMQ_TRANSACTIONAL_ARG";
String ACKNOWLEDGEMENT_KEY = "ACKNOWLEDGEMENT";
String ROCKET_TRANSACTIONAL_ARG = "TRANSACTIONAL_ARG";
/**
* Instrumentation
* Default value
*/
String LASTSEND_TIMESTAMP = "lastSend.timestamp";
String DEFAULT_NAME_SERVER = "127.0.0.1:9876";
interface Metrics {
interface Producer {
String PREFIX = "scs-rocketmq.producer.";
String TOTAL_SENT = "totalSent";
String TOTAL_SENT_FAILURES = "totalSentFailures";
String SENT_PER_SECOND = "sentPerSecond";
String SENT_FAILURES_PER_SECOND = "sentFailuresPerSecond";
}
interface Consumer {
String GROUP_PREFIX = "scs-rocketmq.consumerGroup.";
String PREFIX = "scs-rocketmq.consumer.";
String TOTAL_CONSUMED = "totalConsumed";
String CONSUMED_PER_SECOND = "consumedPerSecond";
String TOTAL_CONSUMED_FAILURES = "totalConsumedFailures";
String CONSUMED_FAILURES_PER_SECOND = "consumedFailuresPerSecond";
}
}
String DEFAULT_GROUP = "rocketmq_binder_default_group_name";
}

72
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQBinderUtils.java Normal file
View File

@ -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 org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
import org.springframework.util.StringUtils;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQBinderUtils {
public static RocketMQBinderConfigurationProperties mergeProperties(
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
RocketMQProperties rocketMQProperties) {
RocketMQBinderConfigurationProperties result = new RocketMQBinderConfigurationProperties();
if (StringUtils.isEmpty(rocketMQProperties.getNameServer())) {
result.setNameServer(rocketBinderConfigurationProperties.getNameServer());
}
else {
result.setNameServer(rocketMQProperties.getNameServer());
}
if (rocketMQProperties.getProducer() == null
|| StringUtils.isEmpty(rocketMQProperties.getProducer().getAccessKey())) {
result.setAccessKey(rocketBinderConfigurationProperties.getAccessKey());
}
else {
result.setAccessKey(rocketMQProperties.getProducer().getAccessKey());
}
if (rocketMQProperties.getProducer() == null
|| StringUtils.isEmpty(rocketMQProperties.getProducer().getSecretKey())) {
result.setSecretKey(rocketBinderConfigurationProperties.getSecretKey());
}
else {
result.setSecretKey(rocketMQProperties.getProducer().getSecretKey());
}
if (rocketMQProperties.getProducer() == null || StringUtils
.isEmpty(rocketMQProperties.getProducer().getCustomizedTraceTopic())) {
result.setCustomizedTraceTopic(
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
}
else {
result.setCustomizedTraceTopic(
rocketMQProperties.getProducer().getCustomizedTraceTopic());
}
if (rocketMQProperties.getProducer() != null
&& rocketMQProperties.getProducer().isEnableMsgTrace()) {
result.setEnableMsgTrace(Boolean.TRUE);
}
else {
result.setEnableMsgTrace(
rocketBinderConfigurationProperties.isEnableMsgTrace());
}
return result;
}
}

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

@ -16,16 +16,21 @@
package org.springframework.cloud.stream.binder.rocketmq;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
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.consuming.RocketMQListenerBindingContainer;
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;
@ -39,10 +44,11 @@ import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQMessageChannelBinder extends
@ -50,22 +56,22 @@ public class RocketMQMessageChannelBinder extends
implements
ExtendedPropertiesBinder<MessageChannel, RocketMQConsumerProperties, RocketMQProducerProperties> {
private static final Logger logger = LoggerFactory
.getLogger(RocketMQMessageChannelBinder.class);
private final RocketMQExtendedBindingProperties extendedBindingProperties;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final RocketMQProperties rocketMQProperties;
private final InstrumentationManager instrumentationManager;
private final ConsumersManager consumersManager;
public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
private Map<String, String> topicInUse = new HashMap<>();
public RocketMQMessageChannelBinder(
RocketMQExtendedBindingProperties extendedBindingProperties,
RocketMQTopicProvisioner provisioningProvider,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
RocketMQProperties rocketMQProperties,
InstrumentationManager instrumentationManager) {
super(true, null, provisioningProvider);
this.consumersManager = consumersManager;
this.extendedBindingProperties = extendedBindingProperties;
this.rocketMQProperties = rocketMQProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.instrumentationManager = instrumentationManager;
}
@ -75,23 +81,79 @@ public class RocketMQMessageChannelBinder extends
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
MessageChannel errorChannel) throws Exception {
if (producerProperties.getExtension().getEnabled()) {
RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
destination.getName(), producerProperties,
rocketBinderConfigurationProperties, instrumentationManager);
// if producerGroup is empty, using destination
String extendedProducerGroup = producerProperties.getExtension().getGroup();
String producerGroup = StringUtils.isEmpty(extendedProducerGroup)
? destination.getName()
: extendedProducerGroup;
RocketMQBinderConfigurationProperties mergedProperties = RocketMQBinderUtils
.mergeProperties(rocketBinderConfigurationProperties,
rocketMQProperties);
RocketMQTemplate rocketMQTemplate;
if (producerProperties.getExtension().getTransactional()) {
// transaction message check LocalTransactionExecuter
messageHandler.setLocalTransactionExecuter(
getClassConfiguration(destination.getName(),
producerProperties.getExtension().getExecuter(),
LocalTransactionExecuter.class));
// transaction message check TransactionCheckListener
messageHandler.setTransactionCheckListener(
getClassConfiguration(destination.getName(),
producerProperties.getExtension()
.getTransactionCheckListener(),
TransactionCheckListener.class));
Map<String, RocketMQTemplate> rocketMQTemplates = getApplicationContext()
.getParent().getBeansOfType(RocketMQTemplate.class);
if (rocketMQTemplates.size() == 0) {
throw new IllegalStateException(
"there is no RocketMQTemplate in Spring BeanFactory");
}
else if (rocketMQTemplates.size() > 1) {
throw new IllegalStateException(
"there is more than 1 RocketMQTemplates in Spring BeanFactory");
}
rocketMQTemplate = rocketMQTemplates.values().iterator().next();
}
else {
rocketMQTemplate = new RocketMQTemplate();
rocketMQTemplate.setObjectMapper(getApplicationContext().getParent()
.getBeansOfType(ObjectMapper.class).values().iterator().next());
DefaultMQProducer producer;
String ak = mergedProperties.getAccessKey();
String sk = mergedProperties.getSecretKey();
if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
RPCHook rpcHook = new AclClientRPCHook(
new SessionCredentials(ak, sk));
producer = new DefaultMQProducer(producerGroup, rpcHook,
mergedProperties.isEnableMsgTrace(),
mergedProperties.getCustomizedTraceTopic());
producer.setVipChannelEnabled(false);
producer.setInstanceName(
RocketMQUtil.getInstanceName(rpcHook, destination.getName()));
}
else {
producer = new DefaultMQProducer(producerGroup);
producer.setVipChannelEnabled(
producerProperties.getExtension().getVipChannelEnabled());
}
producer.setNamesrvAddr(mergedProperties.getNameServer());
producer.setSendMsgTimeout(
producerProperties.getExtension().getSendMessageTimeout());
producer.setRetryTimesWhenSendFailed(
producerProperties.getExtension().getRetryTimesWhenSendFailed());
producer.setRetryTimesWhenSendAsyncFailed(producerProperties
.getExtension().getRetryTimesWhenSendAsyncFailed());
producer.setCompressMsgBodyOverHowmuch(producerProperties.getExtension()
.getCompressMessageBodyThreshold());
producer.setRetryAnotherBrokerWhenNotStoreOK(
producerProperties.getExtension().isRetryNextServer());
producer.setMaxMessageSize(
producerProperties.getExtension().getMaxMessageSize());
rocketMQTemplate.setProducer(producer);
}
RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
rocketMQTemplate, destination.getName(), producerGroup,
producerProperties.getExtension().getTransactional(),
instrumentationManager);
messageHandler.setBeanFactory(this.getApplicationContext().getBeanFactory());
messageHandler.setSync(producerProperties.getExtension().getSync());
if (errorChannel != null) {
messageHandler.setSendFailureChannel(errorChannel);
}
return messageHandler;
}
else {
@ -107,12 +169,25 @@ public class RocketMQMessageChannelBinder extends
throws Exception {
if (group == null || "".equals(group)) {
throw new RuntimeException(
"'group' must be configured for channel + " + destination.getName());
"'group must be configured for channel " + destination.getName());
}
RocketMQListenerBindingContainer listenerContainer = new RocketMQListenerBindingContainer(
consumerProperties, rocketBinderConfigurationProperties, this);
listenerContainer.setConsumerGroup(group);
listenerContainer.setTopic(destination.getName());
listenerContainer.setConsumeThreadMax(consumerProperties.getConcurrency());
listenerContainer.setSuspendCurrentQueueTimeMillis(
consumerProperties.getExtension().getSuspendCurrentQueueTimeMillis());
listenerContainer.setDelayLevelWhenNextConsume(
consumerProperties.getExtension().getDelayLevelWhenNextConsume());
listenerContainer
.setNameServer(rocketBinderConfigurationProperties.getNameServer());
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
consumersManager, consumerProperties, destination.getName(), group,
instrumentationManager);
listenerContainer, consumerProperties, instrumentationManager);
topicInUse.put(destination.getName(), group);
ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination,
group, consumerProperties);
@ -140,45 +215,7 @@ public class RocketMQMessageChannelBinder extends
return extendedBindingProperties.getExtendedProducerProperties(channelName);
}
private <T> T getClassConfiguration(String destName, String className,
Class<T> interfaceClass) {
if (StringUtils.isEmpty(className)) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, should set "
+ interfaceClass.getSimpleName() + " configuration"
+ interfaceClass.getSimpleName() + " should be set, like "
+ "'spring.cloud.stream.rocketmq.bindings.output.producer.xxx=TheFullClassNameOfYour"
+ interfaceClass.getSimpleName() + "'");
}
else if (StringUtils.isNotEmpty(className)) {
Class fieldClass;
// check class exists
try {
fieldClass = ClassUtils.forName(className,
RocketMQMessageChannelBinder.class.getClassLoader());
}
catch (ClassNotFoundException e) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " class is not found");
}
// check interface incompatible
if (!interfaceClass.isAssignableFrom(fieldClass)) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " is incompatible with " + interfaceClass.getSimpleName()
+ " interface");
}
try {
return (T) fieldClass.newInstance();
}
catch (Exception e) {
throw new RuntimeException("Binding for channel " + destName
+ " using transactional message, but " + className
+ " instance error", e);
}
}
return null;
public Map<String, String> getTopicInUse() {
return topicInUse;
}
}

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

@ -1,139 +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.stream.binder.rocketmq;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ACKNOWLEDGEMENT_KEY;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ORIGINAL_ROCKET_MESSAGE;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_FLAG;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_SEND_RESULT;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang3.math.NumberUtils;
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;
/**
* @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 getMessageHeaders().get(MessageConst.PROPERTY_TAGS) == null ? ""
: (String) getMessageHeaders().get(MessageConst.PROPERTY_TAGS);
}
public RocketMQMessageHeaderAccessor withTags(String tag) {
setHeader(MessageConst.PROPERTY_TAGS, tag);
return this;
}
public String getKeys() {
return getMessageHeaders().get(MessageConst.PROPERTY_KEYS) == null ? ""
: (String) getMessageHeaders().get(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 NumberUtils.toInt(
(String) getMessageHeaders().get(MessageConst.PROPERTY_DELAY_TIME_LEVEL),
0);
}
public RocketMQMessageHeaderAccessor withDelayTimeLevel(Integer delayTimeLevel) {
setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel);
return this;
}
public Integer getFlag() {
return NumberUtils.toInt((String) getMessageHeaders().get(ROCKET_FLAG), 0);
}
public RocketMQMessageHeaderAccessor withFlag(Integer delayTimeLevel) {
setHeader(ROCKET_FLAG, delayTimeLevel);
return this;
}
public Object getTransactionalArg() {
return getMessageHeaders().get(ROCKET_TRANSACTIONAL_ARG);
}
public Object withTransactionalArg(Object arg) {
setHeader(ROCKET_TRANSACTIONAL_ARG, arg);
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;
}
}

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

@ -1,55 +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.stream.binder.rocketmq.actuator;
import static org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ENDPOINT_ID;
import java.util.HashMap;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.endpoint.AbstractEndpoint;
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQBinderEndpoint extends AbstractEndpoint<Map<String, Object>> {
@Autowired(required = false)
private InstrumentationManager instrumentationManager;
public RocketMQBinderEndpoint() {
super(ENDPOINT_ID);
}
@Override
public Map<String, Object> invoke() {
Map<String, Object> result = new HashMap<>();
if (instrumentationManager != null) {
result.put("metrics", instrumentationManager.getMetricRegistry());
result.put("runtime", instrumentationManager.getRuntime());
}
else {
result.put("warning",
"please add metrics-core dependency, we use it for metrics");
}
return result;
}
}

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

@ -28,45 +28,37 @@ import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationM
*/
public class RocketMQBinderHealthIndicator extends AbstractHealthIndicator {
@Autowired(required = false)
@Autowired
private InstrumentationManager instrumentationManager;
@Override
protected void doHealthCheck(Health.Builder builder) throws Exception {
int upCount = 0, outOfServiceCount = 0;
if (instrumentationManager != null) {
for (Instrumentation instrumentation : instrumentationManager
.getHealthInstrumentations()) {
if (instrumentation.isUp()) {
upCount++;
}
else if (instrumentation.isOutOfService()) {
upCount++;
}
for (Instrumentation instrumentation : instrumentationManager
.getHealthInstrumentations()) {
if (instrumentation.isUp()) {
upCount++;
}
if (upCount == instrumentationManager.getHealthInstrumentations().size()) {
builder.up();
return;
}
else if (outOfServiceCount == instrumentationManager
.getHealthInstrumentations().size()) {
builder.outOfService();
return;
}
builder.down();
for (Instrumentation instrumentation : instrumentationManager
.getHealthInstrumentations()) {
if (!instrumentation.isStarted()) {
builder.withException(instrumentation.getStartException());
}
else if (instrumentation.isOutOfService()) {
upCount++;
}
}
else {
builder.down();
builder.withDetail("warning",
"please add metrics-core dependency, we use it for metrics");
if (upCount == instrumentationManager.getHealthInstrumentations().size()) {
builder.up();
return;
}
else if (outOfServiceCount == instrumentationManager.getHealthInstrumentations()
.size()) {
builder.outOfService();
return;
}
builder.down();
for (Instrumentation instrumentation : instrumentationManager
.getHealthInstrumentations()) {
if (!instrumentation.isStarted()) {
builder.withException(instrumentation.getStartException());
}
}
}
}

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

@ -16,23 +16,26 @@
package org.springframework.cloud.stream.binder.rocketmq.config;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
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;
import org.springframework.context.annotation.Import;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@Configuration
@Import({ RocketMQAutoConfiguration.class,
RocketMQBinderHealthIndicatorAutoConfiguration.class })
@EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
RocketMQExtendedBindingProperties.class })
public class RocketMQBinderAutoConfiguration {
@ -42,7 +45,7 @@ public class RocketMQBinderAutoConfiguration {
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
@Autowired(required = false)
private InstrumentationManager instrumentationManager;
private RocketMQProperties rocketMQProperties = new RocketMQProperties();
@Autowired
public RocketMQBinderAutoConfiguration(
@ -50,8 +53,6 @@ public class RocketMQBinderAutoConfiguration {
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties) {
this.extendedBindingProperties = extendedBindingProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
System.setProperty(ClientLogger.CLIENT_LOG_LEVEL,
this.rocketBinderConfigurationProperties.getLogLevel());
}
@Bean
@ -62,17 +63,17 @@ public class RocketMQBinderAutoConfiguration {
@Bean
public RocketMQMessageChannelBinder rocketMessageChannelBinder(
RocketMQTopicProvisioner provisioningProvider,
ConsumersManager consumersManager) {
InstrumentationManager instrumentationManager) {
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
consumersManager, extendedBindingProperties, provisioningProvider,
rocketBinderConfigurationProperties, instrumentationManager);
extendedBindingProperties, provisioningProvider,
rocketBinderConfigurationProperties, rocketMQProperties,
instrumentationManager);
return binder;
}
@Bean
public ConsumersManager consumersManager() {
return new ConsumersManager(instrumentationManager,
rocketBinderConfigurationProperties);
public InstrumentationManager instrumentationManager() {
return new InstrumentationManager();
}
}

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

@ -16,13 +16,8 @@
package org.springframework.cloud.stream.binder.rocketmq.config;
import org.springframework.boot.actuate.autoconfigure.EndpointAutoConfiguration;
import org.springframework.boot.actuate.endpoint.Endpoint;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
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;
@ -30,24 +25,12 @@ import org.springframework.context.annotation.Configuration;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@Configuration
@AutoConfigureAfter(EndpointAutoConfiguration.class)
@ConditionalOnClass(Endpoint.class)
public class RocketMQBinderEndpointAutoConfiguration {
@Bean
public RocketMQBinderEndpoint rocketBinderEndpoint() {
return new RocketMQBinderEndpoint();
}
@ConditionalOnClass(name = "org.springframework.boot.actuate.endpoint.AbstractEndpoint")
public class RocketMQBinderHealthIndicatorAutoConfiguration {
@Bean
public RocketMQBinderHealthIndicator rocketBinderHealthIndicator() {
return new RocketMQBinderHealthIndicator();
}
@Bean
@ConditionalOnClass(name = "com.codahale.metrics.Counter")
public InstrumentationManager instrumentationManager() {
return new InstrumentationManager();
}
}

105
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQComponent4BinderAutoConfiguration.java Normal file
View File

@ -0,0 +1,105 @@
/*
* 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.config;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
import org.apache.rocketmq.spring.config.RocketMQTransactionAnnotationProcessor;
import org.apache.rocketmq.spring.config.TransactionHandlerRegistry;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@Configuration
@AutoConfigureAfter(RocketMQAutoConfiguration.class)
@ConditionalOnMissingBean(DefaultMQProducer.class)
public class RocketMQComponent4BinderAutoConfiguration {
private final Environment environment;
public RocketMQComponent4BinderAutoConfiguration(Environment environment) {
this.environment = environment;
}
@Bean
@ConditionalOnMissingBean(DefaultMQProducer.class)
public DefaultMQProducer defaultMQProducer() {
DefaultMQProducer producer;
String configNameServer = environment.resolveRequiredPlaceholders(
"${spring.cloud.stream.rocketmq.binder.name-server:${rocketmq.producer.name-server:}}");
String ak = environment.resolveRequiredPlaceholders(
"${spring.cloud.stream.rocketmq.binder.access-key:${rocketmq.producer.access-key:}}");
String sk = environment.resolveRequiredPlaceholders(
"${spring.cloud.stream.rocketmq.binder.secret-key:${rocketmq.producer.secret-key:}}");
if (!org.springframework.util.StringUtils.isEmpty(ak)
&& !org.springframework.util.StringUtils.isEmpty(sk)) {
producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP,
new AclClientRPCHook(new SessionCredentials(ak, sk)));
producer.setVipChannelEnabled(false);
}
else {
producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP);
}
producer.setNamesrvAddr(configNameServer);
return producer;
}
@Bean(destroyMethod = "destroy")
@ConditionalOnMissingBean
public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
ObjectMapper objectMapper) {
RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
rocketMQTemplate.setProducer(mqProducer);
rocketMQTemplate.setObjectMapper(objectMapper);
return rocketMQTemplate;
}
@Bean
@ConditionalOnBean(RocketMQTemplate.class)
@ConditionalOnMissingBean(TransactionHandlerRegistry.class)
public TransactionHandlerRegistry transactionHandlerRegistry(
RocketMQTemplate template) {
return new TransactionHandlerRegistry(template);
}
@Bean(name = RocketMQConfigUtils.ROCKETMQ_TRANSACTION_ANNOTATION_PROCESSOR_BEAN_NAME)
@ConditionalOnBean(TransactionHandlerRegistry.class)
public static RocketMQTransactionAnnotationProcessor transactionAnnotationProcessor(
TransactionHandlerRegistry transactionHandlerRegistry) {
return new RocketMQTransactionAnnotationProcessor(transactionHandlerRegistry);
}
@Bean
@ConditionalOnMissingBean(ObjectMapper.class)
public ObjectMapper rocketMQBinderObjectMapper() {
return new ObjectMapper();
}
}

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

@ -1,94 +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.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;
}
}

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

@ -1,135 +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.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 InstrumentationManager instrumentationManager;
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 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);
if (instrumentationManager != null) {
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 = null;
if (instrumentationManager != null) {
groupInstrumentation = instrumentationManager
.getConsumerGroupInstrumentation(group);
instrumentationManager.addHealthInstrumentation(groupInstrumentation);
}
try {
consumerGroups.get(group).start();
started.put(group, true);
if (groupInstrumentation != null) {
groupInstrumentation.markStartedSuccessfully();
}
}
catch (MQClientException e) {
if (groupInstrumentation != null) {
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();
}
}

418
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java Normal file
View File

@ -0,0 +1,418 @@
/*
* 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.consuming;
import java.util.List;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.apache.rocketmq.spring.support.RocketMQListenerContainer;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQListenerBindingContainer
implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
private final static Logger log = LoggerFactory
.getLogger(RocketMQListenerBindingContainer.class);
private long suspendCurrentQueueTimeMillis = 1000;
/**
* Message consume retry strategy<br>
* -1,no retry,put into DLQ directly<br>
* 0,broker control retry frequency<br>
* >0,client control retry frequency.
*/
private int delayLevelWhenNextConsume = 0;
private String nameServer;
private String consumerGroup;
private String topic;
private int consumeThreadMax = 64;
private String charset = "UTF-8";
private RocketMQListener rocketMQListener;
private DefaultMQPushConsumer consumer;
private boolean running;
private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
// The following properties came from RocketMQConsumerProperties.
private ConsumeMode consumeMode;
private SelectorType selectorType;
private String selectorExpression;
private MessageModel messageModel;
public RocketMQListenerBindingContainer(
ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
RocketMQMessageChannelBinder rocketMQMessageChannelBinder) {
this.rocketMQConsumerProperties = rocketMQConsumerProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.rocketMQMessageChannelBinder = rocketMQMessageChannelBinder;
this.consumeMode = rocketMQConsumerProperties.getExtension().getOrderly()
? ConsumeMode.ORDERLY
: ConsumeMode.CONCURRENTLY;
if (StringUtils.isEmpty(rocketMQConsumerProperties.getExtension().getSql())) {
this.selectorType = SelectorType.TAG;
this.selectorExpression = rocketMQConsumerProperties.getExtension().getTags();
}
else {
this.selectorType = SelectorType.SQL92;
this.selectorExpression = rocketMQConsumerProperties.getExtension().getSql();
}
this.messageModel = rocketMQConsumerProperties.getExtension().getBroadcasting()
? MessageModel.BROADCASTING
: MessageModel.CLUSTERING;
}
@Override
public void setupMessageListener(RocketMQListener<?> rocketMQListener) {
this.rocketMQListener = rocketMQListener;
}
@Override
public void destroy() throws Exception {
this.setRunning(false);
if (consumer != null) {
consumer.shutdown();
}
log.info("container destroyed, {}", this.toString());
}
@Override
public void afterPropertiesSet() throws Exception {
initRocketMQPushConsumer();
}
@Override
public boolean isAutoStartup() {
return true;
}
@Override
public void stop(Runnable callback) {
stop();
callback.run();
}
@Override
public void start() {
if (this.isRunning()) {
throw new IllegalStateException(
"container already running. " + this.toString());
}
try {
consumer.start();
}
catch (MQClientException e) {
throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
}
this.setRunning(true);
log.info("running container: {}", this.toString());
}
@Override
public void stop() {
if (this.isRunning()) {
if (consumer != null) {
consumer.shutdown();
}
setRunning(false);
}
}
@Override
public boolean isRunning() {
return running;
}
private void setRunning(boolean running) {
this.running = running;
}
@Override
public int getPhase() {
return Integer.MAX_VALUE;
}
private void initRocketMQPushConsumer() throws MQClientException {
Assert.notNull(rocketMQListener, "Property 'rocketMQListener' is required");
Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
Assert.notNull(nameServer, "Property 'nameServer' is required");
Assert.notNull(topic, "Property 'topic' is required");
String ak = rocketBinderConfigurationProperties.getAccessKey();
String sk = rocketBinderConfigurationProperties.getSecretKey();
if (!StringUtils.isEmpty(ak) && !StringUtils.isEmpty(sk)) {
RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(ak, sk));
consumer = new DefaultMQPushConsumer(consumerGroup, rpcHook,
new AllocateMessageQueueAveragely(),
rocketBinderConfigurationProperties.isEnableMsgTrace(),
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook, topic));
consumer.setVipChannelEnabled(false);
}
else {
consumer = new DefaultMQPushConsumer(consumerGroup,
rocketBinderConfigurationProperties.isEnableMsgTrace(),
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
}
consumer.setNamesrvAddr(nameServer);
consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());
switch (messageModel) {
case BROADCASTING:
consumer.setMessageModel(
org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
break;
case CLUSTERING:
consumer.setMessageModel(
org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
break;
default:
throw new IllegalArgumentException("Property 'messageModel' was wrong.");
}
switch (selectorType) {
case TAG:
consumer.subscribe(topic, selectorExpression);
break;
case SQL92:
consumer.subscribe(topic, MessageSelector.bySql(selectorExpression));
break;
default:
throw new IllegalArgumentException("Property 'selectorType' was wrong.");
}
switch (consumeMode) {
case ORDERLY:
consumer.setMessageListener(new DefaultMessageListenerOrderly());
break;
case CONCURRENTLY:
consumer.setMessageListener(new DefaultMessageListenerConcurrently());
break;
default:
throw new IllegalArgumentException("Property 'consumeMode' was wrong.");
}
if (rocketMQListener instanceof RocketMQPushConsumerLifecycleListener) {
((RocketMQPushConsumerLifecycleListener) rocketMQListener)
.prepareStart(consumer);
}
}
@Override
public String toString() {
return "RocketMQListenerBindingContainer{" + "consumerGroup='" + consumerGroup
+ '\'' + ", nameServer='" + nameServer + '\'' + ", topic='" + topic + '\''
+ ", consumeMode=" + consumeMode + ", selectorType=" + selectorType
+ ", selectorExpression='" + selectorExpression + '\'' + ", messageModel="
+ messageModel + '}';
}
public long getSuspendCurrentQueueTimeMillis() {
return suspendCurrentQueueTimeMillis;
}
public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
}
public int getDelayLevelWhenNextConsume() {
return delayLevelWhenNextConsume;
}
public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
}
public String getNameServer() {
return nameServer;
}
public void setNameServer(String nameServer) {
this.nameServer = nameServer;
}
public String getConsumerGroup() {
return consumerGroup;
}
public void setConsumerGroup(String consumerGroup) {
this.consumerGroup = consumerGroup;
}
public String getTopic() {
return topic;
}
public void setTopic(String topic) {
this.topic = topic;
}
public int getConsumeThreadMax() {
return consumeThreadMax;
}
public void setConsumeThreadMax(int consumeThreadMax) {
this.consumeThreadMax = consumeThreadMax;
}
public String getCharset() {
return charset;
}
public void setCharset(String charset) {
this.charset = charset;
}
public RocketMQListener getRocketMQListener() {
return rocketMQListener;
}
public void setRocketMQListener(RocketMQListener rocketMQListener) {
this.rocketMQListener = rocketMQListener;
}
public DefaultMQPushConsumer getConsumer() {
return consumer;
}
public void setConsumer(DefaultMQPushConsumer consumer) {
this.consumer = consumer;
}
public ExtendedConsumerProperties<RocketMQConsumerProperties> getRocketMQConsumerProperties() {
return rocketMQConsumerProperties;
}
public ConsumeMode getConsumeMode() {
return consumeMode;
}
public SelectorType getSelectorType() {
return selectorType;
}
public String getSelectorExpression() {
return selectorExpression;
}
public MessageModel getMessageModel() {
return messageModel;
}
public class DefaultMessageListenerConcurrently
implements MessageListenerConcurrently {
@SuppressWarnings("unchecked")
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
for (MessageExt messageExt : msgs) {
log.debug("received msg: {}", messageExt);
try {
long now = System.currentTimeMillis();
rocketMQListener
.onMessage(RocketMQUtil.convertToSpringMessage(messageExt));
long costTime = System.currentTimeMillis() - now;
log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
}
catch (Exception e) {
log.warn("consume message failed. messageExt:{}", messageExt, e);
context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
@SuppressWarnings("unchecked")
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
for (MessageExt messageExt : msgs) {
log.debug("received msg: {}", messageExt);
try {
long now = System.currentTimeMillis();
rocketMQListener
.onMessage(RocketMQUtil.convertToSpringMessage(messageExt));
long costTime = System.currentTimeMillis() - now;
log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
}
catch (Exception e) {
log.warn("consume message failed. messageExt:{}", messageExt, e);
context.setSuspendCurrentQueueTimeMillis(
suspendCurrentQueueTimeMillis);
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
}
return ConsumeOrderlyStatus.SUCCESS;
}
}
}

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

@ -16,149 +16,116 @@
package org.springframework.cloud.stream.binder.rocketmq.integration;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
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.apache.rocketmq.spring.core.RocketMQListener;
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.consuming.RocketMQListenerBindingContainer;
import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
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.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.MessagingException;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.StringUtils;
import org.springframework.util.Assert;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
private static final Logger logger = LoggerFactory
private static final Logger log = LoggerFactory
.getLogger(RocketMQInboundChannelAdapter.class);
private ConsumerInstrumentation consumerInstrumentation;
private InstrumentationManager instrumentationManager;
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
private final String destination;
private final String group;
private final ConsumersManager consumersManager;
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
public RocketMQInboundChannelAdapter(ConsumersManager consumersManager,
private RocketMQListenerBindingContainer rocketMQListenerContainer;
private final ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties;
private final InstrumentationManager instrumentationManager;
public RocketMQInboundChannelAdapter(
RocketMQListenerBindingContainer rocketMQListenerContainer,
ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties,
String destination, String group,
InstrumentationManager instrumentationManager) {
this.consumersManager = consumersManager;
this.rocketMQListenerContainer = rocketMQListenerContainer;
this.consumerProperties = consumerProperties;
this.destination = destination;
this.group = group;
this.instrumentationManager = instrumentationManager;
}
@Override
protected void onInit() {
if (consumerProperties == null
|| !consumerProperties.getExtension().getEnabled()) {
return;
}
super.onInit();
if (this.retryTemplate != null) {
Assert.state(getErrorChannel() == null,
"Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
+ "send an error message when retries are exhausted");
}
BindingRocketMQListener listener = new BindingRocketMQListener();
rocketMQListenerContainer.setRocketMQListener(listener);
if (retryTemplate != null) {
this.retryTemplate.registerListener(listener);
}
try {
rocketMQListenerContainer.afterPropertiesSet();
}
catch (Exception e) {
log.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
throw new IllegalArgumentException(
"rocketMQListenerContainer init error: " + e.getMessage(), e);
}
instrumentationManager.addHealthInstrumentation(
new Instrumentation(rocketMQListenerContainer.getTopic()
+ rocketMQListenerContainer.getConsumerGroup()));
}
@Override
protected void doStart() {
if (consumerProperties == null
|| !consumerProperties.getExtension().getEnabled()) {
return;
}
String tags = consumerProperties.getExtension().getTags();
Boolean isOrderly = consumerProperties.getExtension().getOrderly();
DefaultMQPushConsumer consumer = consumersManager.getOrCreateConsumer(group,
destination, consumerProperties);
final CloudStreamMessageListener listener = isOrderly
? new CloudStreamMessageListenerOrderly()
: new CloudStreamMessageListenerConcurrently();
if (retryTemplate != null) {
retryTemplate.registerListener(listener);
}
Set<String> tagsSet = new HashSet<>();
if (!StringUtils.isEmpty(tags)) {
for (String tag : tags.split("\\|\\|")) {
tagsSet.add(tag.trim());
}
}
if (instrumentationManager != null) {
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,
org.apache.commons.lang3.StringUtils.join(tagsSet, " || "));
}
if (consumerInstrumentation != null) {
consumerInstrumentation.markStartedSuccessfully();
}
rocketMQListenerContainer.start();
instrumentationManager
.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
+ rocketMQListenerContainer.getConsumerGroup())
.markStartedSuccessfully();
}
catch (MQClientException e) {
if (consumerInstrumentation != null) {
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);
catch (Exception e) {
instrumentationManager
.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
+ rocketMQListenerContainer.getConsumerGroup())
.markStartFailed(e);
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
throw new MessagingException(MessageBuilder.withPayload(
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
.build(), e);
}
}
@Override
protected void doStop() {
consumersManager.stopConsumer(group);
rocketMQListenerContainer.stop();
}
public void setRetryTemplate(RetryTemplate retryTemplate) {
@ -169,84 +136,28 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
this.recoveryCallback = recoveryCallback;
}
protected class CloudStreamMessageListener implements MessageListener, RetryListener {
protected class BindingRocketMQListener
implements RocketMQListener<Message>, RetryListener {
Acknowledgement consumeMessage(final List<MessageExt> msgs) {
@Override
public void onMessage(final Message message) {
boolean enableRetry = RocketMQInboundChannelAdapter.this.retryTemplate != null;
try {
if (enableRetry) {
return RocketMQInboundChannelAdapter.this.retryTemplate
.execute(new RetryCallback<Acknowledgement, Exception>() {
@Override
public Acknowledgement doWithRetry(RetryContext context)
throws Exception {
return 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);
if (RocketMQInboundChannelAdapter.this.instrumentationManager != null) {
RocketMQInboundChannelAdapter.this.instrumentationManager
.getConsumerInstrumentation(
RocketMQInboundChannelAdapter.this.destination)
.markConsumed();
}
return result;
}
if (enableRetry) {
RocketMQInboundChannelAdapter.this.retryTemplate
.execute(new RetryCallback<Object, RuntimeException>() {
@Override
public Object doWithRetry(RetryContext context)
throws RuntimeException {
RocketMQInboundChannelAdapter.this.sendMessage(message);
return null;
}
}, (RecoveryCallback<Object>) RocketMQInboundChannelAdapter.this.recoveryCallback);
}
catch (Exception e) {
logger.error(
"RocketMQ Message hasn't been processed successfully. Caused by ",
e);
if (RocketMQInboundChannelAdapter.this.instrumentationManager != null) {
RocketMQInboundChannelAdapter.this.instrumentationManager
.getConsumerInstrumentation(
RocketMQInboundChannelAdapter.this.destination)
.markConsumedFailure();
}
throw new RuntimeException(
"RocketMQ Message hasn't been processed successfully. Caused by ",
e);
else {
RocketMQInboundChannelAdapter.this.sendMessage(message);
}
}
private Acknowledgement doSendMsgs(final List<MessageExt> msgs,
RetryContext context) {
List<Acknowledgement> acknowledgements = new ArrayList<>();
for (MessageExt msg : msgs) {
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) {
@ -256,54 +167,13 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
@Override
public <T, E extends Throwable> void close(RetryContext context,
RetryCallback<T, E> callback, Throwable throwable) {
if (RocketMQInboundChannelAdapter.this.instrumentationManager == null) {
return;
}
if (throwable != null) {
RocketMQInboundChannelAdapter.this.instrumentationManager
.getConsumerInstrumentation(
RocketMQInboundChannelAdapter.this.destination)
.markConsumedFailure();
}
else {
RocketMQInboundChannelAdapter.this.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 {
@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 {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeOrderlyContext context) {
Acknowledgement acknowledgement = consumeMessage(msgs);
context.setSuspendCurrentQueueTimeMillis(
(acknowledgement.getConsumeOrderlySuspendCurrentQueueTimeMill()));
return acknowledgement.getConsumeOrderlyStatus();
}
}
}

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

@ -16,110 +16,91 @@
package org.springframework.cloud.stream.binder.rocketmq.integration;
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.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQMessageHeaderAccessor;
import org.springframework.cloud.stream.binder.rocketmq.metrics.Instrumentation;
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.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
private DefaultMQProducer producer;
private final static Logger log = LoggerFactory
.getLogger(RocketMQMessageHandler.class);
private ProducerInstrumentation producerInstrumentation;
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
private InstrumentationManager instrumentationManager;
private MessageChannel sendFailureChannel;
private LocalTransactionExecuter localTransactionExecuter;
private final RocketMQTemplate rocketMQTemplate;
private TransactionCheckListener transactionCheckListener;
private final ExtendedProducerProperties<RocketMQProducerProperties> producerProperties;
private final Boolean transactional;
private final String destination;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final String groupName;
private final InstrumentationManager instrumentationManager;
private boolean sync = false;
private volatile boolean running = false;
public RocketMQMessageHandler(String destination,
ExtendedProducerProperties<RocketMQProducerProperties> producerProperties,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
public RocketMQMessageHandler(RocketMQTemplate rocketMQTemplate, String destination,
String group, Boolean transactional,
InstrumentationManager instrumentationManager) {
this.rocketMQTemplate = rocketMQTemplate;
this.destination = destination;
this.producerProperties = producerProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.groupName = group;
this.transactional = transactional;
this.instrumentationManager = instrumentationManager;
}
@Override
public void start() {
if (producerProperties.getExtension().getTransactional()) {
producer = new TransactionMQProducer(destination);
if (transactionCheckListener != null) {
((TransactionMQProducer) producer)
.setTransactionCheckListener(transactionCheckListener);
if (!transactional) {
instrumentationManager
.addHealthInstrumentation(new Instrumentation(destination));
try {
rocketMQTemplate.afterPropertiesSet();
instrumentationManager.getHealthInstrumentation(destination)
.markStartedSuccessfully();
}
}
else {
producer = new DefaultMQProducer(destination);
}
if (instrumentationManager != null) {
producerInstrumentation = instrumentationManager
.getProducerInstrumentation(destination);
instrumentationManager.addHealthInstrumentation(producerInstrumentation);
}
producer.setNamesrvAddr(rocketBinderConfigurationProperties.getNamesrvAddr());
if (producerProperties.getExtension().getMaxMessageSize() > 0) {
producer.setMaxMessageSize(
producerProperties.getExtension().getMaxMessageSize());
}
try {
producer.start();
if (producerInstrumentation != null) {
producerInstrumentation.markStartedSuccessfully();
catch (Exception e) {
instrumentationManager.getHealthInstrumentation(destination)
.markStartFailed(e);
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
throw new MessagingException(MessageBuilder.withPayload(
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
.build(), e);
}
}
catch (MQClientException e) {
if (producerInstrumentation != null) {
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();
if (!transactional) {
rocketMQTemplate.destroy();
}
running = false;
}
@ -130,76 +111,121 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
}
@Override
protected void handleMessageInternal(org.springframework.messaging.Message<?> message)
throws Exception {
protected void handleMessageInternal(
final org.springframework.messaging.Message<?> message) throws Exception {
try {
Message toSend;
if (message.getPayload() instanceof byte[]) {
toSend = new Message(destination, (byte[]) message.getPayload());
final StringBuilder topicWithTags = new StringBuilder(destination);
String tags = null;
if (message.getHeaders().get(RocketMQHeaders.TAGS) != null) {
tags = message.getHeaders().get(RocketMQHeaders.TAGS).toString();
}
else if (message.getPayload() instanceof String) {
toSend = new Message(destination,
((String) message.getPayload()).getBytes());
if (!StringUtils.isEmpty(tags)) {
topicWithTags.append(":").append(tags);
}
SendResult sendRes = null;
if (transactional) {
sendRes = rocketMQTemplate.sendMessageInTransaction(groupName,
topicWithTags.toString(), message, message.getHeaders()
.get(RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG));
log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
}
else {
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());
}
int delayLevel = 0;
try {
Object delayLevelObj = message.getHeaders()
.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
if (delayLevelObj instanceof Number) {
delayLevel = ((Number) delayLevelObj).intValue();
}
else if (delayLevelObj instanceof String) {
delayLevel = Integer.parseInt((String) delayLevelObj);
}
}
catch (Exception e) {
// ignore
}
if (sync) {
sendRes = rocketMQTemplate.syncSend(topicWithTags.toString(), message,
rocketMQTemplate.getProducer().getSendMsgTimeout(),
delayLevel);
log.debug("sync send to topic " + topicWithTags + " " + sendRes);
}
else {
rocketMQTemplate.asyncSend(topicWithTags.toString(), message,
new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.debug("async send to topic " + topicWithTags + " "
+ sendResult);
}
SendResult sendRes;
if (producerProperties.getExtension().getTransactional()) {
sendRes = producer.sendMessageInTransaction(toSend,
localTransactionExecuter, headerAccessor.getTransactionalArg());
@Override
public void onException(Throwable e) {
log.error(
"RocketMQ Message hasn't been sent. Caused by "
+ e.getMessage());
if (getSendFailureChannel() != null) {
getSendFailureChannel().send(
RocketMQMessageHandler.this.errorMessageStrategy
.buildErrorMessage(
new MessagingException(
message, e),
null));
}
}
});
}
}
else {
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);
}
if (instrumentationManager != null) {
instrumentationManager.getRuntime().put(
RocketMQBinderConstants.LASTSEND_TIMESTAMP,
System.currentTimeMillis());
producerInstrumentation.markSent();
if (sendRes != null && !sendRes.getSendStatus().equals(SendStatus.SEND_OK)) {
if (getSendFailureChannel() != null) {
this.getSendFailureChannel().send(message);
}
else {
throw new MessagingException(message,
new MQClientException("message hasn't been sent", null));
}
}
}
catch (MQClientException | RemotingException | MQBrokerException
| InterruptedException | UnsupportedOperationException e) {
if (producerInstrumentation != null) {
producerInstrumentation.markSentFailure();
catch (Exception e) {
log.error("RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
if (getSendFailureChannel() != null) {
getSendFailureChannel().send(this.errorMessageStrategy
.buildErrorMessage(new MessagingException(message, e), null));
}
else {
throw new MessagingException(message, e);
}
logger.error(
"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
throw new MessagingException(e.getMessage(), e);
}
}
public void setLocalTransactionExecuter(
LocalTransactionExecuter localTransactionExecuter) {
this.localTransactionExecuter = localTransactionExecuter;
/**
* Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent
* to this channel with a payload of a {@link MessagingException} with the failed
* message and cause.
* @param sendFailureChannel the failure channel.
* @since 0.2.2
*/
public void setSendFailureChannel(MessageChannel sendFailureChannel) {
this.sendFailureChannel = sendFailureChannel;
}
public void setTransactionCheckListener(
TransactionCheckListener transactionCheckListener) {
this.transactionCheckListener = transactionCheckListener;
/**
* Set the error message strategy implementation to use when sending error messages
* after send failures. Cannot be null.
* @param errorMessageStrategy the implementation.
* @since 0.2.2
*/
public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null");
this.errorMessageStrategy = errorMessageStrategy;
}
public MessageChannel getSendFailureChannel() {
return sendFailureChannel;
}
public void setSync(boolean sync) {
this.sync = sync;
}
}

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

@ -1,33 +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.stream.binder.rocketmq.metrics;
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;
public ConsumerGroupInstrumentation(MetricRegistry metricRegistry, String name) {
super(name);
this.metricRegistry = metricRegistry;
}
}

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

@ -1,59 +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.stream.binder.rocketmq.metrics;
import static com.codahale.metrics.MetricRegistry.name;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
/**
* @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, Consumer.TOTAL_CONSUMED));
this.consumedPerSecond = registry
.meter(name(baseMetricName, Consumer.CONSUMED_PER_SECOND));
this.totalConsumedFailures = registry
.counter(name(baseMetricName, Consumer.TOTAL_CONSUMED_FAILURES));
this.consumedFailuresPerSecond = registry
.meter(name(baseMetricName, Consumer.CONSUMED_FAILURES_PER_SECOND));
}
public void markConsumed() {
totalConsumed.inc();
consumedPerSecond.mark();
}
public void markConsumedFailure() {
totalConsumedFailures.inc();
consumedFailuresPerSecond.mark();
}
}

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

@ -23,44 +23,44 @@ import java.util.concurrent.atomic.AtomicBoolean;
* @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;
private final String name;
protected final AtomicBoolean started = new AtomicBoolean(false);
protected Exception startException = null;
Instrumentation(String name) {
this.name = name;
}
public Instrumentation(String name) {
this.name = name;
}
public boolean isDown() {
return startException != null;
}
public boolean isDown() {
return startException != null;
}
public boolean isUp() {
return started.get();
}
public boolean isUp() {
return started.get();
}
public boolean isOutOfService() {
return !started.get() && startException == null;
}
public boolean isOutOfService() {
return !started.get() && startException == null;
}
public void markStartedSuccessfully() {
started.set(true);
}
public void markStartedSuccessfully() {
started.set(true);
}
public void markStartFailed(Exception e) {
started.set(false);
startException = e;
}
public void markStartFailed(Exception e) {
started.set(false);
startException = e;
}
public String getName() {
return name;
}
public String getName() {
return name;
}
public boolean isStarted() {
return started.get();
}
public boolean isStarted() {
return started.get();
}
public Exception getStartException() {
return startException;
}
public Exception getStartException() {
return startException;
}
}

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

@ -20,57 +20,18 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import com.codahale.metrics.MetricRegistry;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Consumer;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
import java.util.concurrent.ConcurrentHashMap;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class InstrumentationManager {
private final MetricRegistry metricRegistry = new MetricRegistry();
private final Map<String, Object> runtime = new HashMap<>();
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, Object> runtime = new ConcurrentHashMap<>();
private final Map<String, Instrumentation> healthInstrumentations = new HashMap<>();
public ProducerInstrumentation getProducerInstrumentation(String destination) {
String key = Producer.PREFIX + destination;
ProducerInstrumentation producerInstrumentation = producerInstrumentations
.get(key);
if (producerInstrumentation == null) {
producerInstrumentations.put(key,
new ProducerInstrumentation(metricRegistry, key));
}
return producerInstrumentations.get(key);
}
public ConsumerInstrumentation getConsumerInstrumentation(String destination) {
String key = Consumer.PREFIX + destination;
ConsumerInstrumentation consumerInstrumentation = consumeInstrumentations
.get(key);
if (consumerInstrumentation == null) {
consumeInstrumentations.put(key,
new ConsumerInstrumentation(metricRegistry, key));
}
return consumeInstrumentations.get(key);
}
public ConsumerGroupInstrumentation getConsumerGroupInstrumentation(String group) {
String key = Consumer.GROUP_PREFIX + group;
ConsumerGroupInstrumentation consumerGroupInstrumentation = consumerGroupsInstrumentations
.get(key);
if (consumerGroupInstrumentation == null) {
consumerGroupsInstrumentations.put(key,
new ConsumerGroupInstrumentation(metricRegistry, key));
}
return consumerGroupsInstrumentations.get(key);
}
public Set<Instrumentation> getHealthInstrumentations() {
return new HashSet<>(healthInstrumentations.values());
}
@ -79,11 +40,12 @@ public class InstrumentationManager {
healthInstrumentations.put(instrumentation.getName(), instrumentation);
}
public Instrumentation getHealthInstrumentation(String key) {
return healthInstrumentations.get(key);
}
public Map<String, Object> getRuntime() {
return runtime;
}
public MetricRegistry getMetricRegistry() {
return metricRegistry;
}
}

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

@ -1,59 +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.stream.binder.rocketmq.metrics;
import static com.codahale.metrics.MetricRegistry.name;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants.Metrics.Producer;
import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
/**
* @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, Producer.TOTAL_SENT));
this.totalSentFailures = registry
.counter(name(baseMetricName, Producer.TOTAL_SENT_FAILURES));
this.sentPerSecond = registry
.meter(name(baseMetricName, Producer.SENT_PER_SECOND));
this.sentFailuresPerSecond = registry
.meter(name(baseMetricName, Producer.SENT_FAILURES_PER_SECOND));
}
public void markSent() {
totalSent.inc();
sentPerSecond.mark();
}
public void markSentFailure() {
totalSentFailures.inc();
sentFailuresPerSecond.mark();
}
}

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

@ -16,7 +16,9 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import org.apache.rocketmq.common.MixAll;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
/**
* @author Timur Valiev
@ -25,24 +27,69 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
@ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
public class RocketMQBinderConfigurationProperties {
private String namesrvAddr = "127.0.0.1:9876";
/**
* The name server for rocketMQ, formats: `host:port;host:port`.
*/
private String nameServer = RocketMQBinderConstants.DEFAULT_NAME_SERVER;
private String logLevel = "ERROR";
/**
* The property of "access-key".
*/
private String accessKey;
public String getNamesrvAddr() {
return namesrvAddr;
}
/**
* The property of "secret-key".
*/
private String secretKey;
public void setNamesrvAddr(String namesrvAddr) {
this.namesrvAddr = namesrvAddr;
}
/**
* Switch flag instance for message trace.
*/
private boolean enableMsgTrace = true;
public String getLogLevel() {
return logLevel;
}
/**
* The name value of message trace topic.If you don't config,you can use the default
* trace topic name.
*/
private String customizedTraceTopic = MixAll.RMQ_SYS_TRACE_TOPIC;
public void setLogLevel(String logLevel) {
this.logLevel = logLevel;
}
public String getNameServer() {
return nameServer;
}
public void setNameServer(String nameServer) {
this.nameServer = nameServer;
}
public String getAccessKey() {
return accessKey;
}
public void setAccessKey(String accessKey) {
this.accessKey = accessKey;
}
public String getSecretKey() {
return secretKey;
}
public void setSecretKey(String secretKey) {
this.secretKey = secretKey;
}
public boolean isEnableMsgTrace() {
return enableMsgTrace;
}
public void setEnableMsgTrace(boolean enableMsgTrace) {
this.enableMsgTrace = enableMsgTrace;
}
public String getCustomizedTraceTopic() {
return customizedTraceTopic;
}
public void setCustomizedTraceTopic(String customizedTraceTopic) {
this.customizedTraceTopic = customizedTraceTopic;
}
}

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

@ -28,68 +28,93 @@ import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
*/
public class RocketMQConsumerProperties {
/**
* using '||' to split tag
* {@link MQPushConsumer#subscribe(String, String)}
*/
private String tags;
/**
* 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 MQPushConsumer#subscribe(String, MessageSelector)}
* {@link MessageSelector#bySql(String)}
*/
private String sql;
/**
* {@link MessageModel#BROADCASTING}
*/
private Boolean broadcasting = false;
/**
* {@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;
/**
* if orderly is true, using {@link MessageListenerOrderly} else if orderly if false,
* using {@link MessageListenerConcurrently}
*/
private Boolean orderly = false;
private Boolean enabled = true;
/**
* for concurrently listener. message consume retry strategy
*/
private int delayLevelWhenNextConsume = 0;
public String getTags() {
return tags;
}
/**
* for orderly listener. next retry delay time
*/
private long suspendCurrentQueueTimeMillis = 1000;
public void setTags(String tags) {
this.tags = tags;
}
private Boolean enabled = true;
public String getSql() {
return sql;
}
public String getTags() {
return tags;
}
public void setSql(String sql) {
this.sql = sql;
}
public void setTags(String tags) {
this.tags = tags;
}
public Boolean getOrderly() {
return orderly;
}
public String getSql() {
return sql;
}
public void setOrderly(Boolean orderly) {
this.orderly = orderly;
}
public void setSql(String sql) {
this.sql = sql;
}
public Boolean getEnabled() {
return enabled;
}
public Boolean getOrderly() {
return orderly;
}
public void setEnabled(Boolean enabled) {
this.enabled = enabled;
}
public void setOrderly(Boolean orderly) {
this.orderly = orderly;
}
public Boolean getBroadcasting() {
return broadcasting;
}
public Boolean getEnabled() {
return enabled;
}
public void setBroadcasting(Boolean broadcasting) {
this.broadcasting = broadcasting;
}
public void setEnabled(Boolean enabled) {
this.enabled = enabled;
}
public Boolean getBroadcasting() {
return broadcasting;
}
public void setBroadcasting(Boolean broadcasting) {
this.broadcasting = broadcasting;
}
public int getDelayLevelWhenNextConsume() {
return delayLevelWhenNextConsume;
}
public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
}
public long getSuspendCurrentQueueTimeMillis() {
return suspendCurrentQueueTimeMillis;
}
public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
}
}

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

@ -17,8 +17,6 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionExecuter;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
/**
* @author Timur Valiev
@ -29,21 +27,61 @@ public class RocketMQProducerProperties {
private Boolean enabled = true;
/**
* Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
* Name of producer.
*/
private Integer maxMessageSize = 0;
private String group;
/**
* Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}.
*/
private Integer maxMessageSize = 1024 * 1024 * 4;
private Boolean transactional = false;
/**
* full class name of {@link LocalTransactionExecuter}
*/
private String executer;
private Boolean sync = false;
private Boolean vipChannelEnabled = true;
/**
* full class name of {@link TransactionCheckListener}
* Millis of send message timeout.
*/
private String transactionCheckListener;
private int sendMessageTimeout = 3000;
/**
* Compress message body threshold, namely, message body larger than 4k will be
* compressed on default.
*/
private int compressMessageBodyThreshold = 1024 * 4;
/**
* Maximum number of retry to perform internally before claiming sending failure in
* synchronous mode. This may potentially cause message duplication which is up to
* application developers to resolve.
*/
private int retryTimesWhenSendFailed = 2;
/**
* <p>
* Maximum number of retry to perform internally before claiming sending failure in
* asynchronous mode.
* </p>
* This may potentially cause message duplication which is up to application
* developers to resolve.
*/
private int retryTimesWhenSendAsyncFailed = 2;
/**
* Indicate whether to retry another broker on sending failure internally.
*/
private boolean retryNextServer = false;
public String getGroup() {
return group;
}
public void setGroup(String group) {
this.group = group;
}
public Boolean getEnabled() {
return enabled;
@ -69,19 +107,60 @@ public class RocketMQProducerProperties {
this.transactional = transactional;
}
public String getExecuter() {
return executer;
public Boolean getSync() {
return sync;
}
public void setExecuter(String executer) {
this.executer = executer;
public void setSync(Boolean sync) {
this.sync = sync;
}
public String getTransactionCheckListener() {
return transactionCheckListener;
public Boolean getVipChannelEnabled() {
return vipChannelEnabled;
}
public void setTransactionCheckListener(String transactionCheckListener) {
this.transactionCheckListener = transactionCheckListener;
public void setVipChannelEnabled(Boolean vipChannelEnabled) {
this.vipChannelEnabled = vipChannelEnabled;
}
public int getSendMessageTimeout() {
return sendMessageTimeout;
}
public void setSendMessageTimeout(int sendMessageTimeout) {
this.sendMessageTimeout = sendMessageTimeout;
}
public int getCompressMessageBodyThreshold() {
return compressMessageBodyThreshold;
}
public void setCompressMessageBodyThreshold(int compressMessageBodyThreshold) {
this.compressMessageBodyThreshold = compressMessageBodyThreshold;
}
public int getRetryTimesWhenSendFailed() {
return retryTimesWhenSendFailed;
}
public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
}
public int getRetryTimesWhenSendAsyncFailed() {
return retryTimesWhenSendAsyncFailed;
}
public void setRetryTimesWhenSendAsyncFailed(int retryTimesWhenSendAsyncFailed) {
this.retryTimesWhenSendAsyncFailed = retryTimesWhenSendAsyncFailed;
}
public boolean isRetryNextServer() {
return retryNextServer;
}
public void setRetryNextServer(boolean retryNextServer) {
this.retryNextServer = retryNextServer;
}
}

2
spring-cloud-stream-binder-rocketmq/src/main/resources/META-INF/spring.factories
View File

@ -1,2 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.binder.rocketmq.config.RocketMQBinderEndpointAutoConfiguration
org.springframework.cloud.stream.binder.rocketmq.config.RocketMQComponent4BinderAutoConfiguration

10
spring-cloud-stream-binder-rocketmq/src/test/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQAutoConfigurationTests.java
View File

@ -22,7 +22,6 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.boot.builder.SpringApplicationBuilder;
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;
import org.springframework.context.ConfigurableApplicationContext;
@ -36,10 +35,9 @@ public class RocketMQAutoConfigurationTests {
@Before
public void setUp() throws Exception {
this.context = new SpringApplicationBuilder(
RocketMQBinderEndpointAutoConfiguration.class,
RocketMQBinderAutoConfiguration.class).web(false).run(
"--spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876",
this.context = new SpringApplicationBuilder(RocketMQBinderAutoConfiguration.class)
.web(false)
.run("--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",
@ -58,7 +56,7 @@ public class RocketMQAutoConfigurationTests {
public void testProperties() {
RocketMQBinderConfigurationProperties binderConfigurationProperties = context
.getBean(RocketMQBinderConfigurationProperties.class);
assertThat(binderConfigurationProperties.getNamesrvAddr())
assertThat(binderConfigurationProperties.getNameServer())
.isEqualTo("127.0.0.1:9876");
RocketMQExtendedBindingProperties bindingProperties = context
.getBean(RocketMQExtendedBindingProperties.class);