一、同步调用与异步调用
1、同步调用
同步调用长处:
时效性强,可以立即得到结果
同步调用存在的题目
(1)耦合度高:每次加入新的需求,都要修改原来的代码
(2)性能下降:调用者必要期待服务提供者相应,如果调用链过长则相应时间即是每次调用的时间之和
(3)资源浪费:调用链中的每个服务在期待相应过长中,不能开释请求占用的资源,高并发场景下会极度浪费体系资源
(4)级联失败:如果服务提供者出现题目,所有调用方都会跟着出题目,犹如多米诺骨牌一样,迅速导致整个微服务群故障
2、异步调用
长处:
缺点:
- 依赖于Broker的可靠性、安全性、吞吐能力
- 架构复杂,业务没有明显的流程线,欠好追踪管理
二、初识MQ
MQ(MessageQueue),中文是消息队列,字面意思就是存放消息的队列。也就是变乱驱动架构中的Broker。
三、RabbitMQ
docker安装RabbitMQ步调
1、拉取镜像
2、安装RabbitMQ,实行下令
- docker run -d -p 15672:15672 -p 5672:5672 \
- -e RABBITMQ_DEFAULT_USER=admin \
- -e RABBITMQ_DEFAULT_PASS=Miss5016 \
- --hostname mq1 \
- --name my-rabbitmq\
- rabbitmq:latest
复制代码 3、开放防火墙和安全组对应的端口
- # 开放端口防火墙
- firewall-cmd --permanent --zone=public --add-port=15672/tcp
- firewall-cmd --permanent --zone=public --add-port=5672/tcp
- # 重启防火墙
- systemctl restart firewalld.service
复制代码 4、启用RabbitMQ的web端管理界面
- # 进入容器内部
- docker exec -it my-rabbitmq /bin/bash
- # 允许使用web端管理
- rabbitmq-plugins enable rabbitmq_management
复制代码 5、通过浏览器访问,ip+端标语
6、输入docker run下令时的账号暗码即可进入访问
7、题目
通道被禁用 Stats in management UI are disabled on this node
解决方法:
(1)进入容器内部
- docker exec -it my-rabbitmq /bin/bash
复制代码 (2)切换到RabbitMQ的设置目录下
(3)修改设置
- echo management_agent.disable_metrics_collector = false > management_agent.disable_metrics_collector.conf
复制代码 (4)退出并重启docker容器
- # 从容器内部退出
- exit
- # 重启容器
- docker restart my-rabbitmq
复制代码 (5)重新访问通道
RabbitMQ的结构和概念
四、RabbitMQ的消息模子
官网文档:https://www.rabbitmq.com/tutorials
1、根本消息队列(BasicQueue)
总共分为三个角色:
- publisher:消息发布者,将消息发送到队列queue
- queue:消息队列,负责接受消息并缓存消息
- consumer:订阅队列,处置惩罚队列中的消息
示例,通过Springboot的测试类实现
(1)导入依赖
- <dependency>
- <groupId>com.rabbitmq</groupId>
- <artifactId>amqp-client</artifactId>
- <version>5.20.0</version>
- </dependency>
复制代码 (2)生产者代码
- @Test
- void contextLoads() {
- try {
- //1、建立连接
- ConnectionFactory factory = new ConnectionFactory();
- //2、设置连接参数
- factory.setHost("ip地址"); //主机名
- factory.setPort(5672); //连接端口
- factory.setVirtualHost("/"); //虚拟主机
- factory.setUsername("用户名"); //用户名
- factory.setPassword("密码"); //密码
- //3、建立连接
- Connection connection = factory.newConnection();
- //4、创建通道
- Channel channel = connection.createChannel();
- //5、创建队列并声明队列
- String queueName = "simple.queue"; //队列名称
- channel.queueDeclare(queueName,false, false, false, null);
- //6、发送消息
- String message = "hello rabbitMQ";
- channel.basicPublish("", queueName, null, message.getBytes());
- System.out.println("消息发送完毕。。。");
- //关闭管道和连接
- channel.close();
- connection.close();
- }catch (Exception e) {
- System.out.println("生产者发送消息失败 ----- > " + e.getMessage());
- }
- }
复制代码 (3)消耗者代码
- @Test
- void contextLoads() {
- try {
- //1、建立连接
- ConnectionFactory factory = new ConnectionFactory();
- //2、设置连接参数
- factory.setHost("ip地址"); //主机名
- factory.setPort(5672); //连接端口
- factory.setVirtualHost("/"); //虚拟主机
- factory.setUsername("用户名"); //用户名
- factory.setPassword("密码"); //密码
- //3、建立连接
- Connection connection = factory.newConnection();
- //4、创建通道
- Channel channel = connection.createChannel();
- //5、创建队列并声明队列
- String queueName = "simple.queue"; //队列名称
- channel.queueDeclare(queueName,false, false, false, null);
- //6、订阅消息
- channel.basicConsume(queueName, true, new DefaultConsumer(channel){
- @Override
- public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- //将字节流内容转为消息本体
- String message = new String(body);
- System.out.println("消费者接收到消息 --------- > " + message);
- }
- });
- System.out.println("等待消息中。。。。");
- //关闭管道和连接
- channel.close();
- connection.close();
- }catch (Exception e) {
- System.out.println("生产者发送消息失败 ----- > " + e.getMessage());
- }
- }
复制代码 根本消息队列的消息发送流程:
- 建立connection
- 创建channel
- 利用channel声明队列
- 利用channel向队列发送消息
根本消息队列的消息接收流程
- 建立connection
- 创建channel
- 利用channel声明队列
- 界说consumer的消耗行为handleDelivery()
- 利用channel将消耗者与队列绑定
2、工作消息队列(WorkQueue)
3、发布订阅(Publish、Subscribe)模式
根据交换机的类型,可分为不同的三种
(1)广播(Fanout Exchange)
(2)路由(Direct Exchange)
(3)主题(Topic Exchange)
五、SpringAMQP
1、先容
AMQP:Advanced Message Queueing Protocol,是用于在应用步伐或之间传递业务消息的开放标准,该协议与语言平台无关,更符合微服务中独立性的要求。
SpringAMQP:Spring AMQP是基于AMQP协议界说的一套API规范,提供了模板来发送和接收消息,包括两部分,其中spring-amqp是基础抽象,spring-rabbit是底层默认实现。
2、示例
简单消息队列
发送消息:
(1)在父工程中引入依赖,确保每个子工程都能利用
- <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-amqp</artifactId>
- </dependency>
复制代码 (2)编写设置类
- spring:
- rabbitmq:
- host: # 消息队列ip
- port: # 端口
- virtual-host: # 虚拟地址
- username: # 用户名
- password: # 密码
复制代码 (3)测试类
- @SpringBootTest
- class OrderServiceApplicationTests {
- @Autowired
- private RabbitTemplate rabbitTemplate;
- /**
- * 发送消息
- */
- @Test
- void sendMessageMQ() {
- //队列名称
- String queueName = "simple.queue";
- //消息
- String message = "你好,rabbitmq";
- //发送消息
- rabbitTemplate.convertAndSend(queueName, message);
- }
- }
复制代码 接收消息:
(1)引入依赖
- <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-amqp -->
- <dependency>
- <groupId>org.springframework.boot</groupId>
- <artifactId>spring-boot-starter-amqp</artifactId>
- </dependency>
复制代码 (2)编写设置类
- spring:
- rabbitmq:
- host: # 消息队列ip
- port: # 端口
- virtual-host: # 虚拟地址
- username: # 用户名
- password: # 密码
复制代码 (3)接收并处置惩罚消息
- @Component //监听消息的监听器
- public class SpringRabbitListener {
- //需要监听哪个消息队列中的消息
- @RabbitListener(queues = "simple.queue")
- public void listenerSimpleQueue(String message) {
- //处理消息
- System.out.println("监听到 simple.queue 队列中的消息 --------- > " + message);
- }
- }
复制代码 留意:消息一旦消耗就会从队列中移除,RabbitMQ没有消息回溯功能
Work Queue 工作队列
提高消息处置惩罚的速率,避免消息堆积
示例:
生产者,生产100条消息
- @Test
- void sendMessageWorkQueue() throws InterruptedException {
- String queueName = "simple.queue";
- String message = "你好,rabbitmq__";
- for (int i = 1; i <= 100; i++) {
- rabbitTemplate.convertAndSend(queueName, message+i);
- Thread.sleep(20);
- }
- }
复制代码 两个消耗者,共同消耗,设置不同的处置惩罚速率
- @Component
- public class SpringRabbitListener {
- @RabbitListener(queues = "simple.queue")
- public void listenerWork1Queue(String message) throws InterruptedException {
- System.out.println("消费者1消费了消息 --------- > " + message + " " + LocalDateTime.now());
- Thread.sleep(20);
- }
- @RabbitListener(queues = "simple.queue")
- public void listenerWork2Queue(String message) throws InterruptedException {
- System.err.println("消费者2......消费了消息 --------- > " + message + " " + LocalDateTime.now());
- Thread.sleep(200);
- }
- }
复制代码 结果:纵然设置了不同的处置惩罚速率,但是两个消耗者仍然处置惩罚了雷同的消息数量,并没有由于处置惩罚得快的消耗者会处置惩罚更多消息
原因:由于rabbitmq中的一个消息预取机制造成的
消息预取:当有大量消息到达消息队列时,队列会举行消息投递,消耗者的通道(channel)会预先把消息取出,就会均匀分配队列中的消息,并不会由于消耗者的处置惩罚能力举行分配
控制方法:
修改设置文件,设置值,可以控制预取消息的上限
- spring:
- rabbitmq:
- host: # 消息队列ip
- port: # 端口
- virtual-host: # 虚拟地址
- username: # 用户名
- password: # 密码
- listener: simple: prefetch: 1 # 每次只能取1条消息,处置惩罚完后再取
复制代码 发布(Publish)、订阅(Subscribe)
发布订阅模式答应将同一消息发送给多个消耗者。实现方式是加入了exchange(交换机)。
留意:exchange(交换机)负责消息路由,而不是存储,路由失败则消息丢失
发布订阅-Fanout Exchange(广播)
Fanout Exchange会将接收到的消息路由到每一个跟其绑定的queue(队列)
示例:
(1)在消耗者中声明,将交换机与队罗列行绑定
声明两个队列和一个交换机,举行绑定
- @Configuration
- public class FanoutConfig {
- /**
- * 声明交换机
- * @return
- */
- @Bean
- public FanoutExchange fanoutExchange() {
- //返回一个声明的交换机
- return new FanoutExchange("fanout.exchange");
- }
- /**
- * 声明第一个队列
- * @return
- */
- @Bean
- public Queue fanoutQueue1() {
- return new Queue("fanout.queue1");
- }
- /**
- * 声明第二个队列
- * @return
- */
- @Bean
- public Queue fanoutQueue2() {
- return new Queue("fanout.queue2");
- }
- /**
- * 绑定队列到交换机
- * @param fanoutQueue1
- * @param fanoutExchange
- * @return
- */
- @Bean
- public Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
- //将队列1绑定到交换机
- return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
- }
- /**
- * 绑定队列到交换机
- * @param fanoutQueue2
- * @param fanoutExchange
- * @return
- */
- @Bean
- public Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
- //将队列2绑定到交换机
- return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
- }
- }
复制代码 (2)处置惩罚消息
- @Component
- public class SpringRabbitListener {
- @RabbitListener(queues = "fanout.queue1")
- public void listenerFanoutQueue1(String message) {
- System.out.println("监听到 fanout.queue1 队列中的消息 --------- > " + message);
- }
- @RabbitListener(queues = "fanout.queue2")
- public void listenerFanoutQueue2(String message) {
- System.out.println("监听到 fanout.queue2 队列中的消息 --------- > " + message);
- }
- }
复制代码 (3)发送消息
- /**
- * 发送消息到交换机
- */
- @Test
- void testSendFanoutExchange() {
- String exChangeName = "fanout.exchange";
- String message = "交换机转发的消息";
- rabbitTemplate.convertAndSend(exChangeName, "", message);
- }
复制代码 这样就实现了每个队列都能消耗消息,而不是消息一旦被其中一个消耗者消耗后,就会移除的题目
发布订阅-DirectExchange(路由模式)
Direct Exchange会将接收到的消息根据规则路由到指定的Queue,因此称为路由模式(routes)。
- 每一个Queue都会与Exchange设置一个BindingKey
- 发布者发送消息时,指定消息的RoutingKey
- Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
示例:
(1)声明队列,交换机,通过注解的方式
- @Component
- public class SpringDirectListener {
- /**
- * 声明第一个消费者
- * @param message
- */
- @RabbitListener(bindings = @QueueBinding(
- value = @Queue(name = "direct.queue1"),
- exchange = @Exchange(name = "direct.exchange", type = ExchangeTypes.DIRECT),
- key = {"red", "blue"}
- ))
- public void listenerDirect1Queue(String message) {
- System.out.println("direct消费者1监听到的消息 ---------- > " + message);
- }
- /**
- * 声明第二个消费者
- * @param message
- */
- @RabbitListener(bindings = @QueueBinding(
- value = @Queue(name = "direct.queue2"),
- exchange = @Exchange(name = "direct.exchange", type = ExchangeTypes.DIRECT),
- key = {"red", "yellow"}
- ))
- public void listenerDirect2Queue(String message) {
- System.out.println("direct消费者2监听到的消息 ---------- > " + message);
- }
- }
复制代码 (2)发消息
- /**
- * 发送消息到交换机
- */
- @Test
- void testSendDirectExchange() {
- String exChangeName = "direct.exchange";
- String message = "direct交换机转发的消息";
- String routingKey = "red"; //发送给指定routingKey的队列
- rabbitTemplate.convertAndSend(exChangeName, routingKey, message);
- }
复制代码 发布订阅-TopicExchange(主题模式)
TopicExchange与DirectExchange类似,区别在于routingKey必须是多个单词的列表,并且用 . 分割
示例:
(1)声明交换机,队列
- @Component
- public class SpringTopicListener {
- @RabbitListener(bindings = @QueueBinding(
- value = @Queue("topic.queue1"),
- exchange = @Exchange(value = "topic.exchange", type = ExchangeTypes.TOPIC),
- key = {"china.#"}
- ))
- public void listenerTopic1Queue(String message) {
- System.out.println("Topic消费者1监听到的消息 ---------- > " + message);
- }
- @RabbitListener(bindings = @QueueBinding(
- value = @Queue("topic.queue2"),
- exchange = @Exchange(value = "topic.exchange", type = ExchangeTypes.TOPIC),
- key = {"#.news"}
- ))
- public void listenerTopic2Queue(String message) {
- System.out.println("Topic消费者2监听到的消息 ---------- > " + message);
- }
- }
复制代码 (2)测试消息发送
- /**
- * 发送消息到交换机
- */
- @Test
- void testSendTopicExchange() {
- String exChangeName = "topic.exchange";
- String message = "topic交换机转发的消息";
- String routingKey = "11.news";
- rabbitTemplate.convertAndSend(exChangeName, routingKey, message);
- }
复制代码 六、消息转换器
测试:
(1)新增一个队列(queue)
- /**
- * 声明一个队列
- * @return
- */
- @Bean
- public Queue objectQueue() {
- return new Queue("object.queue");
- }
复制代码 (2)测试发送对象类型数据
- /**
- * 发送消息
- */
- @Test
- void sendObjectMessageMQ() {
- String queueName = "object.queue";
- Map<String, Object> map = new HashMap<>();
- map.put("username", "张三");
- map.put("age", 12);
- rabbitTemplate.convertAndSend(queueName, map);
- }
复制代码 结果:
原因:
rabbitmq只支持字节,而spring答应发object对象,说明spring会将对象做序列化,用的序列化方式是jdk的序列化方式x-java-serialized-object,这种序列化缺点是性能差,安全性有题目,可能存在注入风险,数据长度太长了
Spring的对象处置惩罚事由org.springframework.amqp.support.converter.MessageConverter来处置惩罚的,而默认实现的SimpleMessageConverter,基于JDK的ObjectOutputStream完成序列化。如果要修改必要界说一个MessageConverter类型的Bean即可。
步调:
(1)导入依赖
- <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
- <dependency>
- <groupId>com.fasterxml.jackson.dataformat</groupId>
- <artifactId>jackson-dataformat-xml</artifactId>
- <version>2.17.1</version>
- </dependency>
复制代码 (2)声明一个bean,将消息转换的bean注入spring容器中
- /**
- * 声明消息转换类型
- * @return
- */
- @Bean
- public MessageConverter messageConverter() {
- return new Jackson2JsonMessageConverter();
- }
复制代码 (3)将消息清除,重新测试
消息接收:
(1)引入依赖
- <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-xml -->
- <dependency>
- <groupId>com.fasterxml.jackson.dataformat</groupId>
- <artifactId>jackson-dataformat-xml</artifactId>
- <version>2.17.1</version>
- </dependency>
复制代码 (2)注入bean
- /**
- * 声明消息转换类型
- * @return
- */
- @Bean
- public MessageConverter messageConverter() {
- return new Jackson2JsonMessageConverter();
- }
复制代码 (3)消息接收处置惩罚
- @RabbitListener(queues = "object.queue")
- public void listenerObjectQueue(Map<String, Object> map) {
- System.out.println("监听到 object.queue 队列中的消息 --------- > " + map);
- }
复制代码 免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |