diff --git a/spring-cloud-alibaba-examples/pom.xml b/spring-cloud-alibaba-examples/pom.xml index acdcb604..103e15b2 100644 --- a/spring-cloud-alibaba-examples/pom.xml +++ b/spring-cloud-alibaba-examples/pom.xml @@ -27,6 +27,7 @@ ans-example/ans-consumer-ribbon-example ans-example/ans-provider-example acm-example/acm-local-example + rocketmq-example diff --git a/spring-cloud-alibaba-examples/rocketmq-example/pom.xml b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml new file mode 100644 index 00000000..5c793f8a --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/pom.xml @@ -0,0 +1,52 @@ + + + + + org.springframework.cloud + spring-cloud-alibaba-examples + 0.2.1.BUILD-SNAPSHOT + ../../pom.xml + + 4.0.0 + + + rocketmq-example + jar + Example demonstrating how to use rocketmq + + + + + org.springframework.cloud + spring-cloud-starter-stream-rocketmq + + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.boot + spring-boot-starter-actuator + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-deploy-plugin + ${maven-deploy-plugin.version} + + true + + + + + + diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java new file mode 100644 index 00000000..652400a2 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/Foo.java @@ -0,0 +1,39 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +/** + * @author Jim + */ +public class Foo { + + private int id; + private String tag; + + public Foo() { + } + + public Foo(int id, String tag) { + this.id = id; + this.tag = tag; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public String getTag() { + return tag; + } + + public void setTag(String tag) { + this.tag = tag; + } + + @Override + public String toString() { + return "Foo{" + "id=" + id + ", tag='" + tag + '\'' + '}'; + } +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java new file mode 100644 index 00000000..f251902d --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/ReceiveService.java @@ -0,0 +1,33 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Service; + +/** + * @author Jim + */ +@Service +public class ReceiveService { + + @StreamListener("input1") + public void receiveInput1(String receiveMsg) { + System.out.println("input1 receive: " + receiveMsg); + } + + @StreamListener("input2") + public void receiveInput2(String receiveMsg) { + System.out.println("input2 receive: " + receiveMsg); + } + + @StreamListener("input3") + public void receiveInput3(@Payload Foo foo) { + System.out.println("input3 receive: " + foo); + } + + @StreamListener("input1") + public void receiveInput1Again(String receiveMsg) { + System.out.println("input1 receive again: " + receiveMsg); + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java new file mode 100644 index 00000000..2d4deda6 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/RocketMQApplication.java @@ -0,0 +1,65 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.CommandLineRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.alibaba.cloud.examples.RocketMQApplication.MySink; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.SubscribableChannel; + +/** + * @author Jim + */ +@SpringBootApplication +@EnableBinding({ Source.class, MySink.class }) +public class RocketMQApplication { + + public interface MySink { + + @Input("input1") + SubscribableChannel input1(); + + @Input("input2") + SubscribableChannel input2(); + + @Input("input3") + SubscribableChannel input3(); + + } + + public static void main(String[] args) { + SpringApplication.run(RocketMQApplication.class, args); + } + + @Bean + public CustomRunner customRunner() { + return new CustomRunner(); + } + + public static class CustomRunner implements CommandLineRunner { + @Autowired + private SenderService senderService; + + @Override + public void run(String... args) throws Exception { + int count = 5; + for (int index = 1; index <= count; index++) { + String msgContent = "msg-" + index; + if (index % 3 == 0) { + senderService.send(msgContent); + } + else if (index % 3 == 1) { + senderService.sendWithTags(msgContent, "tagStr"); + } + else { + senderService.sendObject(new Foo(index, "foo"), "tagObj"); + } + } + } + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java new file mode 100644 index 00000000..eaf0001f --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/java/org/springframework/cloud/alibaba/cloud/examples/SenderService.java @@ -0,0 +1,43 @@ +package org.springframework.cloud.alibaba.cloud.examples; + +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.apache.rocketmq.common.message.MessageConst; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.stereotype.Service; +import org.springframework.util.MimeTypeUtils; + +/** + * @author Jim + */ +@Service +public class SenderService { + + @Autowired + private Source source; + + public void send(String msg) throws Exception { + source.output().send(MessageBuilder.withPayload(msg).build()); + } + + public void sendWithTags(T msg, String tag) throws Exception { + Message message = MessageBuilder.createMessage(msg, + new MessageHeaders(Stream.of(tag).collect(Collectors + .toMap(str -> MessageConst.PROPERTY_TAGS, String::toString)))); + source.output().send(message); + } + + public void sendObject(T msg, String tag) throws Exception { + Message message = MessageBuilder.withPayload(msg) + .setHeader(MessageConst.PROPERTY_TAGS, tag) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) + .build(); + source.output().send(message); + } + +} diff --git a/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties new file mode 100644 index 00000000..2a41b0d9 --- /dev/null +++ b/spring-cloud-alibaba-examples/rocketmq-example/src/main/resources/application.properties @@ -0,0 +1,31 @@ +spring.cloud.stream.default-binder=rocketmq + +spring.cloud.stream.rocketmq.binder.namesrv-addr=127.0.0.1:9876 + +spring.cloud.stream.bindings.output.destination=test-topic +spring.cloud.stream.bindings.output.content-type=application/json + +spring.cloud.stream.bindings.input1.destination=test-topic +spring.cloud.stream.bindings.input1.content-type=application/json +spring.cloud.stream.bindings.input1.group=test-group1 +spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true +spring.cloud.stream.bindings.input1.consumer.maxAttempts=1 + +spring.cloud.stream.bindings.input2.destination=test-topic +spring.cloud.stream.bindings.input2.content-type=application/json +spring.cloud.stream.bindings.input2.group=test-group2 +spring.cloud.stream.rocketmq.bindings.input2.consumer.orderly=false +spring.cloud.stream.rocketmq.bindings.input2.consumer.tags=tagStr +spring.cloud.stream.bindings.input2.consumer.concurrency=20 +spring.cloud.stream.bindings.input2.consumer.maxAttempts=1 + +spring.cloud.stream.bindings.input3.destination=test-topic +spring.cloud.stream.bindings.input3.content-type=application/json +spring.cloud.stream.bindings.input3.group=test-group3 +spring.cloud.stream.rocketmq.bindings.input3.consumer.tags=tagObj +spring.cloud.stream.bindings.input3.consumer.concurrency=20 +spring.cloud.stream.bindings.input3.consumer.maxAttempts=1 + +server.port=28081 + +management.endpoints.web.exposure.include=* \ No newline at end of file