RabbitMQ ③-Spring使用RabbitMQ

Spring使用RabbitMQ

创建 Spring 项目后，引入依赖：

1. <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
2. <dependency>
3.     <groupId>org.springframework.boot</groupId>
4.     <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

复制代码

配置文件 application.yml：

1. spring:
2.   application:
3.     name: spring-rabbitmq-demo
4.   rabbitmq:
5. #    host: 47.94.9.33
6. #    port: 5672
7. #    username: admin
8. #    password: admin
9. #    virtual-host: /
10.     addresses: amqp://admin:admin@47.94.9.33:5672/

复制代码

Work-Queue（工作队列模式）

声明队列

1. package com.ljh.mq.springrabbitmqdemo.config;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.springframework.amqp.core.*;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. @Configuration
10. public class RabbitMQConfig {
11.     // * 工作队列模式
12.     @Bean
13.     public Queue workQueue() {
14.         return QueueBuilder.durable(Constants.WORK_QUEUE)
15.                 .build();
16.     }
17. }

复制代码

生产者

1. package com.ljh.mq.springrabbitmqdemo.controller;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.web.bind.annotation.PathVariable;
9. import org.springframework.web.bind.annotation.RequestMapping;
10. import org.springframework.web.bind.annotation.RestController;
12. @RequestMapping("/producer")
13. @RestController
14. public class ProducerController {
15.     private static final Logger log = LoggerFactory.getLogger(ProducerController.class);
17.     @Autowired
18.     RabbitTemplate rabbitTemplate;
20.     @RequestMapping("/work")
21.     public String work() {
22.         for (int i = 0; i < 10; i++) {
23.             String msg = "hello work queue mode~ " + i;
24.             // ? 当使用默认交换机时，routingKey 和队列名称保持一致
25.             rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, msg);
26.         }
27.         log.info("消息发送成功");
28.         return "消息发送成功";
29.     }
30. }

复制代码

消耗者

1. package com.ljh.mq.springrabbitmqdemo.listener;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import com.rabbitmq.client.Channel;
5. import org.slf4j.Logger;
6. import org.slf4j.LoggerFactory;
7. import org.springframework.amqp.core.Message;
8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
9. import org.springframework.stereotype.Component;
11. @Component
12. public class WorkListener {
14.     private static final Logger log = LoggerFactory.getLogger(WorkListener.class);
16.     @RabbitListener(queues = Constants.WORK_QUEUE)
17.     public void process1(Message message, Channel channel) {
18.         log.info("[process1]:成功接收到消息:[{}]:{}", Constants.WORK_QUEUE, message);
19.     }
21.     @RabbitListener(queues = Constants.WORK_QUEUE)
22.     public void process2(String message) {
23.         log.info("[process2]:成功接收到消息:[{}]:{}", Constants.WORK_QUEUE, message);
24.     }
25. }

复制代码

Publish/Subscribe（发布/订阅模式）

声明队列和交换机

1. package com.ljh.mq.springrabbitmqdemo.config;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.springframework.amqp.core.*;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. @Configuration
10. public class RabbitMQConfig {
11.     // * 发布订阅模式
12.     @Bean("fanoutExchange")
13.     public FanoutExchange fanoutExchange() {
14.         return ExchangeBuilder.fanoutExchange(Constants.FANOUT_EXCHANGE)
15.                 .durable(true)
16.                 .build();
17.     }
18.     @Bean("fanoutQueue1")
19.     public Queue fanoutQueue1 () {
20.         return QueueBuilder.durable(Constants.FANOUT_QUEUE1)
21.                 .build();
22.     }
23.     @Bean("fanoutQueue2")
24.     public Queue fanoutQueue2 () {
25.         return QueueBuilder.durable(Constants.FANOUT_QUEUE2)
26.                 .build();
27.     }
28.     @Bean("bindingFanout1")
29.     public Binding bindingFanout1(@Qualifier("fanoutExchange") FanoutExchange exchange, @Qualifier("fanoutQueue1") Queue queue) {
30.         return BindingBuilder.bind(queue)
31.                 .to(exchange);
32.     }
33.     @Bean("bindingFanout2")
34.     public Binding bindingFanout2(@Qualifier("fanoutExchange") FanoutExchange exchange, @Qualifier("fanoutQueue2") Queue queue) {
35.         return BindingBuilder.bind(queue)
36.                 .to(exchange);
37.     }
38. }

复制代码

生产者

