SpringBoot 整合 RabbitMQ(快速入门)

╰+哭是因爲堅強的太久メ 2023-05-21 12:53 50阅读 0赞

Linux centOS7 安装rabbitmq的步骤

1.uname -r 检查系统内核版本

![20200418231235946.png][]

2.yum install docker 下载docker

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70][]

3.systemctl start docker 开启运行docker

4.docker -v 查看docker版本

5.systemctl enable docker 设置docker开机启动

6.docker pull rabbitmq:3-management 下载rabbitmq镜像3-management版本的(为什么选management呢?因为这个有web界面)![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 1][]

7.docker images 查看镜像![20200418231733855.png][]

8.docker run -d -p 5672:5672 -p 15672:15672 --name myRabbitmq 30e33de9be86 启动rabbitmq镜像。

\-d:代表后台运行;-p:主机端口映射到docker容器的端口;-p:web管理界面的端口;--name: 设置镜像名称;30e33de9be86 :是rabbitmq镜像的镜像ID

9.测试访问web管理界面: [http://ip:port][http_ip_port]  ;默认用户名guest(密码同);![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 2][]

10.自定义创建交换器,队列,将交换器与队列绑定并设置routingkey,交换器发送消息看队列接收的情况;以及队列响应获取消息等操作。这里就不再赘述了。

[以上内容,见整理文档(链接)][Link 1]

--------------------

### RabbitMQ简介 ###

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的[中间件][Link 2]设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。  
AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。  
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

RabbitMQ常用的交换器类型有direct、topic、fanout、headers四种。

direct:该类型的交换器将所有发送到该交换器的消息被转发到RoutingKey指定的队列中，也就是说路由到BindingKey和RoutingKey完全匹配的队列中.

topic:该类型的交换器将所有发送到Topic Exchange的消息被转发到所有RoutingKey中指定的Topic的队列上面。Exchange将RoutingKey和某Topic进行模糊匹配，其中“*”用来匹配一个词，“\#”用于匹配一个或者多个词。例如“com.\#”能匹配到“com.rabbitmq.oa”和“com.rabbitmq”；而"login.*"只能匹配到“com.rabbitmq”。

fanout:该类型不处理路由键，会把所有发送到交换器的消息路由到所有绑定的队列中。优点是转发消息最快，性能最好。

headers:该类型的交换器不依赖路由规则来路由消息，而是根据消息内容中的headers属性进行匹配。headers类型交换器性能差，在实际中并不常用。

--------------------

SpringBoot 整合RabbitMQ步骤:

目录结构:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 3][]

第一步：创建springboot项目，添加rabbitmq的依赖

<dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-amqp</artifactId>
     </dependency>

第二步：yml文件添加相关配置

#Rabbitmq设置
    spring:
      rabbitmq:
        host: 192.168.1.104 #主机ip
        username: guest #用户名,默认guest
        password: guest #密码,默认guest
        port: 5672 #默认为5672,可以缺省

第三步：SpbRabbitmqApplicationTests类进行测试

@Autowired
        RabbitTemplate rabbitTemplate;
    
        /**
         * exchange.direct 点对点
         * exchange.fanout 广播
         * exchange.topic
         */
    
        /**
         * 发送消息
         * exchange.direct(点对点)
         */
        @Test
        void contextLoads() {
            //Message: 需要手动构造一个消息,定义消息头和消息体.
            //rabbitTemplate.send(String exchange, String routingKey, Message message);
    
            //转换和发送:obj 默认当做消息体的;只需要传入要发送的对象,自动序列化发送给rabbitmq
            //rabbitTemplate.convertAndSend(String exchange, String routingKey, Object obj);
    
            Map<String,Object> map = new HashMap<>();
            map.put("msg","这是用springboot集成rabbitmq发送消息实现的");
            rabbitTemplate.convertAndSend("exchange.direct", "ygzy", map);
            /**
             * 展示在rabbitmq中看到的消息
             * 内容类型:
             * content_type:	application/x-java-serialized-object (默认是使用java序列化对象,所以看到的消息是那样似的,稍后添加配置类,转为json数据进行转发)
             * 接收到的消息:
             * rO0ABXNyABFqYXZhLnV0aWwuSGFzaE1hcAUH2sHDFmDRAwACRgAKbG9hZEZhY3RvckkACXRocmVzaG9sZHhwP0AAAAAAAAx3CAAAABAAAAABdAADbXNndAA2
             * 6L+Z5piv55Soc3ByaW5nYm9vdOmbhuaIkHJhYmJpdG1x5Y+R6YCB5raI5oGv5a6e546w55qEeA==
             */
    
            /**
             * 通过配置消息配置,进行转换后,在rabbitmq中看到的消息。（见第四步）
             * content_encoding:	UTF-8
             * content_type:	application/json
             * {"msg":"这是用springboot集成rabbitmq发送消息实现的"}
             */
        }
    
        @Test
        public void receive(){
            Message ygzy = rabbitTemplate.receive("ygzy");
            System.out.println(ygzy.toString());
            /**
             * (未配置消息转换器之前)打印收到的消息
             * (Body:'{msg=这是用springboot集成rabbitmq发送消息实现的}' MessageProperties [headers={},
             * contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT,
             * priority=0, redelivered=true, receivedExchange=exchange.direct, receivedRoutingKey=ygzy, deliveryTag=1, messageCount=0])
             */
    
            /**
             * 配置了消息转换器之后,打印接收到的消息。（见第四步）
             * (Body:'{"msg":"这是用springboot集成rabbitmq发送消息实现的"}' MessageProperties [headers={__ContentTypeId__=java.lang.Object,
             * __KeyTypeId__=java.lang.Object, __TypeId__=java.util.HashMap}, contentType=application/json, contentEncoding=UTF-8,
             * contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=true, receivedExchange=exchange.direct,
             * receivedRoutingKey=ygzy, deliveryTag=1, messageCount=0])
             */
        }

