RabbitMQ的介绍，安装以及基本的使用整合springboot

野性酷女 2022-01-21 10:39 242阅读 0赞

#### 1.什么是 RabbitMQ ####

RabbitMQ是一个由erlang开发的AMQP(Advanved Message Queue Protocol)的开源实现。

> AMQP（Advanced Message Queuing Protocol）：高级消息队列协议，也是一个消息代理的规范，兼容JMS，RabbitMQ是AMQP的实现

#### 2.核心组件的概念： ####

**（1）Message**  
消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routingkey（路由键）、 priority（相对于其他消息的优先权）、 delivery-mode（指出该消息可能需要持久性存储）等。

**（2）Publisher**  
消息的生产者，也是一个向交换器发布消息的客户端应用程序

**（3）Exchange**  
交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有4种类型： direct(默认)， fanout, topic, 和headers不同类型的Exchange转发消息的策略有所区别。

**（4）Binding**  
绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange 和Queue的绑定可以是多对多的关系。

**(5)Connection**  
网络连接，比如一个TCP连接。

**(6)Channel**  
信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接， AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所  
以引入了信道的概念，以复用一条 TCP 连接。

**(7)Consumer**  
消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。

**(8)Virtual Host**  
虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。 vhost 是 AMQP 概念的基础，必须在连接时指定，  
RabbitMQ 默认的 vhost 是 /。

**(9)Broker**  
表示消息队列服务器实体。

**架构关系图如下：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center]

#### 3.RabbitMQ的安装 ####

**(1)下载镜像**

[root@localhost docker]# dokcer pull pshk7r7h.mirror.swr.myhuaweicloud.com/library/rabbitmq:3-management

> 因为如果采用直接官网拉取镜像的话，速度会比较慢，这里采用了镜像加速

**(2)启动镜像**

//先查看所有镜像
    [root@localhost docker]# docker image
    REPOSITORY                                               TAG                 IMAGE ID            CREATED             SIZE
    pshk7r7h.mirror.swr.myhuaweicloud.com/library/rabbitmq   3-management        c4663bdca2cd        4 days ago          189 MB
    docker.io/tomcat                                         7                   bf7d94e5177f        3 months ago        359 MB
    docker.io/mysql                                          5.7                 ee7cbd482336        3 months ago        372 MB
    //启动rabbitmq，并且映射两个端口，一个是服务器访问端口 5672 ，一个是访问web控制后台的端口 15672
    [root@localhost docker]# docker run -d -p 5672:5672  -p 15672:15672 --name rabbitmq c4663bdca2cd
    2fdec2c4ec2f863c269b3e3d72f0b628b601364ea18a4472937ef14c01a8d627

**访问后台:第一次访问是要用户名跟密码的 默认都是guest**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70]

#### 4.Exchange 类型 ####

Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、 fanout、 topic、 headers 。 **(交换机通过routing key 与queue消息队列进行绑定)**

**(1)direct**  
Direct Exchange 也就所了解的单播形式，消息中的路由键（routing key）如果和 Binding 中的 bindingkey 一致， 交换器就将消息发到对应的队列中。路由键与队列名完全匹配。如果一个队列绑定到交换机要求路由键为“user”，则只转发 routing key 标记为“user”的消息，不会转转发其他的。它是完全匹配、单播的模式

**(2)fanout**  
这是一种多播形式，每一个通过fanout 类型的交换器转发的消息，会转发到所有绑定的队列当中。不管routing key是什么。这种类型就是多播，广播形式。

**(3)topic**交换器通过模式匹配分配消息的路由键属性，将路由键和某个模式进行匹配，此时队列需要绑定到一个模式上。它将路由键和绑定键的字符串切分成单词，这些单词之间用点隔开。它同样也会识别两个通配符：符号“\#”和符号“\*” 。 \#匹配0个或多个单词， \*匹配一个单词。  
比如 queue1 和一个topic 交换机的消息绑定的key 为 \#.user,那么通过该topic,转发的消息 只要该消息的routing-key是这种形式xxx.user或xxx.xxx.user（匹配\#.user形式），那么该消息就会被转发到queue1队列当中。

\*\*(4)headers \*\*  
headers 匹配 AMQP 消息的 header而不是路由键， headers 交换器和 direct 交换器完全一致，但性能差很多，现在很少用到了。

#### 5.spring-boot整合rabbitmq ####

**(1)创建springboot工程，导入rabbitmq场景启动器依赖**

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

**(2)看一下源码是如何进行自动配置**  
`RabbitAutoConfiguration部分代码`

