MQ  
  
  
  
  
，中文是消息队列（MessageQueue） 
  
  
  
  
 ，字面来看就是存放消息的队列  
  
  
  
  
。也就是事件驱动架构中的Broker 
  
  
  
  
 。

快速入门

1.publisher实现

public class PublisherTest { @Test public void testSendMessage() throws IOException, TimeoutException { // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1.设置连接参数


 


 


 


 


 ，分别是：主机名  
  
  
  
  
、端口号 
  
  
  
  
 、vhost


 


 


 


 


 、用户名 
 
 
 
 
、密码 factory.setHost("192.168.150.101"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("rabbit"); factory.setPassword("123321"); // 1.2.建立连接 Connection connection = factory.newConnection(); // 2.创建通道Channel Channel channel = connection.createChannel(); // 3.创建队列 String queueName = "simple.queue"; channel.queueDeclare(queueName, false, false, false, null); // 4.发送消息 String message = "hello, rabbitmq!"; channel.basicPublish("", queueName, null, message.getBytes()); System.out.println("发送消息成功：【" + message + "】"); // 5.关闭通道和连接 channel.close(); connection.close(); } }

2.consumer实现

public class ConsumerTest { public static void main(String[] args) throws IOException, TimeoutException { // 1.建立连接 ConnectionFactory factory = new ConnectionFactory(); // 1.1.设置连接参数 
 
 
 
 
，分别是：主机名  
   
   
   
   
 、端口号

 

 

 

 

 、vhost
 

 

 

 

 
、用户名          、密码 factory.setHost("192.168.150.101"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("itcast"); factory.setPassword("123321"); // 1.2.建立连接 Connection connection = factory.newConnection(); // 2.创建通道Channel Channel channel = connection.createChannel(); // 3.创建队列 String queueName = "simple.queue"; channel.queueDeclare(queueName, false, false, false, null); // 4.订阅消息 channel.basicConsume(queueName, true, new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // 5.处理消息 String message = new String(body); System.out.println("接收到消息：【" + message + "】"); } }); System.out.println("等待接收消息


 


 


 


 


 。 
 
 
 
 
。  
   
   
   
   
 。

 

 

 

 

 。"); } }

SpringAMQP-集成SpringBoot

1.Basic Queue 简单队列模型

a.导入依赖

<!--AMQP依赖  
   
   
   
   
 ，包含RabbitMQ--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>

b.消息发送

首先配置MQ地址

 

 

 

 

 ，在publisher服务的application.yml中添加配置：

spring: rabbitmq: host: 192.168.150.101 # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: rabbit # 用户名 password: 123321 # 密码

然后在publisher服务中编写测试类SpringAmqpTest
 

 

 

 

 
，并利用RabbitTemplate实现消息发送：

@RunWith(SpringRunner.class) @SpringBootTest public class SpringAmqpTest { @Autowired private RabbitTemplate rabbitTemplate; @Test public void testSimpleQueue() { // 队列名称 String queueName = "simple.queue"; // 消息 String message = "hello, spring amqp!"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); } }

c.消息接收

首先配置MQ地址          ，在consumer服务的application.yml中添加配置：

spring: rabbitmq: host: 192.168.150.101 # 主机名 port: 5672 # 端口 virtual-host: / # 虚拟主机 username: rabbit # 用户名 password: 123321 # 密码

然后在consumer服务的cn.itcast.mq.listener包中新建一个类SpringRabbitListener
 

 
 

 
 

 
 

 
 

 ，代码如下：

package cn.itcast.mq.listener; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class SpringRabbitListener { @RabbitListener(queues = "simple.queue") public void listenSimpleQueueMessage(String msg) throws InterruptedException { System.out.println("spring 消费者接收到消息：【" + msg + "】"); } }

2.WorkQueue-任务队列

Work queues









，也被称为（Task queues）               ，任务模型
 

 

 

 

 
。简单来说就是让多个消费者绑定到一个队列
 
 
 
 
 
 
 
 
 
 ，共同消费队列中的消息          。

当消息处理比较耗时的时候














，可能生产消息的速度会远远大于消息的消费速度
 

 
 

 
 

 
 

 
 

 。长此以往          ，消息就会堆积越来越多
 
  
 
  
 
  
 
  
 
  ，无法及时处理









。

此时就可以使用work 模型









，多个消费者共同处理消息处理  
  
  
  
  
，速度就能大大提高了               。

但在默认情况下 
  
  
  
  
 ，消息是平均分配给每个消费者


 


 


 


 


 ，并没有考虑到消费者的处理能力
 
 
 
 
 
 
 
 
 
 。这样显然是有问题的














。

在spring中有一个简单的配置 
 
 
 
 
，可以解决这个问题          。我们修改consumer服务的application.yml文件  
   
   
   
   
 ，添加配置：

spring: rabbitmq: listener: simple: prefetch: 1 # 每次只能获取一条消息

 

 

 

 

 ，处理完成才能获取下一个消息

3.Fanout-广播

在广播模式下
 

 

 

 

 
，消息发送流程是这样的：

1） 可以有多个队列 2） 每个队列都要绑定到Exchange（交换机） 3） 生产者发送的消息          ，只能发送到交换机
 

 
 

 
 

 
 

 
 

 ，交换机来决定要发给哪个队列









，生产者无法决定 4） 交换机把消息发送给绑定过的所有队列 5） 订阅队列的消费者都能拿到消息

a.声明队列和交换机

Spring提供了一个接口Exchange               ，来表示所有不同类型的交换机
 
 
 
 
 
 
 
 
 
 ，

在consumer中创建一个类














，声明队列和交换机：

@Configuration public class FanoutConfig { /** * 声明交换机 * @return Fanout类型交换机 */ @Bean public FanoutExchange fanoutExchange(){ return new FanoutExchange("rabbit.fanout"); } /** * 第1个队列 */ @Bean public Queue fanoutQueue1(){ return new Queue("fanout.queue1"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange); } /** * 第2个队列 */ @Bean public Queue fanoutQueue2(){ return new Queue("fanout.queue2"); } /** * 绑定队列和交换机 */ @Bean public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){ return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange); } }

b.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test public void testFanoutExchange() { // 队列名称 String exchangeName = "rabbit.fanout"; // 消息 String message = "hello, everyone!"; rabbitTemplate.convertAndSend(exchangeName, "", message); }

c.消息接收

在consumer服务的SpringRabbitListener中添加两个方法          ，作为消费者：

@RabbitListener(queues = "fanout.queue1") public void listenFanoutQueue1(String msg) { System.out.println("消费者1接收到Fanout消息：【" + msg + "】"); } @RabbitListener(queues = "fanout.queue2") public void listenFanoutQueue2(String msg) { System.out.println("消费者2接收到Fanout消息：【" + msg + "】"); }

4.Direct-指定

在Fanout模式中
 
  
 
  
 
  
 
  
 
  ，一条消息









，会被所有订阅的队列都消费
 
  
 
  
 
  
 
  
 
  。但是  
  
  
  
  
，在某些场景下 
  
  
  
  
 ，我们希望不同的消息被不同的队列消费









。这时就要用到Direct类型的Exchange  
  
  
  
  
。

在Direct模型下：

队列与交换机的绑定


 


 


 


 


 ，不能是任意绑定了 
 
 
 
 
，而是要指定一个RoutingKey（路由key） 消息的发送方在 向 Exchange发送消息时  
   
   
   
   
 ，也必须指定消息的 RoutingKey 
  
  
  
  
 。 Exchange不再把消息交给每一个绑定的队列

 

 

 

 

 ，而是根据消息的Routing Key进行判断
 

 

 

 

 
，只有队列的Routingkey与消息的 Routing key完全一致          ，才会接收到消息

案例需求如下：

利用@RabbitListener声明Exchange
 

 
 

 
 

 
 

 
 

 、Queue









、RoutingKey

在consumer服务中
 

 
 

 
 

 
 

 
 

 ，编写两个消费者方法









，分别监听direct.queue1和direct.queue2

在publisher中编写测试方法               ，向itcast. direct发送消息

a.基于注解声明队列和交换机

基于@Bean的方式声明队列和交换机比较麻烦
 
 
 
 
 
 
 
 
 
 ，Spring还提供了基于注解方式来声明


 


 


 


 


 。

在consumer的SpringRabbitListener中添加两个消费者














，同时基于注解来声明队列和交换机：

@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "rabbit.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"} )) public void listenDirectQueue1(String msg){ System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "rabbit.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"} )) public void listenDirectQueue2(String msg){ System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】"); }

b.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test public void testSendDirectExchange() { // 交换机名称 String exchangeName = "rabbit.direct"; // 消息 String message = "红色警报！日本乱排核废水          ，导致海洋生物变异
 
  
 
  
 
  
 
  
 
  ，惊现哥斯拉！"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "red", message); }

5.Topic-主题

Topic类型的Exchange与Direct相比









，都是可以根据RoutingKey把消息路由到不同的队列 
 
 
 
 
。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

案例需求：

实现思路如下：

并利用@RabbitListener声明Exchange               、Queue
 
 
 
 
 
 
 
 
 
 、RoutingKey

在consumer服务中  
  
  
  
  
，编写两个消费者方法 
  
  
  
  
 ，分别监听topic.queue1和topic.queue2

在publisher中编写测试方法


 


 


 


 


 ，向itcast. topic发送消息

a.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

/** * topicExchange */ @Test public void testSendTopicExchange() { // 交换机名称 String exchangeName = "rabbit.topic"; // 消息 String message = "喜报！孙悟空大战哥斯拉 
 
 
 
 
，胜!"; // 发送消息 rabbitTemplate.convertAndSend(exchangeName, "china.news", message); }

b.消息接收

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue1"), exchange = @Exchange(name = "rabbit.topic", type = ExchangeTypes.TOPIC), key = "china.#" )) public void listenTopicQueue1(String msg){ System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】"); } @RabbitListener(bindings = @QueueBinding( value = @Queue(name = "topic.queue2"), exchange = @Exchange(name = "rabbit.topic", type = ExchangeTypes.TOPIC), key = "#.news" )) public void listenTopicQueue2(String msg){ System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】"); }

6.消息转换器

Spring会把你发送的消息序列化为字节发送给MQ  
   
   
   
   
 ，接收消息的时候

 

 

 

 

 ，还会把字节反序列化为Java对象  
   
   
   
   
 。

只不过
 

 

 

 

 
，默认情况下Spring采用的序列化方式是JDK序列化

 

 

 

 

 。众所周知          ，JDK序列化存在下列问题：

数据体积过大 有安全漏洞 可读性差

因此可以使用JSON方式来做序列化和反序列化
 

 

 

 

 
。

在publisher和consumer两个服务中都引入依赖：

<dependency> <groupId>com.fasterxml.jackson.dataformat</groupId> <artifactId>jackson-dataformat-xml</artifactId> <version>2.9.5</version> </dependency>

配置消息转换器          。

在启动类中添加一个Bean即可：

@Bean public MessageConverter jsonMessageConverter(){ return new Jackson2JsonMessageConverter(); }

#安装(Docker)

1.拉取镜像

docker pull rabbitmq:3-management

2.部署容器

docker run \ -e RABBITMQ_DEFAULT_USER=rabbit \ -e RABBITMQ_DEFAULT_PASS=123321 \ -v mq-plugins:/plugins \ --name mq \ --hostname mq \ --restart=always \ -p 15672:15672 \ -p 5672:5672 \ -d \ rabbitmq:3-management