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

upgrade rocketmq spring version

Browse Source
This commit is contained in:
fangjian0423 2019-06-12 16:52:47 +08:00
parent 36eda9bd1c
commit 4a1e51d839
5 changed files with 34 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -27,7 +27,7 @@
<aliyun.sdk.version>4.4.1</aliyun.sdk.version>
<alicloud.context.version>1.0.5</alicloud.context.version>
<aliyun.sdk.edas.version>2.44.0</aliyun.sdk.edas.version>
<rocketmq.starter.version>2.0.2</rocketmq.starter.version>
<rocketmq.starter.version>2.0.3</rocketmq.starter.version>
<schedulerX.client.version>2.1.6</schedulerX.client.version>
<dubbo.version>2.7.1</dubbo.version>
<dubbo-spring-boot.version>2.7.1</dubbo-spring-boot.version>

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

@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.rocketmq;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
import org.springframework.util.StringUtils;
@ -66,6 +67,14 @@ public class RocketMQBinderUtils {
result.setEnableMsgTrace(
rocketBinderConfigurationProperties.isEnableMsgTrace());
}
if (!StringUtils.isEmpty(rocketMQProperties.getAccessChannel())) {
result.setAccessChannel(
AccessChannel.valueOf(rocketMQProperties.getAccessChannel()));
}
else {
result.setAccessChannel(
rocketBinderConfigurationProperties.getAccessChannel());
}
return result;
}

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

@ -134,6 +134,11 @@ public class RocketMQMessageChannelBinder extends
producer.setVipChannelEnabled(
producerProperties.getExtension().getVipChannelEnabled());
}
if (mergedProperties.getAccessChannel() != null) {
producer.setAccessChannel(mergedProperties.getAccessChannel());
}
producer.setNamesrvAddr(mergedProperties.getNameServer());
producer.setSendMsgTimeout(
producerProperties.getExtension().getSendMessageTimeout());

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

@ -215,6 +215,11 @@ public class RocketMQListenerBindingContainer
rocketBinderConfigurationProperties.getCustomizedTraceTopic());
}
if (rocketBinderConfigurationProperties.getAccessChannel() != null) {
consumer.setAccessChannel(
rocketBinderConfigurationProperties.getAccessChannel());
}
consumer.setNamesrvAddr(nameServer);
consumer.setConsumeThreadMax(rocketMQConsumerProperties.getConcurrency());
consumer.setConsumeThreadMin(rocketMQConsumerProperties.getConcurrency());

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

@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.common.MixAll;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
@ -53,6 +54,11 @@ public class RocketMQBinderConfigurationProperties {
*/
private String customizedTraceTopic = MixAll.RMQ_SYS_TRACE_TOPIC;
/**
* Enum type for accessChannel, values: LOCAL, CLOUD.
*/
private AccessChannel accessChannel = AccessChannel.LOCAL;
public String getNameServer() {
return nameServer;
}
@ -92,4 +98,12 @@ public class RocketMQBinderConfigurationProperties {
public void setCustomizedTraceTopic(String customizedTraceTopic) {
this.customizedTraceTopic = customizedTraceTopic;
}
public AccessChannel getAccessChannel() {
return accessChannel;
}
public void setAccessChannel(AccessChannel accessChannel) {
this.accessChannel = accessChannel;
}
}