@Configuration
    	@ConditionalOnMissingBean(ConnectionFactory.class)
    	protected static class RabbitConnectionFactoryCreator { 
    
    		// Only available in rabbitmq-java-client 5.4.0 +
    		private static final boolean CAN_ENABLE_HOSTNAME_VERIFICATION = ReflectionUtils
    				.findMethod(com.rabbitmq.client.ConnectionFactory.class,
    						"enableHostnameVerification") != null;
    
    		@Bean
    		public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config)
    				throws Exception { 
    			RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();
    			if (config.determineHost() != null) { 
    				factory.setHost(config.determineHost());
    			}
    			factory.setPort(config.determinePort());
    			if (config.determineUsername() != null) { 
    				factory.setUsername(config.determineUsername());
    			}
    			if (config.determinePassword() != null) { 
    				factory.setPassword(config.determinePassword());
    			}
    			if (config.determineVirtualHost() != null) { 
    				factory.setVirtualHost(config.determineVirtualHost());
    			}
    			if (config.getRequestedHeartbeat() != null) { 
    				factory.setRequestedHeartbeat(config.getRequestedHeartbeat());
    			}
    
    		@Bean
    		@ConditionalOnSingleCandidate(ConnectionFactory.class)
    		@ConditionalOnMissingBean(RabbitTemplate.class)
    		public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { 
    			RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    			MessageConverter messageConverter = this.messageConverter.getIfUnique();
    			if (messageConverter != null) { 
    				rabbitTemplate.setMessageConverter(messageConverter);
    			}
    			rabbitTemplate.setMandatory(determineMandatoryFlag());
    			RabbitProperties.Template templateProperties = this.properties.getTemplate();
    			RabbitProperties.Retry retryProperties = templateProperties.getRetry();
    			if (retryProperties.isEnabled()) { 
    				rabbitTemplate.setRetryTemplate(createRetryTemplate(retryProperties));
    			}
    			if (templateProperties.getReceiveTimeout() != null) { 
    				rabbitTemplate.setReceiveTimeout(templateProperties.getReceiveTimeout());
    			}
    			if (templateProperties.getReplyTimeout() != null) { 
    				rabbitTemplate.setReplyTimeout(templateProperties.getReplyTimeout());
    			}
    			return rabbitTemplate;
    		}
    
    		@Bean
    		@ConditionalOnSingleCandidate(ConnectionFactory.class)
    		@ConditionalOnProperty(prefix = "spring.rabbitmq", name = "dynamic",
    				matchIfMissing = true)
    		@ConditionalOnMissingBean(AmqpAdmin.class)
    		public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) { 
    			return new RabbitAdmin(connectionFactory);
    		}
    
    	}

主要导入了这三个组件：  
rabbitConnectionFactoryl连接工厂，RabbitTemplate用户发送消息（如jmsTemplate）, AmqpAdmin用来管理后台的组件（如添加，删除，队列，交换机，然后将交换机与消息队列通过 routing-key 进行绑定）。

因为是自动配置，我们只要编写最基本的配置文件即可，配置rabbit服务器的信息

###配置rabbitmq服务器信息
    spring.rabbitmq.host=192.168.25.131
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    
    ### 默认就是5672可以不进行配置
    spring.rabbitmq.port=5672

**(3)开始测试**  
用AmqpAdmin，给服务器添加一个消息队列 name为 `my.amqpAdmin.queue,`然后再创建一个类型为`direct`的交换机 `my.direct`，将他们同个进行绑定，绑定的`routing-key`为`my.routing`**代码如下**

package com.kuake.srpingboot;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.amqp.core.*;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringRunner;
    
    @RunWith(SpringRunner.class)
    @SpringBootTest
    public class Srpingboot02RabbitmqApplicationTests { 
    
    	@Autowired
    	private AmqpAdmin amqpAdmin;
    
    	@Test
    	public void contextLoads() { 
    		/* * Exchange是一个接口实现类如下： * 1.DirectExchange * 2.FanoutExchange * 3.TopicExchange * */
    		//创建类型是Direct name=my.direct的交换机
    		Exchange exchange = new DirectExchange("my.direct");
    		amqpAdmin.declareExchange(exchange);
    
    		//创建一个name=my.amqpAdmin.queue的队列
    		/* * Queue中可以设置的属性： * String name, 名字 * boolean durable, 持久化 * boolean exclusive, 声明一个独占队列 * boolean autoDelete, 是否自动删除 * Map<String, Object> arguments 用于声明队列的参数 * * */
    		amqpAdmin.declareQueue(new Queue("my.amqpAdmin.queue"));
    
    		//队列跟交换机进行绑定
    		/** * Binding可以设置的参数： * destination 目的地 * destinationType 绑定的类型 是跟队列绑定 还是交换机绑定 * exchange 交换机的name * routingKey 路由的key * arguments; 参数 可以不设置 * */
    		Binding binding = new Binding("my.amqpAdmin.queue", Binding.DestinationType.QUEUE,"my.direct","my.routing",null);
    		amqpAdmin.declareBinding(binding);
    
    	}
    
    }

**进入后台:**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 3]  
测试成功。  
**(4)使用AmqpTemplate测试发送和接收数据**  
发送消息：

@Test
    	public void testsend(){ 
    		/** * exchange: 使用发送消息的交换机 * routingKey: 发送消息使用路由的key * message: 需要发送的消息 * */
    		//根据 direct交换机的特性 绑定的key 一定要等于 路由的key 相绑定的queue才能接收消息
    		amqpTemplate.convertAndSend("my.direct","routing","hello world !");
    	}

**进入后台检查一下：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 4]  
**发现消息队列中有一条消息：**

**进行接收接收测试：**

@Test
    	public void testrecive(){ 
    		//指定队列的名字，进行接收
    		Object o = amqpTemplate.receiveAndConvert("my.amqpAdmin.queue");
    		System.out.println(o);
    	}

消息成功获取了  
![在这里插入图片描述][20190610165633420.png]  
`Queue my.amqpAdmin.queue`中消息队列也为空了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 5]  
我知道，ActiveMQ的消息形式有五种，但是RabbitMq就只有一种就是通过byte\[\]二进制数组传输的，这样一来就涉及一个序列化机制，可以测试一下发送一个对象。

@Test
    	public void testsendObject(){ 
    		/** * exchange: 使用发送消息的交换机 * routingKey: 发送消息使用路由的key * message: 需要发送的消息 * */
    // Employee employee = new Employee("jack",12,new Date());
    		Map<String,Object>map = new HashMap<>();
    		map.put("msg","hello world");
    		map.put("list",Arrays.asList("abc",null,123));
    		amqpTemplate.convertAndSend("my.direct","my.routing",map);
    	}

发送成功之后，发现  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 6]  
对象被java默认采用的消息转换是`messageConverter = new SimpleMessageConverter();`按照这个序列化，按照我们平时的习惯让对象以json字符串的形式存储，所以我们可以使用`Jackson2JsonMessageConverter`，来进行对象的序列化。  
配置如下：

package com.kuake.srpingboot.config;
    
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /** * @author hao * @create 2019-06-10 ${TIM} */
    @Configuration
    public class MyConfig { 
        @Bean
        public MessageConverter messageConverter(){ 
            return  new Jackson2JsonMessageConverter();
        }
    }

再次发送对象消息：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 7]  
可以看到我们的消息已经就被序列化成json字符串了。

**(5)使用@RabbitListener**来监听消息，只要监听的消息队列新有消息，就可以立刻把他获取出来。(要**使用@EnableRabbit声明开启注解版**)

@Component
    public class QueueListener{ 
        //可以监听多个queue
        @RabbitListener(queues = { "my.amqpAdmin.queue"})
        public void receive(Map map){ 
            System.out.println("监听接收到消息"+map);
        }
    }

**编写测试代码，连续发送多条消息**

@Test
    	public void testsendObject(){ 
    		/** * exchange: 使用发送消息的交换机 * routingKey: 发送消息使用路由的key * message: 需要发送的消息 * */
    		
    		for(int i=0;i<10;i++){ 
    			Map<String,Object>map = new HashMap<>();
    			map.put("mag","这是第"+i+"次=====================》hello world");
    			amqpTemplate.convertAndSend("my.direct","my.routing",map);
    		}
    	}

看控制台打印：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 8]  
消息发送之后，立马被监听到了 然后就打印出来了。

**总结：**  
(1)rabbitmq是基于amqp协议的一种高级消息队列，它可以设置交换器的的类型，从而控制消息转发的类型的类型。  
1.direct 单播模式 一对一  
2.fanout 多播形式，消息经这种类型的交换机转发 会到达与之绑定的任何消息队列  
3.topic 多播形式，根据routing-key的名字 匹配规则来进行转发。  
(2)与springboot整合通过RabbitTemplate用户发送消息 , 通过AmqpAdmin对象管理后台的组件（如添加，删除，队列，交换机，然后将交换机与消息队列通过 routing-key 进行绑定）。

> 知识在于积累，虽然不用很容易忘记，但是记录之后，方便自己复习…贵在坚持

参考资料：

*  rabbitmq官网介绍
 *  rabbitmq 百度百科

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20220121/a671fb9d10fb47cfa45752948a991a2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70]: /images/20220121/1668b84932a54fd19709125e9c100fd9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 1]: /images/20220121/e0b56acc9bc64382ae51dd2e7a2cc670.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 2]: /images/20220121/5d70b43c8b85497390f3a6c8662be3ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 3]: /images/20220121/d901f144d1be4e40bc1d05f46d05b86a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 4]: /images/20220121/3825699e7ca14e9ebdc5547915334c31.png
[20190610165633420.png]: /images/20220121/58b166e847ca4a84abf6339521e572da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 5]: /images/20220121/b8a32f1341ec4a82ba870c59e6cdac73.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 6]: /images/20220121/d26b62e36e034d029f3c0c77ce81ed44.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 7]: /images/20220121/2ef1184f321742c5b2060b379ac822d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzczMjk1NQ_size_16_color_FFFFFF_t_70 8]: /images/20220121/bfd605cdc24a4d2e850ca144501ff951.png