1.概念
- 消息发送者(publisher)：生产消息
- 交换机(exchange)：负责路由消息，把消息路由给队列，可以路由给一个队列，也可以路由给多个队列，这取决于交换机的类型
- 队列(queue)：队列，存储消息
- 消息消费者(coumsmser)：消费消息
- 虚拟主机(virtual-host)：虚拟主机，数据隔离作用
2.数据隔离
在实际工作中，公司一般是在一个指定的服务器上去搭建mq，或者多个机器上去搭建集群模式，那一个公司肯定不止一个项目组，多个项目组的情况下，不可能每个项目都搞一套自己的mq，费时费力不说，维护还麻烦，所以mq就有数据隔离，多个项目组用一个环境的mq，数据不一样而已
3.使用控制台向mq传递消息
1.创建两个队列-“测试队列”，“测试队列2”
2.创建一个交换机-“测试交换机”
3.测试发送消息
3.1让交换机和队列进行绑定
绑定成功之后在指定的"测试队列"中也可以看到他和交换机的绑定关系
3.2发送消息
3.3查看消息
当然你也可以使用这个交换机同时绑定创建的两个队列
4.创建虚拟主机
5.java使用rabbitmq
5.1 发送消息
接着之前的在common里面引入依赖（没看之前的文章的直接就创建一个单体的springboot项目引入这个依赖就行）
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
在用户工程作为消息投递方，订单工程作为消费者，不通过交换机投递消息，并且消费
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
userController
@Autowired
RabbitTemplate rabbitTemplate;
@GetMapping("/sendMassage")
@ApiOperation(value = "不通过交换机发送消息")
public void sendMassage( String queueName ,String msg ){
rabbitTemplate.convertAndSend(queueName,msg);
}
接口测试
查看消息
5.2 消费消息
order工程加配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
创建orderListen
@Component
public class orderListen {
@RabbitListener(queues = "测试队列2")
public void listenOrder(String msg){
System.out.println("我已经接收到订单消息："+msg);
}
}
6.任务模型work queues
简单的说就是多个消费者绑定一个队列
- 创建一个队列work.queue
- 生产者(用户服务)向队列(work.queue)中发送消息，每秒钟100条记录
- 创建两个消费者(订单服务)监听队列，一个消费者一秒钟消费20条，一个消费者一秒钟消费30条记录
生产者代码
@GetMapping("/sendWorkQueueMassage")
@ApiOperation(value = "发送到任务模型")
public void sendWorkQueueMassage() throws InterruptedException {
String queueName="work.queue";
for (int i = 1; i <=100 ; i++) {
String msg="msg_"+i;
rabbitTemplate.convertAndSend(queueName,msg);
//休眠20毫秒
Thread.sleep(20);
}
}
消费者代码
@RabbitListener(queues = "work.queue")
public void listenWorkQueueOrder(String msg) throws InterruptedException {
System.out.println("消费者1已经接收到订单消息："+msg);
// Thread.sleep(30);
}
@RabbitListener(queues = "work.queue")
public void listenWorkQueueOrder2(String msg) throws InterruptedException {
System.err.println("消费者2已经接收到订单消息："+msg);
// Thread.sleep(40);
}
结果：
1.队列在被多个消费者绑定的时候，队列会把消息轮询分配给每一个消费者
2.消息被消费方消费之后就消失了
产生的问题：
问题1.资源浪费：现实生活中，每个服务器的负载能力都是不一样的，假如B服务器一秒钟只能处理2个请求，A服务器一秒钟能处理20个，那在轮询消费的时候，假设时间过去0.3秒，B服务器还没消费完一个消息，按照A服务器的性能，他0.3秒都可以处理好几个了，他应该在0.05秒的时候就处理完毕一个了，但是由于轮询他只能处理一个，这个时候A就要等着B消费完，这样就很浪费A的服务器资源。
2.消息积压，以上代码，生产方发送消息到队列，休眠时间为20毫秒，消费者1消费一个消息要30毫秒,B需要40毫秒，时间长了。生产者发的消息消费者就消费不过来
问题1处理方案:
增加配置
spring:
rabbitmq:
listener:
direct:
prefetch: 1 #保证同一时刻最多投递一条消息给消费者
结果，因为消费者1的消费能力比消费者2要快，所有可以看到他没有等着
问题2处理方案:
很明显能看到，两个消费者的消费能力跟不上生产者的生产速度，所有只能再增加多个消费者，直到消费者的消费能力快过生产者的生产能力
7.交换机
7.1.为什么使用交换机
我们上面的代码，是生产者直接连接队列，然后消费者消费，实际业务中，你在网购平台买东西，购买成功你的订单微服务得知道，积分微服务得知道，购物车微服务得知道，如果按照不用交换机去做，那消息一旦被订单服务消费了，这条消息在队列认为就消费完毕了，直接就会删除，造成的结果就是积分微服务就不知道了。那咋搞，所以就可以用到交换机
7.2.交换机模型
7.2.1交换机模型Fanout(广播)
把消息放到交换机，然后交换机广播给多个队列(积分队列，购物车队列，订单队列)，然后相应得微服务去跟相应得队列绑定，这种方式叫做广播
7.2.1.1改造java代码
- 使用之前创建的交换机，测试交换机，并且绑定"测试队列","测试队列2"两个队列
- 编写两个消费者方法，分别监听两个队列
创建积分微服务
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
listener:
direct:
prefetch: 1 #保证同一时刻最多投递一条消息给消费者
@Component
public class PointsFanoutListen {
@RabbitListener(queues = "测试队列2")
public void listenPoints(String msg){
System.out.println("积分服务已经接收到消息："+msg);
}
}
订单微服务微服务中监听另外一个队列
@Component
public class OrderFanoutListen {
@RabbitListener(queues = "测试队列")
public void listenOrder(String msg){
System.out.println("订单服务已经接收到消息："+msg);
}
}
- 编写生产者方法，向交换机发送消息
@GetMapping("/sendFanoutMassage")
@ApiOperation(value = "发送消息到广播交换机")
public void sendFanoutMassage() throws InterruptedException {
String exchangeName="测试交换机";
String msg="用户成功下单了";
rabbitTemplate.convertAndSend(exchangeName,null,msg);
}
测试
本地调用接口: loclahost:8001/user/sendFanoutMassage
启动两个消费者
7.2.2交换机模型Direct(订阅)
实际业务中，我可能不需要把消息发送给每个队列，比如。我订单交易失败，我的积分微服务就不需要接收到这种，积分微服务只有在交易成功才做积分减少或者增加的操作，那就是我只订阅交易成功的订单消息
7.2.2.1
创建交换机
创建队列
交换机跟队列绑定
创建消费者
消费者1：订单服务监听队列1
@Component
public class PointsDirectListen {
@RabbitListener(queues = "driect.queue2")
public void listenPoints(String msg){
System.out.println("积分服务已经接收到用户成功下单消息："+msg);
}
}
消费者2：积分服务监听队列2
@Component
public class PointsDirectListen {
@RabbitListener(queues = "driect.queue2")
public void listenPoints(String msg){
System.out.println("积分服务已经接收到用户成功下单消息："+msg);
}
}
创建生产者用户服务
@GetMapping("/sendDirectMassage")
@ApiOperation(value = "发送消息到订阅交换机")
public void sendDirectMassage() throws InterruptedException {
String exchangeName="work.dirice";
String msg="用户成功下单了";
rabbitTemplate.convertAndSend(exchangeName,"red",msg);
}
@GetMapping("/sendDirectMassageFaild")
@ApiOperation(value = "发送消息到订阅交换机")
public void sendDirectMassageFaild() throws InterruptedException {
String exchangeName="work.dirice";
String msg="用户下单失败了";
rabbitTemplate.convertAndSend(exchangeName,"blue",msg);
}
@GetMapping("/sendDirectMassageWait")
@ApiOperation(value = "发送消息到订阅交换机")
public void sendDirectMassageWait() throws InterruptedException {
String exchangeName="work.dirice";
String msg="用户下单但是还未付款";
rabbitTemplate.convertAndSend(exchangeName,"yellow",msg);
}
分别调用三个接口.结果如下
sendDirectMassage接口两个消费者都能接收到
sendDirectMassageFaild只有消费者1能接收到
sendDirectMassageWait只有消费者2能接收到
7.2.3交换机模型Topic()
编写案例
创建绑定关系
7.3.队列和交换机的申明
在之前我们都是手动在控制台去创建队列或者交换机，但是在真实企业中，不可能手动在控制台去创建，而且这样创建的，一旦中间件出问题了，所有的队列和交换机就没了，一般是用代码处理。
/**
* 注解的方式创建队列
* 一般在消费方创建
* 1.创建一个名字叫annotate.work的且类型为TOPIC的交换机
* 2.交换机绑定的队列为annotate.queue，该队列持久化
* 3.交换机绑定的key为"red","yellow"
* @param msg
* @throws InterruptedException
*/
@Component
public class orderListen {
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name ="annotate.queue",declare = "true"), //队列名称叫annotate.queue，且需要持久化
exchange = @Exchange(name = "annotate.work",type = ExchangeTypes.TOPIC),//交换机名称和类型
key={"red","yellow"} //路由key
))
public void listenWorkAnnotateQueueOrder2(String msg) throws InterruptedException {
System.err.println("注解方式生成的队列收到消息："+msg);
Thread.sleep(50);
}
}
项目启动之后就能直接创建相应的队列和交换机
8.消息转换器
1.创建一个队列，名字叫object.queue
2.创建生产者往这个队列发送一个消息，消息的类型为map或者java对象
import com.threesum.OderApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.HashMap;
@SpringBootTest(classes = OderApplication.class)
public class orderTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void sendObjectMsg(){
HashMap<String, Object> msg = new HashMap<>();
msg.put("name","aa");
msg.put("age",21);
rabbitTemplate.convertAndSend("object.queue",msg);
}
}
3.观察队列中的消息
结论：会发现变成了一堆乱码(因为默认采用的是java的jdk序列化)
4.采用java序列化方式处理问题
4.1引入依赖
<dependency>
<groupId>com.fasterxml.jackson.dataformat</groupId>
<artifactId>jackson-dataformat-yaml</artifactId>
</dependency>
4.2发送方和消费方都使用java序列化
package com.threesum.config.rabbit;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
@Component
public class JacksonDada {
@Bean
public MessageConverter JacksonJsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
}
4.3再次获取，就转换正常了
