网站建设零基础教学拟采用建站技术

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: localhostport: 5672username: guestpassword: guest

userController

 @AutowiredRabbitTemplate rabbitTemplate;@GetMapping("/sendMassage")@ApiOperation(value = "不通过交换机发送消息")public void sendMassage( String queueName ,String msg ){rabbitTemplate.convertAndSend(queueName,msg);}

接口测试

查看消息

5.2 消费消息

order工程加配置

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest

创建orderListen

@Component
public class orderListen {@RabbitListener(queues = "测试队列2")public void listenOrder(String msg){System.out.println("我已经接收到订单消息："+msg);}
}

6.任务模型work queues

简单的说就是多个消费者绑定一个队列

1. 创建一个队列work.queue
2. 生产者(用户服务)向队列(work.queue)中发送消息，每秒钟100条记录
3. 创建两个消费者(订单服务)监听队列，一个消费者一秒钟消费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代码

1. 使用之前创建的交换机，测试交换机，并且绑定"测试队列","测试队列2"两个队列
   
2. 编写两个消费者方法，分别监听两个队列
   创建积分微服务

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestlistener: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);}}

4. 编写生产者方法，向交换机发送消息

 @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. 创建交换机
   

2. 创建队列
   

3. 交换机跟队列绑定
   

4. 创建消费者
   消费者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*/@Componentpublic 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 {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic 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 {@Beanpublic MessageConverter JacksonJsonMessageConverter(){return new Jackson2JsonMessageConverter();}
}

4.3再次获取，就转换正常了