How to configure message-driven Spring Cloud Stream
This article mainly explains "how to configure message-driven Spring Cloud Stream". Interested friends may wish to take a look. The method introduced in this paper is simple, fast and practical. Let's let the editor take you to learn how to configure message-driven Spring Cloud Stream.
When using the spring cloud cloud architecture, we have to use Spring cloud Stream, because the use of message middleware is ubiquitous in the project. Our company has done APP for entertainment. When using spring cloud as the architecture, the asynchronous notification of messages and the asynchronous processing of business all need to use message middleware mechanism. According to the official integration recommendation of spring cloud (using rabbit mq and kafka), I took a look at the source code and configuration. As long as rabbit mq is integrated, kafka is just changed to a pom configuration jar package. To cut the slack, let's go straight to the configuration implementation:
1. Brief introduction:
Spring cloud Stream data flow operation development kit, encapsulates sending and receiving messages with Redis,Rabbit, Kafka and so on.
two。 Use the tool:
Rabbit, I won't explain too much about download and installation details here. There are too many examples on the Internet.
3. Create a sender project for commonservice-mq-producer messages and configure stream-rabbit dependencies in pom
Org.springframework.cloud spring-cloud-starter-stream-rabbit
4. Configure rabbit mq in the yml file
Server: port: 5666 spring: application: name: commonservice-mq-producer profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server # rabbitmq and kafka both have default values for related configurations. If you modify the You can configure stream: bindings: mqScoreOutput: destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghu eureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true
5. Define interface ProducerService
Package com.honghu.cloud.producer; import org.springframework.cloud.stream.annotation.Output; import org.springframework.messaging.SubscribableChannel; public interface ProducerService {String SCORE_OUPUT = "mqScoreOutput"; @ Output (ProducerService.SCORE_OUPUT) SubscribableChannel sendMessage ();}
6. Define binding
Package com.honghu.cloud.producer; import org.springframework.cloud.stream.annotation.EnableBinding; @ EnableBinding (ProducerService.class) public class SendServerConfig {}
7. Define ProducerController for sending message service
Package com.honghu.cloud.controller; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RestController; import com.honghu.cloud.common.code.ResponseCode Import com.honghu.cloud.common.code.ResponseVO; import com.honghu.cloud.entity.User; import com.honghu.cloud.producer.ProducerService; import net.sf.json.JSONObject; @ RestController @ RequestMapping (value = "producer") public class ProducerController {@ Autowired private ProducerService producerService / * * send object by get * @ param name path parameter * @ return succeeded | failed * / @ RequestMapping (value = "/ sendObj", method = RequestMethod.GET) public ResponseVO sendObj () {User user = new User (1, "hello User"); Message msg = MessageBuilder.withPayload (user). Build () Boolean result = producerService.sendMessage () .send (msg); if (result) {return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_SUCCESS, false);} return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_FAILURE, false) } / * send string message via get * @ param name path parameter * @ return succeeded | failed * / @ RequestMapping (value = "/ send/ {name}", method = RequestMethod.GET) public ResponseVO send (@ PathVariable (value = "name") Required = true) String name) {Message msg = MessageBuilder.withPayload (name.getBytes ()) .build () Boolean result = producerService.sendMessage () .send (msg); if (result) {return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_SUCCESS, false);} return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_FAILURE, false) } / * * send json object * @ param name path parameter * @ return successfully | failed * / @ RequestMapping (value = "/ sendJsonObj", method = RequestMethod.POST) public ResponseVO sendJsonObj (@ RequestBody JSONObject jsonObj) {Message msg = MessageBuilder.withPayload (jsonObj). Build (); boolean result = producerService.sendMessage (). Send (msg) If (result) {return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_SUCCESS, false);} return ResponseCode.buildEnumResponseVO (ResponseCode.RESPONSE_CODE_FAILURE, false);}}
8. Create a consumer project for commonservice-mq-consumer1 messages and configure stream-rabbit dependencies in pom
Org.springframework.cloud spring-cloud-starter-stream-rabbit
9. Configure in the yml file:
Server: port: 5111 spring: application: name: commonservice-mq-consumer1 profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server stream: bindings: mqScoreInput: group: honghu_queue destination: honghu_exchange contentType: application/json Rabbitmq: host: localhost port: 5672 username: honghu password: honghu eureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true
10. Define interface ConsumerService
Package com.honghu.cloud.consumer; import org.springframework.cloud.stream.annotation.Input; import org.springframework.messaging.SubscribableChannel; public interface ConsumerService {String SCORE_INPUT = "mqScoreInput"; @ Input (ConsumerService.SCORE_INPUT) SubscribableChannel sendMessage ();}
11. Define startup classes and message consumption
Package com.honghu.cloud; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.netflix.eureka.EnableEurekaClient; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; import com.honghu.cloud.consumer.ConsumerService; import com.honghu.cloud.entity.User @ EnableEurekaClient @ SpringBootApplication @ EnableBinding (ConsumerService.class) / / you can bind multiple interfaces public class ConsumerApplication {public static void main (String [] args) {SpringApplication.run (ConsumerApplication.class, args);} @ StreamListener (ConsumerService.SCORE_INPUT) public void onMessage (Object obj) {System.out.println ("Consumer 1, received message:" + obj);}}
twelve。 Start commonservice-mq-producer and commonservice-mq-consumer1 respectively
13. Verify the sending and receiving of messages through postman
At this point, I believe you have a deeper understanding of "message-driven Spring Cloud Stream how to configure", might as well come to the actual operation! Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!
