• Spring Boot集成RabbitMQ快速入门Demo

    1.什么是RabbitMQ？

    RabbitMQ是一款使用Erlang语言开发的，基于AMQP协议的消息中间件，作为一款优秀的消息系统，RabbitMQ有高并发、可扩展等优势，并适用于大型系统中各个模块之间的通信。

    RabbitMQ的特点为：

    • 持久化、传输确认、发布确认等功能保证消息可靠

    • 支持多种消息分发模式，处理更加灵活

    • 提供可视化管理界面，使用方便

    • 支持集群部署，保证服务高可用

    2.RabbitMQ环境搭建

    1. version: '3'
    2. services:
    3.   rabbitmq:
    4.     image: registry.cn-hangzhou.aliyuncs.com/zhengqing/rabbitmq:3.7.8-management        # 原镜像`rabbitmq:3.7.8-management` 【 注：该版本包含了web控制页面 】
    5.     container_name: rabbitmq            # 容器名为'rabbitmq'
    6.     hostname: my-rabbit
    7.     restart: unless-stopped                                       # 指定容器退出后的重启策略为始终重启，但是不考虑在Docker守护进程启动时就已经停止了的容器
    8.     environment:                        # 设置环境变量,相当于docker run命令中的-e
    9.       TZ: Asia/Shanghai
    10.       LANG: en_US.UTF-8
    11.       RABBITMQ_DEFAULT_VHOST: my_vhost  # 主机名
    12.       RABBITMQ_DEFAULT_USER: admin      # 登录账号
    13.       RABBITMQ_DEFAULT_PASS: admin      # 登录密码
    14.     volumes: # 数据卷挂载路径设置,将本机目录映射到容器目录
    15.       - "./rabbitmq/data:/var/lib/rabbitmq"
    16.     ports:                              # 映射端口
    17.       - "5672:5672"
    18.       - "15672:15672"

    运行

    docker-compose -f docker-compose-rabbitmq.yml -p rabbitmq up -d

    web管理端：http://127.0.0.1:15672 登录账号密码：admin/admin

    27a7e34328fda7e39066d77e99a7d4a8.png

    3.代码工程

    实验目的：实现通过rabbitmq发送和接收消息

    pom.xml

    1. "1.0" encoding="UTF-8"?>
    2. <project xmlns="http://maven.apache.org/POM/4.0.0"
    3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    5.     <parent>
    6.         <artifactId>springboot-demoartifactId>
    7.         <groupId>com.etgroupId>
    8.         <version>1.0-SNAPSHOTversion>
    9.     parent>
    10.     <modelVersion>4.0.0modelVersion>
    11.  
    12.  
    13.     <artifactId>rabbitmqartifactId>
    14.  
    15.  
    16.     <properties>
    17.         <maven.compiler.source>8maven.compiler.source>
    18.         <maven.compiler.target>8maven.compiler.target>
    19.     properties>
    20.     <dependencies>
    21.  
    22.  
    23.         <dependency>
    24.             <groupId>org.springframework.bootgroupId>
    25.             <artifactId>spring-boot-starter-webartifactId>
    26.         dependency>
    27.  
    28.  
    29.         <dependency>
    30.             <groupId>org.springframework.bootgroupId>
    31.             <artifactId>spring-boot-autoconfigureartifactId>
    32.         dependency>
    33.         <dependency>
    34.             <groupId>org.springframework.bootgroupId>
    35.             <artifactId>spring-boot-starter-testartifactId>
    36.             <scope>testscope>
    37.         dependency>
    38.         <dependency>
    39.             <groupId>org.springframework.bootgroupId>
    40.             <artifactId>spring-boot-starter-amqpartifactId>
    41.         dependency>
    42.         <dependency>
    43.             <groupId>org.springframework.bootgroupId>
    44.             <artifactId>spring-boot-starter-amqpartifactId>
    45.         dependency>
    46.         <dependency>
    47.             <groupId>org.projectlombokgroupId>
    48.             <artifactId>lombokartifactId>
    49.         dependency>
    50.     dependencies>
    51. project>

    application.properties

    1. server.port=8088
    2.   #rabbitmq
    3. spring.rabbitmq.host=localhost
    4. spring.rabbitmq.port=5672
    5. spring.rabbitmq.username=admin
    6. spring.rabbitmq.password=admin
    7. spring.rabbitmq.virtual-host=my_vahost

    config

    简单使用

    1. package com.et.rabbitmq.config;
    2.  
    3.  
    4.  
    5.  
    6. import org.springframework.amqp.core.Queue;
    7. import org.springframework.context.annotation.Bean;
    8. import org.springframework.context.annotation.Configuration;
    9.  
    10.  
    11.  
    12.  
    13. @Configuration
    14. public class RabbitConfig {
    15.  
    16.  
    17.     @Bean
    18.     public Queue Queue() {
    19.         return new Queue("hello");
    20.     }
    21. }

    topic 是RabbitMQ中最灵活的一种方式，可以根据routing_key自由的绑定不同的队列 首先对topic规则配置，这里使用两个队列来测试

    1. package com.et.rabbitmq.config;
    2.  
    3.  
    4. import org.springframework.amqp.core.Binding;
    5. import org.springframework.amqp.core.BindingBuilder;
    6. import org.springframework.amqp.core.Queue;
    7. import org.springframework.amqp.core.TopicExchange;
    8. import org.springframework.context.annotation.Bean;
    9. import org.springframework.context.annotation.Configuration;
    10.  
    11.  
    12. @Configuration
    13. public class TopicRabbitConfig {
    14.  
    15.  
    16.    public final static String TOPIC_ONE = "topic.one";
    17.    public final static String TOPIC_TWO = "topic.two";
    18.    public final static String TOPIC_EXCHANGE = "topicExchange";
    19.  
    20.  
    21.    @Bean
    22.    public Queue queue_one(){
    23.       return new Queue(TOPIC_ONE);
    24.    }
    25.  
    26.  
    27.    @Bean
    28.    public Queue queue_two(){
    29.       return new Queue(TOPIC_TWO);
    30.    }
    31.  
    32.  
    33.    @Bean
    34.    TopicExchange exchange(){
    35.       return new TopicExchange(TOPIC_EXCHANGE);
    36.    }
    37.  
    38.  
    39.    @Bean
    40.    Binding bindingExchangeOne(Queue queue_one, TopicExchange exchange){
    41.       return BindingBuilder.bind(queue_one).to(exchange).with("topic.one");
    42.    }
    43.  
    44.  
    45.    @Bean
    46.    Binding bindingExchangeTwo(Queue queue_two, TopicExchange exchange){
    47.       //# 表示零个或多个词
    48.       //* 表示一个词
    49.       return BindingBuilder.bind(queue_two).to(exchange).with("topic.#");
    50.    }
    51.  
    52.  
    53. }

    Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

    1. package com.et.rabbitmq.config;
    2.  
    3.  
    4. import org.springframework.amqp.core.Binding;
    5. import org.springframework.amqp.core.BindingBuilder;
    6. import org.springframework.amqp.core.FanoutExchange;
    7. import org.springframework.amqp.core.Queue;
    8. import org.springframework.context.annotation.Bean;
    9. import org.springframework.context.annotation.Configuration;
    10.  
    11.  
    12. @Configuration
    13. public class FanoutRabbitConfig {
    14.  
    15.  
    16.     @Bean
    17.     public Queue AMessage() {
    18.         return new Queue("fanout.A");
    19.     }
    20.  
    21.  
    22.     @Bean
    23.     public Queue BMessage() {
    24.         return new Queue("fanout.B");
    25.     }
    26.  
    27.  
    28.     @Bean
    29.     public Queue CMessage() {
    30.         return new Queue("fanout.C");
    31.     }
    32.  
    33.  
    34.     @Bean
    35.     FanoutExchange fanoutExchange() {
    36.         return new FanoutExchange("fanoutExchange");
    37.     }
    38.  
    39.  
    40.     @Bean
    41.     Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
    42.         return BindingBuilder.bind(AMessage).to(fanoutExchange);
    43.     }
    44.  
    45.  
    46.     @Bean
    47.     Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
    48.         return BindingBuilder.bind(BMessage).to(fanoutExchange);
    49.     }
    50.  
    51.  
    52.     @Bean
    53.     Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
    54.         return BindingBuilder.bind(CMessage).to(fanoutExchange);
    55.     }
    56.  
    57.  
    58. }

    receiver

    1. package com.et.rabbitmq.receiver;
    2.  
    3.  
    4. import lombok.extern.slf4j.Slf4j;
    5. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    6. import org.springframework.amqp.rabbit.annotation.RabbitListener;
    7. import org.springframework.messaging.handler.annotation.Payload;
    8. import org.springframework.stereotype.Component;
    9. import org.springframework.stereotype.Service;
    10.  
    11.  
    12. @Service
    13. @Slf4j
    14. public class HelloReceiver {
    15.  
    16.  
    17.  
    18.  
    19.     @RabbitListener(queues = "hello")
    20.     public void process(String hello) {
    21.         System.out.println("Receiver  : " + hello);
    22.     }
    25.     @RabbitListener(queues = {"topic.one"})
    26.     public void receiveTopic1(@Payload String fileBody) {
    27.         log.info("topic1：" + fileBody);
    28.     }
    31.     @RabbitListener(queues = {"topic.two"})
    32.     public void receiveTopic2(@Payload String fileBody) {
    33.         log.info("topic2：" + fileBody);
    34.     }
    35.     @RabbitListener(queues = {"fanout.A"})
    36.     public void fanoutA(@Payload String fileBody) {
    37.         log.info("fanoutA：" + fileBody);
    38.     }
    41.     @RabbitListener(queues = {"fanout.B"})
    42.     public void fanoutB(@Payload String fileBody) {
    43.         log.info("fanoutB：" + fileBody);
    44.     }
    45.     @RabbitListener(queues = {"fanout.C"})
    46.     public void fanoutC(@Payload String fileBody) {
    47.         log.info("fanoutC：" + fileBody);
    48.     }
    49. }

    sender

    1. package com.et.rabbitmq.sender;
    2.  
    3.  
    4. import org.springframework.amqp.core.AmqpTemplate;
    5. import org.springframework.beans.factory.annotation.Autowired;
    6. import org.springframework.stereotype.Component;
    7.  
    8.  
    9. import java.util.Date;
    10. @Component
    11. public class HelloSender {
    12.  
    13.  
    14.    @Autowired
    15.    private AmqpTemplate rabbitTemplate;
    16.  
    17.  
    18.    public void send() {
    19.       String context = "hello " + new Date();
    20.       System.out.println("Sender : " + context);
    21.       this.rabbitTemplate.convertAndSend("hello", context);
    22.    }
    23.  
    24.  
    25. }
    1. package com.et.rabbitmq.sender;
    2.  
    3.  
    4. import com.et.rabbitmq.config.TopicRabbitConfig;
    5. import org.springframework.amqp.core.AmqpTemplate;
    6. import org.springframework.beans.factory.annotation.Autowired;
    7. import org.springframework.stereotype.Component;
    8.  
    9.  
    10. @Component
    11. public class TopicSender {
    12.  
    13.  
    14.    @Autowired
    15.    private AmqpTemplate rabbitTemplate;
    16.  
    17.  
    18.    //两个消息接受者都可以收到
    19.    public void send_one() {
    20.       String context = "Hi, I am message one";
    21.       System.out.println("Sender : " + context);
    22.       this.rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,"topic.one",context);
    23.    }
    24.  
    25.  
    26.  
    27.  
    28.    //只有TopicReceiverTwo都可以收到
    29.    public void send_two() {
    30.       String context = "Hi, I am message two";
    31.       System.out.println("Sender : " + context);
    32.       this.rabbitTemplate.convertAndSend(TopicRabbitConfig.TOPIC_EXCHANGE,"topic.two",context);
    33.    }
    34.  
    35.  
    36. }

    DemoApplication.java

    1. package com.et.quartz;
    2.  
    3.  
    4. import org.springframework.boot.SpringApplication;
    5. import org.springframework.boot.autoconfigure.SpringBootApplication;
    6.  
    7.  
    8. @SpringBootApplication
    9. public class DemoApplication {
    10.  
    11.  
    12.    public static void main(String[] args) {
    13.       SpringApplication.run(DemoApplication.class, args);
    14.    }
    15. }

    以上只是一些关键代码，所有代码请参见下面代码仓库

    代码仓库

    • https://github.com/Harries/springboot-demo

    4.测试

    简单使用

    1. @Test
    2. public void hello() throws Exception {
    3.     helloSender.send();
    4.     Thread.sleep(50000);
    5. }

    Topic Exchange

    1. @Test
    2. public void topicOne() throws Exception {
    3.     topicSender.send_one();
    4.     Thread.sleep(50000);
    5. }
    6. @Test
    7. public void topicTwo() throws Exception {
    8.     topicSender.send_two();
    9.     Thread.sleep(50000);
    10. }

    Fanout Exchange

    1. @Test
    2. public void sendFanout() throws InterruptedException {
    3.     String context = "hi, fanout msg ";
    4.     System.out.println("Sender : " + context);
    5.     this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
    6.     Thread.sleep(50000);
    7. }

    5.参考连接

    • https://www.rabbitmq.com/

    • https://spring.io/projects/spring-amqp

  • 相关阅读:
    2023 彩虹全新 SUP 模板，卡卡云模板修复版
    数据库 explain 关键字解析
    Nginx
    理解C#里面的集合有哪些？怎么用，什么是安全集合？
    java基于ssm+Vue儿童福利院管理系统 element 前后端分离
    Java两周半速成之路（第十天）
    51单片机数字电压表仿真设计_LCD显示（仿真+程序+原理图+PCB+设计报告+讲解）
    计算机考研考研院校难度等级，建议收藏
    如何测试服务器的速度
    Fabric.js在vue2中使用
  • 原文地址：https://blog.csdn.net/dot_life/article/details/138174172