1. package com.ljh.mq.springrabbitmqdemo.controller;
2. package com.ljh.mq.springrabbitmqdemo.controller;
4. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
5. import org.slf4j.Logger;
6. import org.slf4j.LoggerFactory;
7. import org.springframework.amqp.rabbit.core.RabbitTemplate;
8. import org.springframework.beans.factory.annotation.Autowired;
9. import org.springframework.web.bind.annotation.PathVariable;
10. import org.springframework.web.bind.annotation.RequestMapping;
11. import org.springframework.web.bind.annotation.RestController;
13. @RequestMapping("/producer")
14. @RestController
15. public class ProducerController {
16.     private static final Logger log = LoggerFactory.getLogger(ProducerController.class);
18.     @Autowired
19.     RabbitTemplate rabbitTemplate;
21.     @RequestMapping("/fanout")
22.     public String fanout() {
23.         for (int i = 0; i < 10; i++) {
24.             String msg = "hello publish fanout mode~ " + i;
25.             // ? 当使用默认交换机时，routingKey 和队列名称保持一致
26.             rabbitTemplate.convertAndSend(Constants.FANOUT_EXCHANGE, "", msg);
27.         }
28.         log.info("消息发送成功");
29.         return "消息发送成功";
30.     }
31. }

复制代码

消耗者

1. package com.ljh.mq.springrabbitmqdemo.listener;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import com.rabbitmq.client.Channel;
5. import org.slf4j.Logger;
6. import org.slf4j.LoggerFactory;
7. import org.springframework.amqp.core.Message;
8. import org.springframework.amqp.rabbit.annotation.RabbitListener;
9. import org.springframework.stereotype.Component;
11. @Component
12. public class FanoutListener {
14.     private static final Logger log = LoggerFactory.getLogger(FanoutListener.class);
16.     @RabbitListener(queues = Constants.FANOUT_QUEUE1)
17.     public void process1(String message) {
18.         log.info("[process1]:成功接收到消息:[{}]:{}", Constants.FANOUT_QUEUE1, message);
19.     }
21.     @RabbitListener(queues = Constants.FANOUT_QUEUE2)
22.     public void process2(String message) {
23.         log.info("[process2]:成功接收到消息:[{}]:{}", Constants.FANOUT_QUEUE2, message);
24.     }
25. }

复制代码

Routing（路由模式）

声明队列和交换机

1. package com.ljh.mq.springrabbitmqdemo.config;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.springframework.amqp.core.*;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. @Configuration
10. public class RabbitMQConfig {
11.     // * 路由模式
12.     @Bean("directExchange")
13.     public DirectExchange directExchange() {
14.         return ExchangeBuilder.directExchange(Constants.DIRECT_EXCHANGE)
15.                 .durable(true)
16.                 .build();
17.     }
18.     @Bean("directQueue1")
19.     public Queue directQueue1() {
20.         return QueueBuilder.durable(Constants.DIRECT_QUEUE1)
21.                 .build();
22.     }
23.     @Bean("directQueue2")
24.     public Queue directQueue2() {
25.         return QueueBuilder.durable(Constants.DIRECT_QUEUE2)
26.                 .build();
27.     }
28.     @Bean("bindingDirect1")
29.     public Binding bindingDirect1(@Qualifier("directExchange") DirectExchange exchange, @Qualifier("directQueue1") Queue queue) {
30.         return BindingBuilder.bind(queue)
31.                 .to(exchange)
32.                 .with("orange");
33.     }
34.     @Bean("bindingDirect2")
35.     public Binding bindingDirect2(@Qualifier("directExchange") DirectExchange exchange, @Qualifier("directQueue2") Queue queue) {
36.         return BindingBuilder.bind(queue)
37.                 .to(exchange)
38.                 .with("orange");
39.     }
40.     @Bean("bindingDirect3")
41.     public Binding bindingDirect3(@Qualifier("directExchange") DirectExchange exchange, @Qualifier("directQueue2") Queue queue) {
42.         return BindingBuilder.bind(queue)
43.                 .to(exchange)
44.                 .with("black");
45.     }
46. }

复制代码

生产者

1. package com.ljh.mq.springrabbitmqdemo.controller;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.web.bind.annotation.PathVariable;
9. import org.springframework.web.bind.annotation.RequestMapping;
10. import org.springframework.web.bind.annotation.RestController;
12. @RequestMapping("/producer")
13. @RestController
14. public class ProducerController {
15.     private static final Logger log = LoggerFactory.getLogger(ProducerController.class);
17.     @Autowired
18.     RabbitTemplate rabbitTemplate;
20.     @RequestMapping("/direct/{routingKey}")
21.     public String direct(@PathVariable("routingKey") String routingKey) {
22.         rabbitTemplate.convertAndSend(Constants.DIRECT_EXCHANGE, routingKey, "hello routing mode~；routingKey is " + routingKey);
23.         log.info("消息发送成功：{}", routingKey);
24.         return "消息发送成功：" + routingKey;
25.     }
26. }

复制代码

消耗者

1. package com.ljh.mq.springrabbitmqdemo.listener;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
7. import org.springframework.stereotype.Component;
9. @Component
10. public class DirectListener {
11.     private static final Logger log = LoggerFactory.getLogger(DirectListener.class);
13.     @RabbitListener(queues = Constants.DIRECT_QUEUE1)
14.     public void process1(String message) {
15.         log.info("队列[{}]成功接收到消息：{}", Constants.DIRECT_QUEUE1, message);
16.     }
18.     @RabbitListener(queues = Constants.DIRECT_QUEUE2)
19.     public void process2(String message) {
20.         log.info("队列[{}]成功接收到消息：{}", Constants.DIRECT_QUEUE2, message);
21.     }
22. }

复制代码

Topics（通配符模式）

声明队列和交换机

1. package com.ljh.mq.springrabbitmqdemo.config;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.springframework.amqp.core.*;
5. import org.springframework.beans.factory.annotation.Qualifier;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
9. @Configuration
10. public class RabbitMQConfig {
11.     // * 通配符模式
12.     @Bean("topicExchange")
13.     public TopicExchange topicExchange() {
14.         return ExchangeBuilder.topicExchange(Constants.TOPIC_EXCHANGE)
15.                 .durable(true)
16.                 .build();
17.     }
18.     @Bean("topicQueue1")
19.     public Queue topicQueue1() {
20.         return QueueBuilder.durable(Constants.TOPIC_QUEUE1)
21.                 .build();
22.     }
23.     @Bean("topicQueue2")
24.     public Queue topicQueue2() {
25.         return QueueBuilder.durable(Constants.TOPIC_QUEUE2)
26.                 .build();
27.     }
28.     @Bean("bindingTopic1")
29.     public Binding bindingTopic1(@Qualifier("topicExchange") TopicExchange exchange, @Qualifier("topicQueue1") Queue queue) {
30.         return BindingBuilder.bind(queue)
31.                 .to(exchange)
32.                 .with("*.orange.*");
33.     }
34.     @Bean("bindingTopic2")
35.     public Binding bindingTopic2(@Qualifier("topicExchange") TopicExchange exchange, @Qualifier("topicQueue2") Queue queue) {
36.         return BindingBuilder.bind(queue)
37.                 .to(exchange)
38.                 .with("*.*.rabbit");
39.     }
40.     @Bean("bindingTopic3")
41.     public Binding bindingTopic3(@Qualifier("topicExchange") TopicExchange exchange, @Qualifier("topicQueue2") Queue queue) {
42.         return BindingBuilder.bind(queue)
43.                 .to(exchange)
44.                 .with("lazy.#");
45.     }
46. }

复制代码

生产者

1. package com.ljh.mq.springrabbitmqdemo.controller;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.web.bind.annotation.PathVariable;
9. import org.springframework.web.bind.annotation.RequestMapping;
10. import org.springframework.web.bind.annotation.RestController;
12. @RequestMapping("/producer")
13. @RestController
14. public class ProducerController {
15.     private static final Logger log = LoggerFactory.getLogger(ProducerController.class);
17.     @Autowired
18.     RabbitTemplate rabbitTemplate;
20.     @RequestMapping("/topic/{routingKey}")
21.     public String topic(@PathVariable("routingKey") String routingKey) {
22.         rabbitTemplate.convertAndSend(Constants.TOPIC_EXCHANGE, routingKey, "hello topic mode~；routingKey is " + routingKey);
23.         log.info("消息发送成功：{}", routingKey);
24.         return "消息发送成功：" + routingKey;
25.     }
26. }

复制代码

消耗者

1. package com.ljh.mq.springrabbitmqdemo.listener;
3. import com.ljh.mq.springrabbitmqdemo.constants.Constants;
4. import org.slf4j.Logger;
5. import org.slf4j.LoggerFactory;
6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
7. import org.springframework.stereotype.Component;
9. @Component
10. public class TopicListener {
11.     private static final Logger log = LoggerFactory.getLogger(TopicListener.class);
13.     @RabbitListener(queues = Constants.TOPIC_QUEUE1)
14.     public void process1(String message) {
15.         log.info("队列[{}]成功接收到消息：{}", Constants.TOPIC_QUEUE1, message);
16.     }
18.     @RabbitListener(queues = Constants.TOPIC_QUEUE2)
19.     public void process2(String message) {
20.         log.info("队列[{}]成功接收到消息：{}", Constants.TOPIC_QUEUE2, message);
21.     }
22. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。