网站换代理,网站建站公司哪家价钱合理,网站建设过程中的系统结构图,国内免费建站网站目录
1.初识MQ
1.1.同步调用
1.2.异步调用
1.3.技术选型
2.RabbitMQ
2.1.安装
2.2.收发消息
2.2.1.交换机
2.2.2.队列
2.2.3.绑定关系
2.2.4.发送消息
2.3.数据隔离
2.3.1.用户管理
2.3.2.virtual host
3.SpringAMQP
3.1.导入Demo工程
3.2.快速入门
3.2.1.消…目录
1.初识MQ
1.1.同步调用
1.2.异步调用
1.3.技术选型
2.RabbitMQ
2.1.安装
2.2.收发消息
2.2.1.交换机
2.2.2.队列
2.2.3.绑定关系
2.2.4.发送消息
2.3.数据隔离
2.3.1.用户管理
2.3.2.virtual host
3.SpringAMQP
3.1.导入Demo工程
3.2.快速入门
3.2.1.消息发送
3.2.2.消息接收
3.2.3.测试
3.3.WorkQueues模型
3.3.1.消息发送
3.3.2.消息接收
3.3.3.测试
3.3.4.能者多劳
3.3.5.总结
3.4.交换机类型
3.5.Fanout交换机
3.5.1.声明队列和交换机
3.5.2.消息发送
3.5.3.消息接收
3.6.Direct交换机
3.6.1.声明队列和交换机
3.6.2.消息接收
3.6.3.消息发送
3.6.4.总结
3.7.Topic交换机
3.7.1.说明
3.7.2.消息发送
3.7.3.消息接收
3.7.4.总结
3.8.Java代码声明交换机和队列
3.8.1.基本API
3.8.2.fanout示例
3.8.2.direct示例
3.8.4.基于注解声明
3.9.消息转换器
3.9.1.测试默认转换器
3.9.2.配置JSON转换器
3.9.3.消费者接收Object
4.业务改造
4.1.配置MQ
4.1.接收消息
4.2.发送消息 微服务一旦拆分必然涉及到服务之间的相互调用目前我们服务之间的调用采用的是基于OpenFeign的调用。这种调用中调用者发起请求后需要等待服务提供者执行业务后返回结果才能继续执行后面的业务。也就是说调用者在调用过程中处于阻塞状态因此我们称这种调用方式为同步调用也可以叫同步通讯。但在很多场景下我们可能需要采用异步通讯的方式为什么呢 我们先来看看什么是同步通讯和异步通讯。如图 解读 同步通讯就如同打视频电话双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。 异步通讯就如同发微信聊天双方的交互不是实时的你不需要立刻给对方回应。因此你可以多线操作同时跟多人聊天。 两种方式各有优劣打电话可以立即得到响应但是你却不能跟多个人同时通话。发微信可以同时与多个人收发微信但是往往响应会有延迟。 所以如果我们的业务需要实时得到服务提供方的响应则应该选择同步通讯同步调用。而如果我们追求更高的效率并且不需要实时响应则应该选择异步通讯异步调用。
同步调用的方式我们已经学过了之前的OpenFeign调用就是。但是 异步调用又该如何实现 哪些业务适合用异步调用来实现呢
1.初识MQ
1.1.同步调用
基于OpenFeign的调用都属于是同步调用那么这种方式存在哪些问题呢
举个例子以昨天余额支付功能为例来分析首先看下整个流程 目前采用的是基于OpenFeign的同步调用也就是业务的执行流程是这样的 支付服务需要先调用用户服务扣减余额 然后支付服务自己要更新支付流水单的状态 然后支付服务调用交易服务更新业务订单状态为已支付
三个步骤依次执行。
这其中就存在3个问题
第一拓展性差
我们目前的业务相对简单但是随着业务规模扩大产品的功能也在不断完善。
在大多数电商业务中用户支付成功后都会以短信或者其它方式通知用户告知支付成功。假如后期产品经理提出这样新的需求你怎么办是不是要在上述业务中再加入通知用户的业务
某些电商项目中还会有积分或金币的概念。假如产品经理提出需求用户支付成功后给用户以积分奖励或者返还金币你怎么办是不是要在上述业务中再加入积分业务、返还金币业务
。。。
最终你的支付业务会越来越臃肿 也就是说每次有新的需求现有支付逻辑都要跟着变化代码经常变动不符合开闭原则拓展性不好。 第二性能下降
由于我们采用了同步调用调用者需要等待服务提供者执行完返回结果后才能继续向下执行也就是说每次远程调用调用者都是阻塞等待状态。最终整个业务的响应时长就是每次远程调用的执行时长之和 假如每个微服务的执行时长都是50ms则最终整个业务的耗时可能高达300ms性能太差了。
第三级联失败
由于我们是基于OpenFeign调用交易服务、通知服务。当交易服务、通知服务出现故障时整个事务都会回滚交易失败。
这其实就是同步调用的级联失败问题。 但是大家思考一下我们假设用户余额充足扣款已经成功此时我们应该确保支付流水单更新为已支付确保交易成功。毕竟收到手里的钱没道理再退回去吧。 因此这里不能因为短信通知、更新订单状态失败而回滚整个事务。 综上同步调用的方式存在下列问题 拓展性差 性能下降 级联失败
而要解决这些问题我们就必须用异步调用的方式来代替同步调用。
1.2.异步调用
异步调用方式其实就是基于消息通知的方式一般包含三个角色 消息发送者投递消息的人就是原来的调用方 消息Broker管理、暂存、转发消息你可以把它理解成微信服务器 消息接收者接收和处理消息的人就是原来的服务提供方 在异步调用中发送者不再直接同步调用接收者的业务接口而是发送一条消息到消息Broker然后接收者根据自己的需求从消息Broker中订阅消息。每当发送方发送消息后接受者都能获取消息并处理。
这样发送消息的人和接收消息的人就完全解耦了。
还是以余额支付业务为例 除了扣减余额、更新支付流水单状态以外其它调用逻辑全部取消。而是改为发送一条消息到Broker。而相关的微服务都可以订阅消息通知一旦消息到达Broker则会分发给每一个订阅了的微服务处理各自的业务。 假如产品经理提出了新的需求比如要在支付成功后更新用户积分。支付代码完全不用变更而仅仅是让积分服务也订阅消息即可 不管后期增加了多少消息订阅者作为支付服务来讲执行完扣减余额、更新支付流水状态后发送消息即可。业务耗时仅仅是这三部分业务耗时仅仅100ms大大提高了业务性能。 另外不管是交易服务、通知服务还是积分服务他们的业务与支付关联度低。现在采用了异步调用解除了耦合他们即便执行过程中出现了故障也不会影响到支付服务。 综上异步调用的优势包括 耦合度更低 性能更好 业务拓展性强 故障隔离避免级联失败 当然异步通信也并非完美无缺它存在下列缺点 完全依赖于Broker的可靠性、安全性和性能 架构复杂后期维护和调试麻烦
1.3.技术选型
消息Broker目前常见的实现方案就是消息队列MessageQueue简称为MQ.
目比较常见的MQ实现 ActiveMQ RabbitMQ RocketMQ Kafka
几种常见MQ的对比
RabbitMQActiveMQRocketMQKafka公司/社区RabbitApache阿里Apache开发语言ErlangJavaJavaScalaJava协议支持AMQPXMPPSMTPSTOMPOpenWire,STOMPREST,XMPP,AMQP自定义协议自定义协议可用性高一般高高单机吞吐量一般差高非常高消息延迟微秒级毫秒级毫秒级毫秒以内消息可靠性高一般高一般
追求可用性Kafka、 RocketMQ 、RabbitMQ
追求可靠性RabbitMQ、RocketMQ
追求吞吐能力RocketMQ、Kafka
追求消息低延迟RabbitMQ、Kafka 据统计目前国内消息队列使用最多的还是RabbitMQ再加上其各方面都比较均衡稳定性也好因此我们课堂上选择RabbitMQ来学习。
2.RabbitMQ
RabbitMQ是基于Erlang语言开发的开源消息通信中间件官网地址
https://www.rabbitmq.com/
接下来我们就学习它的基本概念和基础用法。
2.1.安装
同样基于Docker来安装RabbitMQ使用下面的命令即可
docker run \-e RABBITMQ_DEFAULT_USERitheima \-e RABBITMQ_DEFAULT_PASS123321 \-v mq-plugins:/plugins \--name mq \--hostname mq \-p 15672:15672 \-p 5672:5672 \--network hm-net\-d \rabbitmq:3.8-management
可以看到在安装命令中有两个映射的端口 15672RabbitMQ中管理控制台的端口 5672RabbitMQ中消息发送处理的端口
安装完成后我们访问 http://192.168.150.101:15672即可看到管理控制台。首次访问需要登录默认的用户名和密码在配置文件中已经指定了。
登录后即可看到管理控制台总览页面 RabbitMQ对应的架构如图 其中包含几个概念 publisher生产者发送消息的一方 consumer消费者消费消息的一方 queue队列存储消息。生产者发送的消息会暂存在消息队列中等待消费者处理 exchange交换机负责消息的路由。生产者发送的消息由交换机决定投递到哪个队列。 virtual host虚拟主机起到数据隔离的作用。每个虚拟主机相互独立有各自的exchange、queue
上述这些东西都可以在RabbitMQ的管理控制台来管理下一节我们就一起来学习控制台的使用。
2.2.收发消息
2.2.1.交换机
注意
交换机只能路由消息而无法存储消息队列必须与交换机绑定因为交换机只会路由消息给与其绑定的队列
我们打开Exchanges选项卡可以看到已经存在很多交换机 我们点击任意交换机即可进入交换机详情页面。仍然会利用控制台中的publish message 发送一条消息 这里通过控制台模拟生产者发送消息。由于没有消费者存在最终消息丢失了这样说明交换机没有存储消息的能力。
2.2.2.队列
我们打开Queues选项卡新建一个队列 命名为hello.queue1 再以相同的方式创建一个队列密码为hello.queue2最终队列列表如下 此时我们再次向amq.fanout交换机发送一条消息。会发现消息依然没有到达队列
怎么回事呢
发送到交换机的消息只会路由到与其绑定的队列因此仅仅创建队列是不够的我们还需要将其与交换机绑定。
2.2.3.绑定关系
点击Exchanges选项卡点击amq.fanout交换机进入交换机详情页然后点击Bindings菜单在表单中填写要绑定的队列名称 相同的方式将hello.queue2也绑定到改交换机。
最终绑定结果如下 2.2.4.发送消息
再次回到exchange页面找到刚刚绑定的amq.fanout点击进入详情页再次发送一条消息 回到Queues页面可以发现hello.queue中已经有一条消息了 点击队列名称进入详情页查看队列详情这次我们点击get message 可以看到消息到达队列了 这个时候如果有消费者监听了MQ的hello.queue1或hello.queue2队列自然就能接收到消息了。
2.3.数据隔离
为不同项目/业务创建不同用户,并分配不同的虚拟主机;各个虚拟主机之间是数据隔离的
2.3.1.用户管理
点击Admin选项卡首先会看到RabbitMQ控制台的用户管理界面 这里的用户都是RabbitMQ的管理或运维人员。目前只有安装RabbitMQ时添加的itheima这个用户。仔细观察用户表格中的字段如下 Nameitheima也就是用户名 Tagsadministrator说明itheima用户是超级管理员拥有所有权限 Can access virtual host /可以访问的virtual host这里的/是默认的virtual host
对于小型企业而言出于成本考虑我们通常只会搭建一套MQ集群公司内的多个不同项目同时使用。这个时候为了避免互相干扰 我们会利用virtual host的隔离特性将不同项目隔离。一般会做两件事情 给每个项目创建独立的运维账号将管理权限分离。 给每个项目创建不同的virtual host将每个项目的数据隔离。
比如我们给黑马商城创建一个新的用户命名为hmall 你会发现此时hmall用户没有任何virtual host的访问权限 2.3.2.virtual host
我们先退出登录 切换到刚刚创建的hmall用户登录然后点击Virtual Hosts菜单进入virtual host管理页 可以看到目前只有一个默认的virtual host名字为 /。
我们可以给黑马商城项目创建一个单独的virtual host而不是使用默认的/。 创建完成后如图 由于我们是登录hmall账户后创建的virtual host因此回到users菜单你会发现当前用户已经具备了对/hmall这个virtual host的访问权限了用哪个用户创建的虚拟主机哪个用户就对那个虚拟主机有访问权限 此时点击页面右上角的virtual host下拉菜单切换virtual host为 /hmall 然后再次查看queues选项卡会发现之前的队列已经看不到了 这就是基于virtual host 的隔离效果。
3.SpringAMQP 将来我们开发业务功能的时候肯定不会在控制台收发消息而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是RabbitMQ官方提供的Java客户端编码相对复杂一般生产环境下我们更多会结合Spring来使用Spring的官方基于RabbitMQ提供了一套消息收发的模板工具SpringAMQP。并且还基于SpringBoot对其实现了自动装配使用起来非常方便。
SpringAmqp的官方地址
Docshttps://b11et3un53m.feishu.cn/wiki/OQH4weMbcimUSLkIzD6cCpN0nvc#share-JoqFdiNIbolTRVxduGBcCWC7nuf
SpringAMQP提供了三个功能 自动声明队列、交换机及其绑定关系 基于注解的监听器模式异步接收消息 封装了RabbitTemplate工具用于发送消息
这一章我们就一起学习一下如何利用SpringAMQP实现对RabbitMQ的消息收发。
3.1.导入Demo工程
在课前资料给大家提供了一个Demo工程方便我们学习SpringAMQP的使用 将其复制到你的工作空间然后用Idea打开项目结构如图 包括三部分 mq-demo父工程管理项目依赖 publisher消息的发送者 consumer消息的消费者 其中publisher和consumer都是微服务publisher用于发送消息consumer用于接收消息从而演示跨微服务的消息收发
在mq-demo这个父工程中已经配置好了SpringAMQP相关的依赖因此子工程中可以直接使用SpringAMQP
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcn.itcast.demo/groupIdartifactIdmq-demo/artifactIdversion1.0-SNAPSHOT/versionmodulesmodulepublisher/modulemoduleconsumer/module/modulespackagingpom/packagingparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.12/versionrelativePath//parentpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.target/propertiesdependenciesdependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency!--AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!--单元测试--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependency/dependencies
/project
3.2.快速入门
SpringAMQP收发消息的流程总结
1、引入spring-boot-starter-amqp依赖
2、配置rabbitmq服务端信息
3、利用RabbitTemplate发送消息
4、利用RabbitListener注解声明要监听的队列监听消息 在之前的案例中都是经过交换机发送消息到队列不过为了测试方便我们可以直接向队列发送消息跳过交换机。
在入门案例中我们就演示这样的简单模型如图 也就是 publisher直接发送消息到队列 消费者监听并处理队列中的消息
注意这种模式一般测试使用很少在生产中使用。
为了方便测试我们现在控制台新建一个队列simple.queue 添加成功 接下来我们就可以利用Java代码收发消息了。
3.2.1.消息发送
首先配置MQ地址在publisher服务的application.yml中添加配置
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
然后在publisher服务中编写测试类SpringAmqpTest并利用RabbitTemplate实现消息发送
package com.itheima.publisher.amqp;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;SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSimpleQueue() {// 队列名称String queueName simple.queue;// 消息String message hello, spring amqp!;// 发送消息转换消息并将消息发送到队列//queueName是队列名要发送到哪个队列//message是消息要发送的消息rabbitTemplate.convertAndSend(queueName, message); }
}
打开控制台可以看到消息已经发送到队列中 接下来我们再来实现消息接收。
3.2.2.消息接收
先在consumer服务的application.yml中配置MQ的地址
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
然后在consumer服务的com.itheima.consumer.listener包中新建一个类SpringRabbitListener代码如下
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class SpringRabbitListener {// 利用RabbitListener注解标注在方法上声明要监听的队列表示消费者// 将来一旦监听的此队列中有消息了MQ就会推送给当前服务然后调用这个方法处理消息。// 此方法的形参就是用于接收消息的消息发的什么类型这里就用什么类型接收// 可以看到方法体中接收的就是消息体的内容RabbitListener(queues simple.queue)public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println(spring 消费者接收到消息【 msg 】);}
}3.2.3.测试
启动consumer服务然后在publisher服务中运行测试代码发送MQ消息。最终consumer收到消息 3.3.WorkQueues模型
Work queues任务模型。简单来说就是让多个消费者绑定到一个队列共同消费队列中的消息。 当消息处理比较耗时的时候可能生产消息的速度会远远大于消息的消费速度。长此以往消息就会堆积越来越多无法及时处理。
此时可以使用Work queues模型多个消费者共同处理消息处理消息处理的速度就能大大提高。
注意虽然多个消费者绑定到一个队列但同一条消息只会被一个消费者处理
接下来我们就来模拟这样的场景。
首先我们在控制台创建一个新的队列命名为work.queue 3.3.1.消息发送
这次我们循环发送模拟大量消息堆积现象。
在publisher服务中的SpringAmqpTest类中添加一个测试方法
/*** workQueue* 向队列中不停发送消息模拟消息堆积。*/
Test
public void testWorkQueue() throws InterruptedException {// 队列名称String queueName work.queue;// 消息String message hello, message_;for (int i 0; i 50; i) {// 发送消息每20毫秒发送一次相当于每秒发送50条消息rabbitTemplate.convertAndSend(queueName, message i);Thread.sleep(20);}
}
3.3.2.消息接收
要模拟多个消费者绑定同一个队列我们在consumer服务的SpringRabbitListener中添加2个新的方法
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class SpringRabbitListener {// 利用RabbitListener来声明要监听的队列信息// 将来一旦监听的队列中有了消息就会推送给当前服务调用当前方法处理消息。// 可以看到方法体中接收的就是消息体的内容RabbitListener(queues simple.queue)public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println(spring 消费者接收到消息【 msg 】);}RabbitListener(queues work.queue)public void listenWorkQueue1(String msg) throws InterruptedException {System.out.println(消费者1接收到消息【 msg 】 LocalTime.now());Thread.sleep(20);}RabbitListener(queues work.queue)public void listenWorkQueue2(String msg) throws InterruptedException {System.err.println(消费者2........接收到消息【 msg 】 LocalTime.now());Thread.sleep(200);}
}
注意到这两消费者都设置了Thead.sleep模拟任务耗时 消费者1 sleep了20毫秒相当于每秒钟处理50个消息 消费者2 sleep了200毫秒相当于每秒处理5个消息
3.3.3.测试
启动ConsumerApplication后执行publisher服务中编写的发送测试方法testWorkQueue。
最终结果如下
消费者1接收到消息【hello, message_0】21:06:00.869555300
消费者2........接收到消息【hello, message_1】21:06:00.884518
消费者1接收到消息【hello, message_2】21:06:00.907454400
消费者1接收到消息【hello, message_4】21:06:00.953332100
消费者1接收到消息【hello, message_6】21:06:00.997867300
消费者1接收到消息【hello, message_8】21:06:01.042178700
消费者2........接收到消息【hello, message_3】21:06:01.086478800
消费者1接收到消息【hello, message_10】21:06:01.087476600
消费者1接收到消息【hello, message_12】21:06:01.132578300
消费者1接收到消息【hello, message_14】21:06:01.175851200
消费者1接收到消息【hello, message_16】21:06:01.218533400
消费者1接收到消息【hello, message_18】21:06:01.261322900
消费者2........接收到消息【hello, message_5】21:06:01.287003700
消费者1接收到消息【hello, message_20】21:06:01.304412400
消费者1接收到消息【hello, message_22】21:06:01.349950100
消费者1接收到消息【hello, message_24】21:06:01.394533900
消费者1接收到消息【hello, message_26】21:06:01.439876500
消费者1接收到消息【hello, message_28】21:06:01.482937800
消费者2........接收到消息【hello, message_7】21:06:01.488977100
消费者1接收到消息【hello, message_30】21:06:01.526409300
消费者1接收到消息【hello, message_32】21:06:01.572148
消费者1接收到消息【hello, message_34】21:06:01.618264800
消费者1接收到消息【hello, message_36】21:06:01.660780600
消费者2........接收到消息【hello, message_9】21:06:01.689189300
消费者1接收到消息【hello, message_38】21:06:01.705261
消费者1接收到消息【hello, message_40】21:06:01.746927300
消费者1接收到消息【hello, message_42】21:06:01.789835
消费者1接收到消息【hello, message_44】21:06:01.834393100
消费者1接收到消息【hello, message_46】21:06:01.875312100
消费者2........接收到消息【hello, message_11】21:06:01.889969500
消费者1接收到消息【hello, message_48】21:06:01.920702500
消费者2........接收到消息【hello, message_13】21:06:02.090725900
消费者2........接收到消息【hello, message_15】21:06:02.293060600
消费者2........接收到消息【hello, message_17】21:06:02.493748
消费者2........接收到消息【hello, message_19】21:06:02.696635100
消费者2........接收到消息【hello, message_21】21:06:02.896809700
消费者2........接收到消息【hello, message_23】21:06:03.099533400
消费者2........接收到消息【hello, message_25】21:06:03.301446400
消费者2........接收到消息【hello, message_27】21:06:03.504999100
消费者2........接收到消息【hello, message_29】21:06:03.705702500
消费者2........接收到消息【hello, message_31】21:06:03.906601200
消费者2........接收到消息【hello, message_33】21:06:04.108118500
消费者2........接收到消息【hello, message_35】21:06:04.308945400
消费者2........接收到消息【hello, message_37】21:06:04.511547700
消费者2........接收到消息【hello, message_39】21:06:04.714038400
消费者2........接收到消息【hello, message_41】21:06:04.916192700
消费者2........接收到消息【hello, message_43】21:06:05.116286400
消费者2........接收到消息【hello, message_45】21:06:05.318055100
消费者2........接收到消息【hello, message_47】21:06:05.520656400
消费者2........接收到消息【hello, message_49】21:06:05.723106700
可以看到消费者1和消费者2每人都消费了25条消息 消费者1很快完成了自己的25条消息 消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者并没有考虑到消费者的处理能力。导致1个消费者空闲另一个消费者忙的不可开交。没有充分利用每一个消费者的能力最终消息处理的耗时远远超过了1秒。这样显然是有问题的。
默认情况下RabbitMQ采用轮询你处理一条我处理一条换着来)的方式将消息分配给绑定在队列上的每个消费者
3.3.4.能者多劳
在spring中有一个简单的配置可以解决这个问题。我们修改consumer服务的application.yml文件添加配置
通过设置prefetch来控制消费者预取的消息数量如果不设置就不管消费者有没有处理完消息都一直给它分配
spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息
再次测试发现结果如下
消费者1接收到消息【hello, message_0】21:12:51.659664200
消费者2........接收到消息【hello, message_1】21:12:51.680610
消费者1接收到消息【hello, message_2】21:12:51.703625
消费者1接收到消息【hello, message_3】21:12:51.724330100
消费者1接收到消息【hello, message_4】21:12:51.746651100
消费者1接收到消息【hello, message_5】21:12:51.768401400
消费者1接收到消息【hello, message_6】21:12:51.790511400
消费者1接收到消息【hello, message_7】21:12:51.812559800
消费者1接收到消息【hello, message_8】21:12:51.834500600
消费者1接收到消息【hello, message_9】21:12:51.857438800
消费者1接收到消息【hello, message_10】21:12:51.880379600
消费者2........接收到消息【hello, message_11】21:12:51.899327100
消费者1接收到消息【hello, message_12】21:12:51.922828400
消费者1接收到消息【hello, message_13】21:12:51.945617400
消费者1接收到消息【hello, message_14】21:12:51.968942500
消费者1接收到消息【hello, message_15】21:12:51.992215400
消费者1接收到消息【hello, message_16】21:12:52.013325600
消费者1接收到消息【hello, message_17】21:12:52.035687100
消费者1接收到消息【hello, message_18】21:12:52.058188
消费者1接收到消息【hello, message_19】21:12:52.081208400
消费者2........接收到消息【hello, message_20】21:12:52.103406200
消费者1接收到消息【hello, message_21】21:12:52.123827300
消费者1接收到消息【hello, message_22】21:12:52.146165100
消费者1接收到消息【hello, message_23】21:12:52.168828300
消费者1接收到消息【hello, message_24】21:12:52.191769500
消费者1接收到消息【hello, message_25】21:12:52.214839100
消费者1接收到消息【hello, message_26】21:12:52.238998700
消费者1接收到消息【hello, message_27】21:12:52.259772600
消费者1接收到消息【hello, message_28】21:12:52.284131800
消费者2........接收到消息【hello, message_29】21:12:52.306190600
消费者1接收到消息【hello, message_30】21:12:52.325315800
消费者1接收到消息【hello, message_31】21:12:52.347012500
消费者1接收到消息【hello, message_32】21:12:52.368508600
消费者1接收到消息【hello, message_33】21:12:52.391785100
消费者1接收到消息【hello, message_34】21:12:52.416383800
消费者1接收到消息【hello, message_35】21:12:52.439019
消费者1接收到消息【hello, message_36】21:12:52.461733900
消费者1接收到消息【hello, message_37】21:12:52.485990
消费者1接收到消息【hello, message_38】21:12:52.509219900
消费者2........接收到消息【hello, message_39】21:12:52.523683400
消费者1接收到消息【hello, message_40】21:12:52.547412100
消费者1接收到消息【hello, message_41】21:12:52.571191800
消费者1接收到消息【hello, message_42】21:12:52.593024600
消费者1接收到消息【hello, message_43】21:12:52.616731800
消费者1接收到消息【hello, message_44】21:12:52.640317
消费者1接收到消息【hello, message_45】21:12:52.663111100
消费者1接收到消息【hello, message_46】21:12:52.686727
消费者1接收到消息【hello, message_47】21:12:52.709266500
消费者2........接收到消息【hello, message_48】21:12:52.725884900
消费者1接收到消息【hello, message_49】21:12:52.746299900
可以发现由于消费者1处理速度较快所以处理了更多的消息消费者2处理速度较慢只处理了6条消息。而最终总的执行耗时也在1秒左右大大提升。
正所谓能者多劳这样充分利用了每一个消费者的处理能力可以有效避免消息积压问题。
3.3.5.总结
Work模型的使用 多个消费者绑定到一个队列同一条消息只会被一个消费者处理 通过设置prefetch来控制消费者预取的消息数量
3.4.交换机类型
在之前的测试案例中都没有交换机生产者是直接发送消息到队列。引入交换机后消息发送的模式会有很大变化 可以看到在订阅模型中多了一个exchange角色而且过程略有变化 Publisher生产者不再直接发送消息到队列中而是发给交换机 Exchange交换机一方面接收生产者发送的消息另一方面是处理消息例如将传递给某个特别队列、传递给所有队列、或将消息丢弃。如何操作取决于Exchange的类型。 Queue消息队列也与以前一样接收消息、缓存消息。不过队列一定要与交换机绑定。 Consumer消费者与以前一样订阅队列没有变化
Exchange交换机只负责转发消息不能存储消息因此如果没有队列与Exchange绑定或者没有符合路由规则的队列消息将会丢失
交换机的类型有四种 Fanout交换机广播将收到的消息 传给 所有与该交换机绑定的队列。我们最早在控制台使用的正是Fanout交换机 Direct交换机订阅会将收到的消息 根据RoutingKey 路由 到具有同样RoutingKey的Queue中 每一个Queue都设置了一个BindingKeyRoutingKey 发布者发送消息时需要指定消息的RoutingKey Exchange会将消息路由到BindingKey与RoutingKey一致的队列中 Topic交换机通配符订阅与Direct一样只不过RoutingKey可以使用通配符 Headers交换机头匹配基于MQ的消息头匹配用的较少。
我们讲解前面的三种交换机模式。
3.5.Fanout交换机
Fanout英文翻译是扇出我觉得在MQ中叫广播更合适。
在广播模式下消息发送流程是这样的 1 可以有多个队列 2 每个队列都要绑定到Exchange交换机 3 生产者发送的消息只能发送到交换机 4 交换机把消息发送给绑定过的所有队列 5 订阅队列的消费者都能拿到消息
我们的计划是这样的 创建一个名为 hmall.fanout的Fanout交换机 创建两个队列fanout.queue1和fanout.queue2并绑定到hmall.fanout交换机上
3.5.1.声明队列和交换机
在控制台创建队列fanout.queue1: 再创建一个队列fanout.queue2 然后再创建一个交换机 然后绑定两个队列到交换机 3.5.2.消息发送
在publisher服务的SpringAmqpTest类中添加测试方法
第二个参数是路由键对于 Fanout 交换机路由键通常为空字符串因为 Fanout 交换机会将消息广播到所有与其绑定的队列而忽略路由键。
package com.itheima.publisher.amqp;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;SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testFanoutExchange() {// 交换机名称String exchangeName hmall.fanout;// 消息String message hello, everyone!;rabbitTemplate.convertAndSend(exchangeName, , message);}
}
3.5.3.消息接收
在consumer服务的SpringRabbitListener中添加两个方法作为消费者
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class 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 】);}
}
3.6.Direct交换机
在某些场景下我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。 在Direct模型下 队列与交换机的绑定不是任意绑定了而要给队列指定一个RoutingKey路由key 消息的发送方在 向 Exchange发送消息时也必须指定消息的 RoutingKey。 Exchange不再把消息交给每一个绑定的队列而是根据消息的Routing Key进行判断只有队列的Routingkey与消息的 Routing key完全一致才会接收到消息
案例需求如图 声明一个名为hmall.direct的交换机 声明队列direct.queue1绑定hmall.directbindingKey为blud和red 声明队列direct.queue2绑定hmall.directbindingKey为yellow和red 在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2 在publisher中编写测试方法向hmall.direct发送消息
3.6.1.声明队列和交换机
首先在控制台声明两个队列direct.queue1和direct.queue2这里不再展示过程 然后声明一个direct类型的交换机命名为hmall.direct: 然后使用red和blue作为RoutingKey绑定direct.queue1到hmall.direct 同理使用red和yellow作为key绑定direct.queue2到hmall.direct步骤略最终结果 3.6.2.消息接收
在consumer服务的SpringRabbitListener中添加方法
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class SpringRabbitListener {RabbitListener(queues direct.queue1)public void listenDirectQueue1(String msg) {System.out.println(消费者1接收到direct.queue1的消息【 msg 】);}RabbitListener(queues direct.queue2)public void listenDirectQueue2(String msg) {System.out.println(消费者2接收到direct.queue2的消息【 msg 】);}
}
3.6.3.消息发送
在publisher服务的SpringAmqpTest类中添加测试方法
package com.itheima.publisher.amqp;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;SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendDirectExchange() {// 交换机名称String exchangeName hmall.direct;// 消息String message 红色警报日本乱排核废水导致海洋生物变异惊现哥斯拉;// 发送消息rabbitTemplate.convertAndSend(exchangeName, red, message);}
}
由于使用的red这个key所以两个消费者都收到了消息 我们再切换为blue这个key
Test
public void testSendDirectExchange() {// 交换机名称String exchangeName hmall.direct;// 消息String message 最新报道哥斯拉是居民自治巨型气球虚惊一场;// 发送消息rabbitTemplate.convertAndSend(exchangeName, blue, message);
}
你会发现只有消费者1收到了消息 3.6.4.总结
描述下Direct交换机与Fanout交换机的差异 Fanout交换机将消息路由给每一个与之绑定的队列 Direct交换机根据RoutingKey判断路由给哪个队列 如果多个队列具有相同的RoutingKey则与Fanout功能类似
3.7.Topic交换机
3.7.1.说明
Topic交换机与Direct交换机相比都是根据RoutingKey把消息路由到不同的队列。
只不过Topic交换机可以让队列在绑定BindingKey 的时候使用通配符
BindingKey的组成BindingKey 一般都是有一个或多个单词组成多个单词之间以.分割例如 item.insert
通配符规则 #匹配0个或多个词 *匹配不多不少恰好1个词
举例 item.#能够匹配item.spu.insert 或者 item.spu item.*只能匹配item.spu
图示 假如此时publisher发送的消息使用的RoutingKey共有四种 china.news 代表有中国的新闻消息 china.weather 代表中国的天气消息 japan.news 则代表日本新闻 japan.weather 代表日本的天气消息
解释 topic.queue1绑定的是china.# 凡是以 china.开头的routing key 都会被匹配到包括 china.news china.weather topic.queue2绑定的是#.news 凡是以 .news结尾的 routing key 都会被匹配。包括: china.news japan.news
接下来我们就按照上图所示来演示一下Topic交换机的用法。
首先在控制台按照图示例子创建队列、交换机并利用通配符绑定队列和交换机。此处步骤略。最终结果如下 3.7.2.消息发送
在publisher服务的SpringAmqpTest类中添加测试方法
package com.itheima.publisher.amqp;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;SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;/*** topicExchange*/Testpublic void testSendTopicExchange() {// 交换机名称String exchangeName hmall.topic;// 消息String message 喜报孙悟空大战哥斯拉胜!;// 发送消息rabbitTemplate.convertAndSend(exchangeName, china.news, message);}
}
3.7.3.消息接收
在consumer服务的SpringRabbitListener中添加方法
package com.itheima.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class SpringRabbitListener {RabbitListener(queues topic.queue1)public void listenTopicQueue1(String msg){System.out.println(消费者1接收到topic.queue1的消息【 msg 】);}RabbitListener(queues topic.queue2)public void listenTopicQueue2(String msg){System.out.println(消费者2接收到topic.queue2的消息【 msg 】);}
}
3.7.4.总结
描述下Direct交换机与Topic交换机的差异 Topic交换机接收的消息RoutingKey必须是多个单词以 . 分割 Topic交换机与队列绑定时的bindingKey可以指定通配符 #代表0个或多个词 *代表1个词
3.8.Java代码声明交换机和队列 3.8.1.基本API
SpringAMQP提供了一个Queue类用来创建队列 SpringAMQP还提供了一个Exchange接口来表示所有不同类型的交换机 我们可以自己创建队列和交换机SpringAMQP还提供了ExchangeBuilder来简化这个过程 而在绑定队列和交换机时则需要使用BindingBuilder来创建Binding对象 3.8.2.fanout示例
在consumer中创建一个配置类声明队列和交换机
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.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class FanoutConfig {/*** 声明交换机* return Fanout类型交换机*/Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange(hmall.fanout);}/*** 第1个队列*/Beanpublic Queue fanoutQueue1(){return new Queue(fanout.queue1);}/*** 绑定队列和交换机*/Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/Beanpublic Queue fanoutQueue2(){return new Queue(fanout.queue2);}/*** 绑定队列和交换机*/Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
3.8.2.direct示例
direct模式由于要绑定多个KEY会非常麻烦每一个Key都要编写一个binding
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);}
}
3.8.4.基于注解声明
基于Bean方式的声明队列和交换机比较麻烦Spring还提供了基于注解方式来声明。
例如我们同样声明Direct模式的交换机和队列
RabbitListener注解的bindings属性指定 哪个队列 绑定到 哪个交换机上同时指定队列的BindingKey 指定RabbitListener注解的bindings属性的值为QueueBinding()注解 在QueueBinding()注解中又通过其value属性、exchange属性、key属性分别指定队列、交换机、队列的BindingKey value Queue(name direct.queue1)队列名 exchange Exchange(name hmall.direct, type ExchangeTypes.DIRECT)交换机名和交换机的类型 key {red, blue}队列的BindingKey
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 】);
}
3.9.消息转换器
SpringAMQP中发送消息的方法中接收消息的参数是Object类型 而在数据传输时它会把你发送的消息序列化为字节再发送给MQ接收消息的时候会把字节反序列化为Java对象。
只不过默认情况下SpringAMQP采用的序列化方式是JDK序列化。
JDK序列化存在下列问题在做消息转换时就是采用jdk自带的序列化方式 数据体积过大 有安全漏洞 可读性差
3.9.1.测试默认转换器
1创建测试队列
首先我们在consumer服务中声明一个新的配置类 利用Bean的方式创建一个队列
具体代码
package com.itheima.consumer.config;import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class MessageConfig {Beanpublic Queue objectQueue() {return new Queue(object.queue);}
}
注意这里我们先不要给这个队列添加消费者我们要查看消息体的格式。
重启consumer服务以后该队列就会被自动创建出来了 2发送消息
我们在publisher模块的SpringAmqpTest中新增一个消息发送的代码发送一个Map对象
Test
public void testSendMap() throws InterruptedException {// 准备消息MapString,Object msg new HashMap();msg.put(name, 柳岩);msg.put(age, 21);// 发送消息rabbitTemplate.convertAndSend(object.queue, msg);
}
发送消息后查看控制台 3.9.2.配置JSON转换器
显然JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高因此可以使用JSON方式来做序列化和反序列化。
①在publisher和consumer两个服务中都引入依赖 注意如果项目中引入了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可以便于我们将来做幂等性判断。
此时我们到MQ控制台删除object.queue中的旧的消息。然后再次执行刚才的消息发送的代码到MQ的控制台查看消息结构 3.9.3.消费者接收Object
我们在consumer服务中定义一个新的消费者publisher服务是用Map集合发送的消息所以消费者也一定要用Map接收格式如下
RabbitListener(queues object.queue)
public void listenSimpleQueueMessage(MapString, Object msg) throws InterruptedException {System.out.println(消费者接收到object.queue消息【 msg 】);
}
4.业务改造
案例需求改造余额支付功能将支付成功后使用OpenFeign调用交易服务的更新订单状态接口改为基于RabbitMQ的异步通知。
如图 步骤如下 定义direct类型交换机命名为pay.direct 定义消息队列命名为trade.pay.success.queue 将trade.pay.success.queue与pay.direct绑定BindingKey为pay.success 支付成功时不再调用交易服务更新订单状态的接口而是发送一条消息到pay.direct发送消息的RoutingKey 为pay.success消息内容是订单id 交易服务监听trade.pay.success.queue队列接收到消息后更新订单状态为已支付
4.1.配置MQ
不管是生产者还是消费者都需要配置MQ的基本信息。分为两步
1添加依赖 !--消息发送--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
2配置MQ地址
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
4.1.接收消息
在trade-service服务中定义一个消息监听的类 其代码如下 4.2.发送消息
修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法
private final RabbitTemplate rabbitTemplate;Override
Transactional
public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {// 1.查询支付单PayOrder po getById(payOrderDTO.getId());// 2.判断状态if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){// 订单不是未支付状态异常throw new BizIllegalException(交易已支付或关闭);}// 3.尝试扣减余额userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());// 4.修改支付单状态boolean success markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());if (!success) {throw new BizIllegalException(交易已支付或关闭);}// 5.修改订单状态// tradeClient.markOrderPaySuccess(po.getBizOrderNo());try {rabbitTemplate.convertAndSend(pay.direct, pay.success, po.getBizOrderNo());} catch (Exception e) {log.error(支付成功的消息发送失败支付单id{} 交易单id{}, po.getId(), po.getBizOrderNo(), e);}
} 文章转载自： http://www.morning.sbqrm.cn.gov.cn.sbqrm.cn http://www.morning.pljxz.cn.gov.cn.pljxz.cn http://www.morning.c7497.cn.gov.cn.c7497.cn http://www.morning.pjxw.cn.gov.cn.pjxw.cn http://www.morning.rgrz.cn.gov.cn.rgrz.cn http://www.morning.xcszl.cn.gov.cn.xcszl.cn http://www.morning.jhkzl.cn.gov.cn.jhkzl.cn http://www.morning.yktr.cn.gov.cn.yktr.cn http://www.morning.gmgyt.cn.gov.cn.gmgyt.cn http://www.morning.nkmw.cn.gov.cn.nkmw.cn http://www.morning.rjrlx.cn.gov.cn.rjrlx.cn http://www.morning.yhpq.cn.gov.cn.yhpq.cn http://www.morning.kqnwy.cn.gov.cn.kqnwy.cn http://www.morning.bwmm.cn.gov.cn.bwmm.cn http://www.morning.dxpqd.cn.gov.cn.dxpqd.cn http://www.morning.hdrsr.cn.gov.cn.hdrsr.cn http://www.morning.wcqkp.cn.gov.cn.wcqkp.cn http://www.morning.fnrkh.cn.gov.cn.fnrkh.cn http://www.morning.wsgyq.cn.gov.cn.wsgyq.cn http://www.morning.ftgwj.cn.gov.cn.ftgwj.cn http://www.morning.schwr.cn.gov.cn.schwr.cn http://www.morning.alwpc.cn.gov.cn.alwpc.cn http://www.morning.rwjfs.cn.gov.cn.rwjfs.cn http://www.morning.qwlml.cn.gov.cn.qwlml.cn http://www.morning.rwxnn.cn.gov.cn.rwxnn.cn http://www.morning.qjldz.cn.gov.cn.qjldz.cn http://www.morning.xmjzn.cn.gov.cn.xmjzn.cn http://www.morning.xswrb.cn.gov.cn.xswrb.cn http://www.morning.rtryr.cn.gov.cn.rtryr.cn http://www.morning.lonlie.com.gov.cn.lonlie.com http://www.morning.zylrk.cn.gov.cn.zylrk.cn http://www.morning.lgwjh.cn.gov.cn.lgwjh.cn http://www.morning.rbyz.cn.gov.cn.rbyz.cn http://www.morning.zxgzp.cn.gov.cn.zxgzp.cn http://www.morning.drqrl.cn.gov.cn.drqrl.cn http://www.morning.prplf.cn.gov.cn.prplf.cn http://www.morning.qmrsf.cn.gov.cn.qmrsf.cn http://www.morning.lwcqh.cn.gov.cn.lwcqh.cn http://www.morning.ltksw.cn.gov.cn.ltksw.cn http://www.morning.frsbf.cn.gov.cn.frsbf.cn http://www.morning.gwxwl.cn.gov.cn.gwxwl.cn http://www.morning.zdmlt.cn.gov.cn.zdmlt.cn http://www.morning.bxrqf.cn.gov.cn.bxrqf.cn http://www.morning.fznj.cn.gov.cn.fznj.cn http://www.morning.kchwr.cn.gov.cn.kchwr.cn http://www.morning.rfkyb.cn.gov.cn.rfkyb.cn http://www.morning.hxpsp.cn.gov.cn.hxpsp.cn http://www.morning.thzwj.cn.gov.cn.thzwj.cn http://www.morning.kdgcx.cn.gov.cn.kdgcx.cn http://www.morning.wscfl.cn.gov.cn.wscfl.cn http://www.morning.lnbcg.cn.gov.cn.lnbcg.cn http://www.morning.kfmnf.cn.gov.cn.kfmnf.cn http://www.morning.ymhjb.cn.gov.cn.ymhjb.cn http://www.morning.mqnbm.cn.gov.cn.mqnbm.cn http://www.morning.tclqf.cn.gov.cn.tclqf.cn http://www.morning.qnjcx.cn.gov.cn.qnjcx.cn http://www.morning.zlgbx.cn.gov.cn.zlgbx.cn http://www.morning.pngfx.cn.gov.cn.pngfx.cn http://www.morning.ppzgr.cn.gov.cn.ppzgr.cn http://www.morning.kqpq.cn.gov.cn.kqpq.cn http://www.morning.gqfks.cn.gov.cn.gqfks.cn http://www.morning.mzhgf.cn.gov.cn.mzhgf.cn http://www.morning.jcxgr.cn.gov.cn.jcxgr.cn http://www.morning.pcbfl.cn.gov.cn.pcbfl.cn http://www.morning.xqxlb.cn.gov.cn.xqxlb.cn http://www.morning.zlnmm.cn.gov.cn.zlnmm.cn http://www.morning.xgzwj.cn.gov.cn.xgzwj.cn http://www.morning.hmmnb.cn.gov.cn.hmmnb.cn http://www.morning.jpbky.cn.gov.cn.jpbky.cn http://www.morning.c7501.cn.gov.cn.c7501.cn http://www.morning.nyqzz.cn.gov.cn.nyqzz.cn http://www.morning.wzknt.cn.gov.cn.wzknt.cn http://www.morning.jhwwr.cn.gov.cn.jhwwr.cn http://www.morning.qytby.cn.gov.cn.qytby.cn http://www.morning.cnqwn.cn.gov.cn.cnqwn.cn http://www.morning.rbqlw.cn.gov.cn.rbqlw.cn http://www.morning.mrfr.cn.gov.cn.mrfr.cn http://www.morning.kfwrq.cn.gov.cn.kfwrq.cn http://www.morning.qbwyd.cn.gov.cn.qbwyd.cn http://www.morning.rzmzm.cn.gov.cn.rzmzm.cn
