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

sync rocketmq binder to 1.x

Browse Source
This commit is contained in:
fangjian0423 2019-02-28 16:39:22 +08:00
parent 348f699eb8
commit 97ddae22fc
22 changed files with 473 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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>

9
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,14 +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>
</dependencies>
<build>
@ -48,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 + '\'' + '}';
}

0
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

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);
}
}

27
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,39 +1,32 @@
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.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
spring.cloud.stream.bindings.input2.group=test-group2
spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false
spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr
spring.cloud.stream.bindings.input2.consumer.concurrency=5
spring.cloud.stream.bindings.input2.consumer.concurrency=20
spring.cloud.stream.bindings.input2.consumer.maxAttempts=1
spring.cloud.stream.bindings.input3.destination=test-topic
spring.cloud.stream.bindings.input3.content-type=application/json
spring.cloud.stream.bindings.input3.group=test-group3
spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj
spring.cloud.stream.bindings.input3.consumer.concurrency=5
spring.cloud.stream.bindings.input3.consumer.concurrency=20
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=10
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);
}

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

@ -27,13 +27,20 @@ import org.springframework.messaging.Message;
/**
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@RocketMQTransactionListener(txProducerGroup = "TransactionTopic", corePoolSize = 5, maximumPoolSize = 10)
class TransactionListenerImpl implements RocketMQLocalTransactionListener {
@RocketMQTransactionListener(txProducerGroup = "myTxProducerGroup", corePoolSize = 5, maximumPoolSize = 10)
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
@Override
public RocketMQLocalTransactionState executeLocalTransaction(Message msg,
Object arg) {
if ("1".equals(((HashMap) msg.getHeaders().get(RocketMQHeaders.PROPERTIES))
.get("USERS_test"))) {
Object num = ((HashMap) msg.getHeaders().get(RocketMQHeaders.PROPERTIES))
.get("USERS_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;

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

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 mergeProducerProperties(
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;
}
}

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

@ -21,8 +21,13 @@ import java.util.HashSet;
import java.util.Map;
import java.util.Set;
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;
@ -41,6 +46,7 @@ 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.StringUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -54,6 +60,7 @@ public class RocketMQMessageChannelBinder extends
private final RocketMQExtendedBindingProperties extendedBindingProperties;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final RocketMQProperties rocketMQProperties;
private final InstrumentationManager instrumentationManager;
private Set<String> clientConfigId = new HashSet<>();
@ -63,9 +70,11 @@ public class RocketMQMessageChannelBinder extends
RocketMQExtendedBindingProperties extendedBindingProperties,
RocketMQTopicProvisioner provisioningProvider,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
RocketMQProperties rocketMQProperties,
InstrumentationManager instrumentationManager) {
super(true, null, provisioningProvider);
this.extendedBindingProperties = extendedBindingProperties;
this.rocketMQProperties = rocketMQProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.instrumentationManager = instrumentationManager;
}
@ -76,6 +85,10 @@ public class RocketMQMessageChannelBinder extends
MessageChannel errorChannel) throws Exception {
if (producerProperties.getExtension().getEnabled()) {
RocketMQBinderConfigurationProperties mergedProperties = RocketMQBinderUtils
.mergeProducerProperties(rocketBinderConfigurationProperties,
rocketMQProperties);
RocketMQTemplate rocketMQTemplate;
if (producerProperties.getExtension().getTransactional()) {
Map<String, RocketMQTemplate> rocketMQTemplates = getApplicationContext()
@ -95,9 +108,27 @@ public class RocketMQMessageChannelBinder extends
rocketMQTemplate = new RocketMQTemplate();
rocketMQTemplate.setObjectMapper(getApplicationContext().getParent()
.getBeansOfType(ObjectMapper.class).values().iterator().next());
DefaultMQProducer producer = new DefaultMQProducer(destination.getName());
producer.setNamesrvAddr(
rocketBinderConfigurationProperties.getNamesrvAddr());
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(
producerProperties.getExtension().getGroup(), rpcHook,
mergedProperties.isEnableMsgTrace(),
mergedProperties.getCustomizedTraceTopic());
producer.setVipChannelEnabled(false);
producer.setInstanceName(
RocketMQUtil.getInstanceName(rpcHook, destination.getName()));
}
else {
producer = new DefaultMQProducer(
producerProperties.getExtension().getGroup());
producer.setVipChannelEnabled(
producerProperties.getExtension().getVipChannelEnabled());
}
producer.setNamesrvAddr(mergedProperties.getNameServer());
producer.setSendMsgTimeout(
producerProperties.getExtension().getSendMessageTimeout());
producer.setRetryTimesWhenSendFailed(
@ -110,14 +141,13 @@ public class RocketMQMessageChannelBinder extends
producerProperties.getExtension().isRetryNextServer());
producer.setMaxMessageSize(
producerProperties.getExtension().getMaxMessageSize());
producer.setVipChannelEnabled(
producerProperties.getExtension().getVipChannelEnabled());
rocketMQTemplate.setProducer(producer);
clientConfigId.add(producer.buildMQClientId());
}
RocketMQMessageHandler messageHandler = new RocketMQMessageHandler(
rocketMQTemplate, destination.getName(),
producerProperties.getExtension().getGroup(),
producerProperties.getExtension().getTransactional(),
instrumentationManager);
messageHandler.setBeanFactory(this.getApplicationContext().getBeanFactory());
@ -145,7 +175,7 @@ public class RocketMQMessageChannelBinder extends
}
RocketMQListenerBindingContainer listenerContainer = new RocketMQListenerBindingContainer(
consumerProperties, this);
consumerProperties, rocketBinderConfigurationProperties, this);
listenerContainer.setConsumerGroup(group);
listenerContainer.setTopic(destination.getName());
listenerContainer.setConsumeThreadMax(consumerProperties.getConcurrency());
@ -154,7 +184,7 @@ public class RocketMQMessageChannelBinder extends
listenerContainer.setDelayLevelWhenNextConsume(
consumerProperties.getExtension().getDelayLevelWhenNextConsume());
listenerContainer
.setNameServer(rocketBinderConfigurationProperties.getNamesrvAddr());
.setNameServer(rocketBinderConfigurationProperties.getNameServer());
RocketMQInboundChannelAdapter rocketInboundChannelAdapter = new RocketMQInboundChannelAdapter(
listenerContainer, consumerProperties, instrumentationManager);

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

@ -16,6 +16,8 @@
package org.springframework.cloud.stream.binder.rocketmq.config;
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;
@ -32,7 +34,8 @@ import org.springframework.context.annotation.Import;
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@Configuration
@Import(RocketMQBinderHealthIndicatorAutoConfiguration.class)
@Import({ RocketMQAutoConfiguration.class,
RocketMQBinderHealthIndicatorAutoConfiguration.class })
@EnableConfigurationProperties({ RocketMQBinderConfigurationProperties.class,
RocketMQExtendedBindingProperties.class })
public class RocketMQBinderAutoConfiguration {
@ -41,6 +44,9 @@ public class RocketMQBinderAutoConfiguration {
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
@Autowired(required = false)
private RocketMQProperties rocketMQProperties = new RocketMQProperties();
@Autowired
public RocketMQBinderAutoConfiguration(
RocketMQExtendedBindingProperties extendedBindingProperties,
@ -60,7 +66,8 @@ public class RocketMQBinderAutoConfiguration {
InstrumentationManager instrumentationManager) {
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
extendedBindingProperties, provisioningProvider,
rocketBinderConfigurationProperties, instrumentationManager);
rocketBinderConfigurationProperties, rocketMQProperties,
instrumentationManager);
return binder;
}

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

@ -16,11 +16,10 @@
package org.springframework.cloud.stream.binder.rocketmq.config;
import org.apache.commons.lang3.StringUtils;
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.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties.Producer;
import org.apache.rocketmq.spring.config.RocketMQConfigUtils;
import org.apache.rocketmq.spring.config.RocketMQTransactionAnnotationProcessor;
import org.apache.rocketmq.spring.config.TransactionHandlerRegistry;
@ -32,7 +31,6 @@ 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 org.springframework.util.Assert;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -51,48 +49,35 @@ public class RocketMQComponent4BinderAutoConfiguration {
}
@Bean
@ConditionalOnMissingBean(DefaultMQProducer.class)
public DefaultMQProducer defaultMQProducer() {
RocketMQProperties rocketMQProperties = new RocketMQProperties();
String configNameServer = environment
.getProperty("spring.cloud.stream.rocketmq.binder.namesrv-addr");
if (StringUtils.isEmpty(configNameServer)) {
rocketMQProperties.setNameServer(RocketMQBinderConstants.DEFAULT_NAME_SERVER);
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 {
rocketMQProperties.setNameServer(configNameServer);
producer = new DefaultMQProducer(RocketMQBinderConstants.DEFAULT_GROUP);
}
RocketMQProperties.Producer producerConfig = new Producer();
rocketMQProperties.setProducer(producerConfig);
producerConfig.setGroup(RocketMQBinderConstants.DEFAULT_GROUP);
String nameServer = rocketMQProperties.getNameServer();
String groupName = producerConfig.getGroup();
Assert.hasText(nameServer, "[rocketmq.name-server] must not be null");
Assert.hasText(groupName, "[rocketmq.producer.group] must not be null");
DefaultMQProducer producer = new DefaultMQProducer(groupName);
producer.setNamesrvAddr(nameServer);
producer.setSendMsgTimeout(producerConfig.getSendMessageTimeout());
producer.setRetryTimesWhenSendFailed(
producerConfig.getRetryTimesWhenSendFailed());
producer.setRetryTimesWhenSendAsyncFailed(
producerConfig.getRetryTimesWhenSendAsyncFailed());
producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
producer.setCompressMsgBodyOverHowmuch(
producerConfig.getCompressMessageBodyThreshold());
producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryNextServer());
producer.setNamesrvAddr(configNameServer);
return producer;
}
@Bean(destroyMethod = "destroy")
@ConditionalOnBean(DefaultMQProducer.class)
@ConditionalOnMissingBean(RocketMQTemplate.class)
@ConditionalOnMissingBean
public RocketMQTemplate rocketMQTemplate(DefaultMQProducer mqProducer,
ObjectMapper rocketMQMessageObjectMapper) {
ObjectMapper objectMapper) {
RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
rocketMQTemplate.setProducer(mqProducer);
rocketMQTemplate.setObjectMapper(rocketMQMessageObjectMapper);
rocketMQTemplate.setObjectMapper(objectMapper);
return rocketMQTemplate;
}

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

@ -18,6 +18,8 @@ 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;
@ -26,8 +28,10 @@ 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;
@ -40,6 +44,7 @@ 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;
@ -83,6 +88,7 @@ public class RocketMQListenerBindingContainer
private final ExtendedConsumerProperties<RocketMQConsumerProperties> rocketMQConsumerProperties;
private final RocketMQMessageChannelBinder rocketMQMessageChannelBinder;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
// The following properties came from RocketMQConsumerProperties.
private ConsumeMode consumeMode;
@ -92,8 +98,10 @@ public class RocketMQListenerBindingContainer
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
@ -189,7 +197,23 @@ public class RocketMQListenerBindingContainer
Assert.notNull(nameServer, "Property 'nameServer' is required");
Assert.notNull(topic, "Property 'topic' is required");
consumer = new DefaultMQPushConsumer(consumerGroup);
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());
@ -238,6 +262,15 @@ public class RocketMQListenerBindingContainer
}
@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;
}

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

@ -40,7 +40,7 @@ import org.springframework.util.Assert;
*/
public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
private static final Logger logger = LoggerFactory
private static final Logger log = LoggerFactory
.getLogger(RocketMQInboundChannelAdapter.class);
private RetryTemplate retryTemplate;
@ -88,7 +88,7 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
}
catch (Exception e) {
logger.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
log.error("rocketMQListenerContainer init error: " + e.getMessage(), e);
throw new IllegalArgumentException(
"rocketMQListenerContainer init error: " + e.getMessage(), e);
}
@ -116,7 +116,7 @@ public class RocketMQInboundChannelAdapter extends MessageProducerSupport {
.getHealthInstrumentation(rocketMQListenerContainer.getTopic()
+ rocketMQListenerContainer.getConsumerGroup())
.markStartFailed(e);
logger.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
throw new MessagingException(MessageBuilder.withPayload(
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
.build(), e);

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

@ -23,6 +23,8 @@ import org.apache.rocketmq.client.producer.SendStatus;
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.metrics.Instrumentation;
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
@ -42,6 +44,9 @@ import org.springframework.util.StringUtils;
*/
public class RocketMQMessageHandler extends AbstractMessageHandler implements Lifecycle {
private final static Logger log = LoggerFactory
.getLogger(RocketMQMessageHandler.class);
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
private MessageChannel sendFailureChannel;
@ -52,6 +57,8 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
private final String destination;
private final String groupName;
private final InstrumentationManager instrumentationManager;
private boolean sync = false;
@ -59,9 +66,11 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
private volatile boolean running = false;
public RocketMQMessageHandler(RocketMQTemplate rocketMQTemplate, String destination,
Boolean transactional, InstrumentationManager instrumentationManager) {
String group, Boolean transactional,
InstrumentationManager instrumentationManager) {
this.rocketMQTemplate = rocketMQTemplate;
this.destination = destination;
this.groupName = group;
this.transactional = transactional;
this.instrumentationManager = instrumentationManager;
}
@ -79,8 +88,7 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
catch (Exception e) {
instrumentationManager.getHealthInstrumentation(destination)
.markStartFailed(e);
logger.error(
"RocketMQTemplate startup failed, Caused by " + e.getMessage());
log.error("RocketMQTemplate startup failed, Caused by " + e.getMessage());
throw new MessagingException(MessageBuilder.withPayload(
"RocketMQTemplate startup failed, Caused by " + e.getMessage())
.build(), e);
@ -106,19 +114,20 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
protected void handleMessageInternal(
final org.springframework.messaging.Message<?> message) throws Exception {
try {
StringBuilder topicWithTags = new StringBuilder(destination);
final StringBuilder topicWithTags = new StringBuilder(destination);
String tags = null;
if (message.getHeaders().get(RocketMQHeaders.TAGS) != null) {
tags = message.getHeaders().get(RocketMQHeaders.TAGS).toString();
}
if (!StringUtils.isEmpty(tags)) {
topicWithTags = topicWithTags.append(":").append(tags);
topicWithTags.append(":").append(tags);
}
SendResult sendRes = null;
if (transactional) {
sendRes = rocketMQTemplate.sendMessageInTransaction(destination,
sendRes = rocketMQTemplate.sendMessageInTransaction(groupName,
topicWithTags.toString(), message, message.getHeaders()
.get(RocketMQBinderConstants.ROCKET_TRANSACTIONAL_ARG));
log.debug("transactional send to topic " + topicWithTags + " " + sendRes);
}
else {
int delayLevel = 0;
@ -139,17 +148,22 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
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);
}
@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
@ -173,8 +187,7 @@ public class RocketMQMessageHandler extends AbstractMessageHandler implements Li
}
}
catch (Exception e) {
logger.error(
"RocketMQ Message hasn't been sent. Caused by " + e.getMessage());
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));

66
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.common.MixAll;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
@ -26,24 +27,69 @@ import org.springframework.cloud.stream.binder.rocketmq.RocketMQBinderConstants;
@ConfigurationProperties(prefix = "spring.cloud.stream.rocketmq.binder")
public class RocketMQBinderConfigurationProperties {
private String namesrvAddr = RocketMQBinderConstants.DEFAULT_NAME_SERVER;
/**
* 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;
/**
* Switch flag instance for message trace.
*/
private boolean enableMsgTrace = true;
/**
* 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 String getNameServer() {
return nameServer;
}
public void setNamesrvAddr(String namesrvAddr) {
this.namesrvAddr = namesrvAddr;
public void setNameServer(String nameServer) {
this.nameServer = nameServer;
}
public String getLogLevel() {
return logLevel;
public String getAccessKey() {
return accessKey;
}
public void setLogLevel(String logLevel) {
this.logLevel = logLevel;
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;
}
}

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

@ -27,7 +27,12 @@ public class RocketMQProducerProperties {
private Boolean enabled = true;
/**
* Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}
* Name of producer.
*/
private String group;
/**
* Maximum allowed message size in bytes {@link DefaultMQProducer#maxMessageSize}.
*/
private Integer maxMessageSize = 1024 * 1024 * 4;
@ -70,6 +75,14 @@ public class RocketMQProducerProperties {
*/
private boolean retryNextServer = false;
public String getGroup() {
return group;
}
public void setGroup(String group) {
this.group = group;
}
public Boolean getEnabled() {
return enabled;
}
@ -150,4 +163,4 @@ public class RocketMQProducerProperties {
this.retryNextServer = retryNextServer;
}
}
}

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

@ -56,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);