RabbitMQ-SpringAMQP使用介绍

RabbitMQ采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。

1. Spring AMQP

RabbitMQ官方提供的Java客户端编码相对复杂，一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具：SpringAMQP。并且还基于SpringBoot对其实现了自动装配，使用起来非常方便。

官网：https://spring.io/projects/spring-amqp/

SpringAMQP提供了三个功能：

• 自动声明队列、交换机及其绑定关系
• 基于注解的监听器模式，异步接收消息
• 封装了RabbitTemplate工具，用于发送消息

1.1 引入依赖

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

1.2 消息发送

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

spring:rabbitmq:host: 192.168.1.200 # 虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: feng # 用户名password: 123 # 密码

然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：

@SpringBootTest
public class SpringAmopTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void SimpleQueue(){//队列String queueName = "simpleQueue";//消息String message = "Hello World";//发送消息rabbitTemplate.convertAndSend(queueName,message);}
}

1.3 消息接收

在consumer端的application.xml配置MQ信息：

spring:rabbitmq:host: 192.168.1.200 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: feng # 用户名password: 123321 # 密码

编写监听器

@Component
@Slf4j
public class SpringRabbitListener {@RabbitListener(queues = "simpleQueue")public void listenSimpleQueue(String message){log.info("Simple Queue消息： "+message);}
}

重启服务，即可实现监听！

1.4 WorkQueue模型

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

1.4.1 实例代码

publisher段代码：

@Test
public void WorkQueue(){//队列String queueName = "work.queue";//消息String message = "Hello World_";//发送消息50次for (int i = 0; i < 50; i++) {rabbitTemplate.convertAndSend(queueName,message+i);}}

consumer段消费代码：

//消费者1
@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String message){System.out.println("消费者1处理任务{："+message+"}"+ LocalTime.now());
}
//消费者2
@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String message){System.err.println("消费者2处理任务{："+message+"}"+ LocalTime.now());
}

work queue模式，消息分配模式默认是平均分配，每个消费者处理的任务数量是平均的

也就是上面案例中，消息队列一共五十条消息，消费者1和消费者2，每人会处理25条。

1.4.2 能者多劳

如果想修改work模式的分配模式，比如想按性能分配，性能好的，多处理任务，那些就需要添加配置：

application.xml：

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

这样，50条消息，由消费者1和消费者2 按性能快慢决定执行数量的多少，谁先执行完成当前任务，就可以去执行消息队列中的下一个任务，没有预分配机制；

1.4.3 总结

work queue模型的使用：

• 多个消费者绑定到一个消息队列，可以加快消息处理速度；
• 同一条消息只会被一个消费者处理；
• 通过设置prefetch来控制消费者预取的消息数量，处理完一条再处理下一条，实现能者多劳；

1.5交换机

交换机exchange，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。

• Publisher：生产者，不再发送消息到队列中，而是发给交换机
• Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
• Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
• Consumer：消费者，与以前一样，订阅队列，没有变化

交换机的作用：

• 接收publisher发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失

交换机的类型有四种：

• Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
• Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
• Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
• Headers：头匹配，基于MQ的消息头匹配，用的较少。

1.6 Fanout交换机（广播）

FanoutExchange会将消息路由到每个绑定的队列。

• 可以有多个队列
• 每个队列都要绑定到Exchange（交换机）
• 生产者发送的消息，只能发送到交换机
• 交换机把消息发送给绑定过的所有队列
• 订阅队列的消费者都能拿到消息

1.7 Direct交换机（订阅）

基于RoutingKey（路由key）发送给订阅了消息的队列。

在Direct模型下：

• 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
• 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。
• Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

Direct交换机与Fanout交换机差异：

• Fanout交换机将消息路由给每一个与之绑定的队列
• Direct交换机根据RoutingKey判断路由给哪个队列
• 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

1.8 Topic交换机（通配符订阅）

Topic与direct相似，都是通过RoutingKey绑定，按照RoutingKey进行路由，只不过Topic类型的Exchange在绑定bandingKey时，可以使用通配符配置：

通配符规则：

• #：匹配一个或多个词
• *：匹配不多不少恰好1个词

举例：

• item.#：能够匹配item.spu.insert 或者 item.spu
• item.*：只能匹配item.spu

Direct交换机与Topic交换机的差异：

• Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割
• Topic交换机与队列绑定时的bindingKey可以指定通配符
• #：代表0个或多个词
• *：代表1个词

1.9 声明队列and交换机

fanout交换机

package com.itheima.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutConfig {/*** 声明Fanout交换机* @return Fanout交换机类型*/public FanoutExchange fanoutExchange() {return new FanoutExchange("fanoutExchange", true, false);}/*** 声明队列* @return Queue类型*/public Queue queue() {return new Queue("queue", true, false, false);}/*** 队列绑定到交换机* @param queue* @param fanoutExchange* @return*/public Binding binding(Queue queue ,FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue).to(fanoutExchange);}
}

Direct交换机

package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectConfig {/*** 声明交换机* @return Direct类型交换机*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1个队列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2个队列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}

1.10 基于注解声明

Direct模式

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

Topic模式：

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

1.11 消息转换器

Spring的消息发送代码接收的消息体是一个Object，而在数据传输时，它会把发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

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

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

1.11.1 配置JSON转换器

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

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

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。

配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：

@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;
}

消息转换器中添加的messageId可以便于我们将来做幂等性判断。

1.11.2 示例

publisher

    @Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void SimpleQueue(){//队列String queueName = "simpleQueue";//消息Map<String,Object> map = new HashMap<>(2);map.put("name", "xiaoming");map.put("age", 18);
//        String message = "测试消息！！！";//发送消息rabbitTemplate.convertAndSend(queueName,map);}

consumer

@RabbitListener(queues = "simpleQueue")
public void listenSimpleQueue(Map<String, Object> map) throws InterruptedException {log.info("Simple Queue消息： "+map);System.out.println("sout=============:"+map.toString());
}
w HashMap<>(2);map.put("name", "xiaoming");map.put("age", 18);
//        String message = "测试消息！！！";//发送消息rabbitTemplate.convertAndSend(queueName,map);}

consumer

@RabbitListener(queues = "simpleQueue")
public void listenSimpleQueue(Map<String, Object> map) throws InterruptedException {log.info("Simple Queue消息： "+map);System.out.println("sout=============:"+map.toString());
}