第四步：配置消息转换器配置类

@Configuration //说明该类是一个配置类
    public class MyAMQPConfig {
        /**
         * 自定义一个消息转换器,将发送的消息转换为json数据
         * @return
         */
        @Bean
        public MessageConverter messageConverter() {
            return new Jackson2JsonMessageConverter();
        }
    }

第五步：增加了第四步操作，可以回到第三步测试，看一下效果。

第六步：创建消息服务类监听消息，并在项目启动类上开启该功能。

@Service
    public class MessageService {
    
        @RabbitListener(queues = {"lzygzy.news","ygzy.news","ygzy.users"})
        public void reciev(Object msg) {
            System.out.println(Object.class);
            System.out.println(msg.toString());
        }
    }
    
    /**
     * 1、添加RabbitMQ依赖
     * 2、配置RabbitMQ基本配置
     *
     * 自动配置
     * 1.RabbitAutoConfiguration
     * 2.有自动配置连接工厂ConnectionFactory
     * 3.RabbitProperties 封装了RabbitMQ的配置
     * 4.RabbitTemplate 给RabbitMQ方和接受消息
     * 5.AmqpAdmin: RabbitMQ系统管理功能组件,负责创建交换器和队列用的
     * 6.@EnableRabbit + @RabbitListener 监听消息队列的内容
     */
    @EnableRabbit // 开启
    @SpringBootApplication
    public class SpbRabbitmqApplication {
    
        public static void main(String[] args) {
            SpringApplication.run(SpbRabbitmqApplication.class, args);
        }
    
    }

这一步，通过测试类进行发送数据，然后启动项目监听接收数据。

第七步：系统管理组件的应用

@Autowired
        AmqpAdmin amqpAdmin;//注入系统管理功能组件
    
        /**
         * 系统管理功能组件的使用
         * 以点对点为例
         */
        @Test
        public void amqpAdminManager(){
            //创建一个(点对点)交换器
            amqpAdmin.declareExchange(new DirectExchange("exchange.ygzy"));
    
            //创建一个队列
            amqpAdmin.declareQueue(new Queue("ygzylz", true));
    
            //绑定的有参构造方法Binding(String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map<String, Object> arguments)
            /**
             * destination :目的地(队列)
             * destinationType :类型(是一个队列)
             * exchange: 交换器
             * routingKey:路由件
             */
            amqpAdmin.declareBinding(new Binding("ygzylz", Binding.DestinationType.QUEUE,"exchange.ygzy","amqpadmin.ygzy",null));
    
            Map<String, Object> paramMap = new HashMap<String, Object>();
            paramMap.put("msg","这是用系统管理组件创建exchange,queue,routingkey 并完成绑定的操作");
            rabbitTemplate.convertAndSend("exchange.ygzy","amqpadmin.ygzy",paramMap);
    
            Message ygzylz = rabbitTemplate.receive("ygzylz");
            System.out.println(ygzylz.toString());//输出的内容:(Body:'{"msg":"这是用系统管理组件创建exchange,queue,routingkey 并完成绑定的操作"}' MessageProperties [headers={__ContentTypeId__=java.lang.Object, __KeyTypeId__=java.lang.Object, __TypeId__=java.util.HashMap}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=exchange.ygzy, receivedRoutingKey=amqpadmin.ygzy, deliveryTag=1, messageCount=0])
             
        }

[20200418231235946.png]: /images/20230521/650c0a2b7b5c40d4a5875ea1506ffeb7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70]: /images/20230521/b478aaf46bb64c36848ce79559cc56ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 1]: /images/20230521/4c53c576ff09478396a2c918b93c1565.png
[20200418231733855.png]: /images/20230521/d6e5a11fbcc44a22b953a6f09d6b0285.png
[http_ip_port]: http://ip
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 2]: /images/20230521/871009ba7b9a424db772268993d3f64f.png
[Link 1]: https://download.csdn.net/download/lz527657138/12338411
[Link 2]: http://www.diggerplus.org/archives/tag/%E4%B8%AD%E9%97%B4%E4%BB%B6
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x6NTI3NjU3MTM4_size_16_color_FFFFFF_t_70 3]: /images/20230521/246268cbd19842fdadd2936d6f45240b.png