分布式应用服务开发的一站式解决方案 Spring Cloud Alibaba

# RocketMQ Example ## Project Instruction This example illustrates how to use RocketMQ Binder implement pub/sub messages for Spring Cloud applications. [RocketMQ](https://rocketmq.apache.org/) is a distributed messaging and streaming platform with low latency, high performance and reliability, trillion-level capacity and flexible scalability. Before we start the demo, let's look at Spring Cloud Stream. Spring Cloud Stream is a framework for building message-driven microservice applications. Spring Cloud Stream builds upon Spring Boot to create standalone, production-grade Spring applications and uses Spring Integration to provide connectivity to message brokers. It provides opinionated configuration of middleware from several vendors, introducing the concepts of persistent publish-subscribe semantics, consumer groups, and partitions. There are two concepts in Spring Cloud Stream: Binder 和 Binding. * Binder: A strategy interface used to bind an app interface to a logical name. Binder Implementations includes `KafkaMessageChannelBinder` of kafka, `RabbitMessageChannelBinder` of RabbitMQ and `RocketMQMessageChannelBinder` of `RocketMQ`. * Binding: Including Input Binding and Output Binding. Binding is Bridge between the external messaging systems and application provided Producers and Consumers of messages. This is a overview of Spring Cloud Stream.  ## Preparation ### Download and Startup RocketMQ You should startup Name Server and Broker before using RocketMQ Binder. 1. Download [RocketMQ](https://archive.apache.org/dist/rocketmq/4.3.2/rocketmq-all-4.3.2-bin-release.zip) and unzip it. 2. Startup Name Server ```bash sh bin/mqnamesrv ``` 3. Startup Broker ```bash sh bin/mqbroker -n localhost:9876 ``` ### Declare dependency Add dependency spring-cloud-starter-stream-rocketmq to the `pom.xml` file in your Spring Cloud project. ```xml <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-stream-rocketmq</artifactId> </dependency> ``` ## Simple example ### Create topic ```sh sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t test-topic ``` ### Integration with RocketMQ Binder Configure Input and Output Binding and cooperate with `@EnableBinding` annotation ```java @SpringBootApplication @EnableBinding({ Source.class, Sink.class }) public class RocketMQApplication { public static void main(String[] args) { SpringApplication.run(RocketMQApplication.class, args); } } ``` Configure Binding: ```properties # configure the nameserver of rocketmq spring.cloud.stream.rocketmq.binder.name-server=127.0.0.1:9876 # configure the output binding named output spring.cloud.stream.bindings.output.destination=test-topic spring.cloud.stream.bindings.output.content-type=application/json # configure the input binding named input spring.cloud.stream.bindings.input.destination=test-topic spring.cloud.stream.bindings.input.content-type=application/json spring.cloud.stream.bindings.input.group=test-group ``` ### Start Application 1. Add necessary configurations to file `/src/main/resources/application.properties`. ```properties spring.application.name=rocketmq-example server.port=28081 ``` 2. Start the application in IDE or by building a fatjar. 1. Start in IDE: Find main class `RocketMQApplication`, and execute the main method. 2. Build a fatjar: Execute command `mvn clean package` to build a fatjar, and run command `java -jar rocketmq-example.jar` to start the application. ### Message Handle Using the binding named output and sent messages to `test-topic` topic. And using two input bindings to subscribe messages. * input1: subscribe the message of `test-topic` topic and consume ordered messages(all messages should in the same MessageQueue if you want to consuming ordered messages). * input2: subscribe the message of `test-topic` topic and consume concurrent messages which tags is `tagStr`, the thread number in pool is 20 in Consumer side. see the configuration below: ```properties spring.cloud.stream.rocketmq.binder.name-server=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=text/plain spring.cloud.stream.bindings.input1.group=test-group1 spring.cloud.stream.rocketmq.bindings.input1.consumer.orderly=true 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=20 ``` #### Pub Messages Using MessageChannel to send messages: ```java public class ProducerRunner implements CommandLineRunner { @Autowired private MessageChannel output; @Override public void run(String... args) throws Exception { Map<String, Object> headers = new HashMap<>(); headers.put(MessageConst.PROPERTY_TAGS, "tagStr"); Message message = MessageBuilder.createMessage(msg, new MessageHeaders(headers)); output.send(message); } } ``` Or you can using the native API of RocketMQ to send messages: ```java public class RocketMQProducer { DefaultMQProducer producer = new DefaultMQProducer("producer_group"); producer.setNamesrvAddr("127.0.0.1:9876"); producer.start(); Message msg = new Message("test-topic", "tagStr", "message from rocketmq producer".getBytes()); producer.send(msg); } ``` #### Sub Messages Using `@StreamListener` to receive messages: ```java @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); } } ``` ## Broadcasting exmaple ### Create topic ```sh sh bin/mqadmin updateTopic -n localhost:9876 -c DefaultCluster -t broadcast ``` ### Producer **application.yml** ```yaml server: port: 28085 spring: application: name: rocketmq-broadcast-producer-example cloud: stream: rocketmq: binder: name-server: localhost:9876 bindings: producer-out-0: producer: group: output_1 bindings: producer-out-0: destination: broadcast logging: level: org.springframework.context.support: debug ``` **code** Use `ApplicationRunner` and `StreamBridge` to send messages. ```java @SpringBootApplication public class RocketMQBroadcastProducerApplication { private static final Logger log = LoggerFactory .getLogger(RocketMQBroadcastProducerApplication.class); @Autowired private StreamBridge streamBridge; public static void main(String[] args) { SpringApplication.run(RocketMQBroadcastProducerApplication.class, args); } @Bean public ApplicationRunner producer() { return args -> { for (int i = 0; i < 100; i++) { String key = "KEY" + i; Map<String, Object> headers = new HashMap<>(); headers.put(MessageConst.PROPERTY_KEYS, key); headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i); Message<SimpleMsg> msg = new GenericMessage<SimpleMsg>(new SimpleMsg("Hello RocketMQ " + i), headers); streamBridge.send("producer-out-0", msg); } }; } } ``` ### Consumer Startup two consumers. #### Consumer1 **application.yml** ```yaml server: port: 28084 spring: application: name: rocketmq-broadcast-consumer1-example cloud: stream: function: