SpringBoot整合Rabbitmq

转x33g5p2x  于2022-02-11 转载在 Spring  
字(7.3k)|赞(0)|评价(0)|浏览(512)

1.安装rabbitmq

可参考这篇文章链接: https://www.jianshu.com/p/14ffe0f3db94.

2.SpringBoot整合Rabbitmq案例

2.1 Direct模式（点对点）

1.docker上启动rabbitmq实例
2.pom.xml

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.     <parent>
  6.         <groupId>org.springframework.boot</groupId>
  7.         <artifactId>spring-boot-starter-parent</artifactId>
  8.         <version>2.6.3</version>
  9.         <relativePath/> <!-- lookup parent from repository -->
  10.     </parent>
  11.     <groupId>com.yl</groupId>
  12.     <artifactId>amqp</artifactId>
  13.     <version>0.0.1-SNAPSHOT</version>
  14.     <name>amqp</name>
  15.     <description>Demo project for Spring Boot</description>
  16.     <properties>
  17.         <java.version>11</java.version>
  18.     </properties>
  19.     <dependencies>
  20.         <dependency>
  21.             <groupId>org.springframework.boot</groupId>
  22.             <artifactId>spring-boot-starter-amqp</artifactId>
  23.         </dependency>
  24.         <dependency>
  25.             <groupId>org.springframework.boot</groupId>
  26.             <artifactId>spring-boot-starter-web</artifactId>
  27.         </dependency>
  29.         <dependency>
  30.             <groupId>org.springframework.boot</groupId>
  31.             <artifactId>spring-boot-starter-test</artifactId>
  32.             <scope>test</scope>
  33.         </dependency>
  34.         <dependency>
  35.             <groupId>org.springframework.amqp</groupId>
  36.             <artifactId>spring-rabbit-test</artifactId>
  37.             <scope>test</scope>
  38.         </dependency>
  39.     </dependencies>
  41.     <build>
  42.         <plugins>
  43.             <plugin>
  44.                 <groupId>org.springframework.boot</groupId>
  45.                 <artifactId>spring-boot-maven-plugin</artifactId>
  46.             </plugin>
  47.         </plugins>
  48.     </build>
  50. </project>

3.application.properties

  1. # rabbitmq的配置
  2. # 主机
  3. spring.rabbitmq.host=192.168.244.135
  4. # 端口
  5. spring.rabbitmq.port=5672
  6. # 用户名
  7. spring.rabbitmq.username=guest
  8. # 密码
  9. spring.rabbitmq.password=guest

4.Direct模式配置文件

  1. package com.yl.amqp.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.DirectExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. @Configuration
  11. public class DirectConfig {
  13.     //定义队列
  14.     @Bean
  15.     Queue directQueue() {
  16.         return new Queue("yl-queue",true);
  17.     }
  19.     //定义Direct交换机
  20.     @Bean
  21.     DirectExchange directExchange() {
  22.         return new DirectExchange("yl-direct",true,false);
  23.     }
  25.     //将队列绑定到交换机上
  26.     @Bean
  27.     Binding directBiding() {
  28.         return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");
  29.     }
  30. }

5.监听消息队列

  1. package com.yl.amqp.receicer;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class DirectReceiver {
  9.     // 监听接收消息
  10.     @RabbitListener(queues = "yl-queue")
  11.     public void handler(String msg) {
  12.         System.out.println(msg);
  13.     }
  14. }

6.测试

2.2 Fanout模式（点对面）

1.Fanout模式配置

  1. package com.yl.amqp.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.FanoutExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. @Configuration
  11. public class FanoutConfig {
  12.     @Bean
  13.     Queue queueOne() {
  14.         return new Queue("queue-one",true);
  15.     }
  16.     @Bean
  17.     Queue queueTwo() {
  18.         return new Queue("queue-two",true);
  19.     }
  21.     // 定义fanout交换机
  22.     @Bean
  23.     FanoutExchange fanoutExchange() {
  24.         return new FanoutExchange("yl-fanout",true,false);
  25.     }
  27.     // 将队列绑定到交换机
  28.     @Bean
  29.     Binding bindingOne() {
  30.         return BindingBuilder.bind(queueOne()).to(fanoutExchange());
  31.     }
  33.     // 将队列绑定到交换机
  34.     @Bean
  35.     Binding bindingTwo() {
  36.         return BindingBuilder.bind(queueTwo()).to(fanoutExchange());
  37.     }
  38. }

2.监听消息

  1. package com.yl.amqp.receicer;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class FanoutReceiver {
  8.     @RabbitListener(queues = "queue-one")
  9.     public void handler1(String msg) {
  10.         System.out.println("handle1()"+msg);
  11.     }
  12.     @RabbitListener(queues = "queue-two")
  13.     public void handle2(String msg) {
  14.         System.out.println("handle2()"+msg);
  15.     }
  16. }

3.测试

2.3 Tpoic模式（点对面）

1.Topic模式配置

  1. package com.yl.amqp.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.Queue;
  6. import org.springframework.amqp.core.TopicExchange;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. @Configuration
  11. public class TopicConfig {
  12.     @Bean
  13.     Queue huawei() {
  14.         return new Queue("huawei",true);
  15.     }
  16.     @Bean
  17.     Queue apple() {
  18.         return new Queue("apple",true);
  19.     }
  20.     @Bean
  21.     Queue phone() {
  22.         return new Queue("phone",true);
  23.     }
  25.     @Bean
  26.     TopicExchange topicExchange() {
  27.         return new TopicExchange("yl-topic",true,false);
  28.     }
  30.     @Bean
  31.     Binding huaweiBinding() {
  32.         return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
  33.     }
  34.     @Bean
  35.     Binding appleBinding() {
  36.         return BindingBuilder.bind(apple()).to(topicExchange()).with("apple.#");
  37.     }
  38.     @Bean
  39.     Binding phoneBinding() {
  40.         return BindingBuilder.bind(phone()).to(topicExchange()).with("#.phone.#");
  41.     }
  43. }

2.监听消息

  1. package com.yl.amqp.receicer;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class TopicReceiver {
  8.     @RabbitListener(queues = "huawei")
  9.     public void handle1(String msg) {
  10.         System.out.println("handle1()" + msg);
  11.     }
  12.     @RabbitListener(queues = "apple")
  13.     public void handle2(String msg) {
  14.         System.out.println("handle2()" + msg);
  15.     }
  16.     @RabbitListener(queues = "phone")
  17.     public void handle3(String msg) {
  18.         System.out.println("handle3()" + msg);
  19.     }
  20. }

3.测试1

4.测试2

2.4 Headers模式（很少用到）

1.Headers模式的配置

  1. package com.yl.amqp.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.HeadersExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. import java.util.HashMap;
  11. import java.util.Map;
  13. @Configuration
  14. public class HeaderConfig {
  15.     @Bean
  16.     Queue ageQueue() {
  17.         return new Queue("queue-age",true);
  18.     }
  19.     @Bean
  20.     Queue nameQueue() {
  21.         return new Queue("queue-name",true);
  22.     }
  24.     @Bean
  25.     HeadersExchange headersExchange() {
  26.         return new HeadersExchange("yl-headers",true,false);
  27.     }
  29.     @Bean
  30.     Binding ageBinding() {
  31.         Map<String,Object> map = new HashMap<>();
  32.         map.put("age",18);
  33.         return BindingBuilder.bind(ageQueue()).to(headersExchange()).whereAny(map).match();
  34.     }
  36.     @Bean
  37.     Binding nameBinding() {
  38.         return BindingBuilder.bind(nameQueue()).to(headersExchange()).where("name").exists();
  39.     }
  40. }

2.监听消息

  1. package com.yl.amqp.receicer;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class HeadersReceiver {
  8.     @RabbitListener(queues = "queue-age")
  9.     public void handle1(String msg) {
  10.         System.out.println("handle1(),queue-age:" + msg);
  11.     }
  13.     @RabbitListener(queues = "queue-name")
  14.     public void handle2(String msg) {
  15.         System.out.println("handle2(),queue-name:" + msg);
  16.     }
  17. }

3.测试1

4.测试2

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_41359273/article/details/122870599

内容来源于网络，如有侵权，请联系作者删除！

Javarabbitmq

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com