朝阳专业网站建设,网站开发与移动互联,如何通过建设网站赚钱,可以做免费广告的网站有哪些目录
MQ
MQ概述
MQ 的优势
1.应用解耦
2.异步提速
3.削峰填谷
MQ 的劣势
1.系统可用性降低
2.系统复杂度提高
3.一致性问题
使用 MQ 需要满足什么条件呢#xff1f; RabbitMQ 简介
编辑RabbitMQ 中的相关概念
RabbitMQ 提供了 6 种工作模式
JMS
java实现Ra…目录
MQ
MQ概述
MQ 的优势
1.应用解耦
2.异步提速
3.削峰填谷
MQ 的劣势
1.系统可用性降低
2.系统复杂度提高
3.一致性问题
使用 MQ 需要满足什么条件呢 RabbitMQ 简介
编辑RabbitMQ 中的相关概念
RabbitMQ 提供了 6 种工作模式
JMS
java实现Rabbitmq
依赖导入和基础配置
简单模式
Work queues 工作队列模式
Pub/Sub 订阅模式
Routing 路由模式
Topics 通配符模式 Spring整合Rabbitmq
供应者
消费者 SpringBoot整合Rabbitmq MQ 目前业界有很多的 MQ 产品如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等也有直接使用 Redis 充当消息队列的案例而这些消息队列产品各有侧重在实际选型时需要结合自身需 求及 MQ 产品特征综合考虑。 MQ概述 MQ全称 Message Queue消息队列是在消息的传输过程中保存消息的容器。多用于分布式系统之间进 行通信。 MQ 的优势
1.应用解耦 使用 MQ 使得应用间解耦提升容错性和可维护性。
2.异步提速 一个下单操作耗时20 300 300 300 920ms 用户点击完下单按钮后需要等待920ms才能得到下单响应太慢了 用户点击完下单按钮后只需等待25ms就能得到下单响应 (20 5 25ms)。
提升用户体验和系统吞吐量单位时间内处理请求的数目。
3.削峰填谷 MQ 之后限制消费消息的速度为1000这样一来高峰期产生的数据势必会被积压在 MQ 中高峰 就被“削”掉了但是因为消息积压在高峰期过后的一段时间内消费消息的速度还是会维持在1000直 到消费完积压的消息这就叫做“填谷”。 MQ 的劣势
1.系统可用性降低 系统引入的外部依赖越多系统稳定性越差。一旦 MQ 宕机就会对业务造成影响。如何保证MQ的高可用
2.系统复杂度提高 MQ 的加入大大增加了系统的复杂度以前系统间是同步的远程调用现在是通过 MQ 进行异步调用。如何 保证消息没有被重复消费怎么处理消息丢失情况那么保证消息传递的顺序性
3.一致性问题 A 系统处理完业务通过 MQ 给B、C、D三个系统发消息数据如果 B 系统、C 系统处理成功D 系统处理 失败。如何保证消息数据处理的一致性
使用 MQ 需要满足什么条件呢
生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用其接口的返回值应该为空这才让明 明下层的动作还没做上层却当成动作做完了继续往后走即所谓异步成为了可能。容许短暂的不一致性。确实是用了有效果。即解耦、提速、削峰这些方面的收益超过加入MQ管理MQ这些成本。 RabbitMQ 简介 AMQP即 Advanced Message Queuing Protocol高级消息队列协议是一个网络协议是应用层协议 的一个开放标准为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息并不受客户端/中 间件不同产品不同的开发语言等条件的限制。2006年AMQP 规范发布。类比HTTP。 RabbitMQ 中的相关概念
⚫ Broker接收和分发消息的应用RabbitMQ Server就是 Message Broker
⚫ Virtual host出于多租户和安全因素设计的把 AMQP 的基本组件划分到一个虚拟的分组中类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时可以划分出多 个vhost每个用户在自己的 vhost 创建 exchangequeue
⚫ Connectionpublisherconsumer 和 broker 之间的 TCP 连接
⚫ Channel如果每一次访问 RabbitMQ 都建立一个 Connection在消息量大的时候建立 TCP Connection 的开销将是巨大的效率也较低。Channel 是在 connection 内部建立的逻辑连接如果应用程序支持多线 程通常每个thread创建单独的 channel 进行通讯AMQP method 包含了channel id 帮助客户端和 message broker 识别 channel所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销
⚫ Exchangemessage 到达 broker 的第一站根据分发规则匹配查询表中的 routing key分发消息到 queue 中去。常用的类型有direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
⚫ Queue消息最终被送到这里等待 consumer 取走
⚫ Bindingexchange 和 queue 之间的虚拟连接binding 中可以包含 routing key。Binding 信息被保存 到 exchange 中的查询表中用于 message 的分发依据
RabbitMQ 提供了 6 种工作模式 简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式远程调用不太算 MQ暂不作介绍。 官网对应模式介绍https://www.rabbitmq.com/getstarted.html JMS JMS 即 Java 消息服务JavaMessage Service应用程序接口是一个 Java 平台中关于面向消息中间件的API JMS 是 JavaEE 规范中的一种API 规范接口类比JDBC 很多消息中间件都实现了JMS规范例如ActiveMQ。RabbitMQ 官方没有提供 JMS 的实现包但是开源社区有
java实现Rabbitmq
依赖导入和基础配置 dependencies!--rabbitmq java 客户端--dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.6.0/version/dependency/dependenciesbuildpluginsplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-compiler-plugin/artifactIdversion3.8.0/versionconfigurationsource1.8/sourcetarget1.8/target/configuration/plugin/plugins/build
基础配置信息 //1.创建连接工厂ConnectionFactory factory new ConnectionFactory();//2. 设置参数factory.setHost(121.37.118.193);//ip 默认值 localhostfactory.setPort(5672); //端口 默认值 5672factory.setVirtualHost(dobby);//虚拟机 默认值/factory.setUsername(dobby);//用户名 默认 guestfactory.setPassword(123456);//密码 默认值 guest//3. 创建连接 ConnectionConnection connection factory.newConnection();//4. 创建ChannelChannel channel connection.createChannel(); 简单模式 P生产者也就是要发送消息的程序C消费者消息的接收者会一直等待消息到来queue消息队列图中红色部分。类似一个邮箱可以缓存消息生产者向其中投递消息消费者从其中取出消息 //创建队列Queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, MapString, Object arguments)参数1. queue队列名称2. durable:是否持久化当mq重启之后还在3. exclusive* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时是否删除队列*4. autoDelete:是否自动删除。当没有Consumer时自动删除掉5. arguments参数。*///如果没有一个名字叫hello_world的队列则会创建该队列如果有则不会创建channel.queueDeclare(hello_world,true,false,false,null);/*basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)参数1. exchange交换机名称。简单模式下交换机会使用默认的 2. routingKey路由名称3. props配置信息4. body发送消息数据*/String body hello rabbitmq~~~;//发送消息channel.basicPublish(,hello_world,null,body.getBytes());
Work queues 工作队列模式 Work Queues与入门程序的简单模式相比多了一个或一些消费端多个消费端共同消费同一个队列中的消息。在一个队列中如果有多个消费者那么消费者之间对于同一个消息的关系是竞争的关系。
与简单模式供应端代码一致只不过对应多个消费端以下消费端代码示例 String queue1Name test_fanout_queue1;// 接收消息Consumer consumer new DefaultConsumer(channel){/*回调方法当收到消息后会自动执行该方法1. consumerTag标识2. envelope获取一些信息交换机路由key...3. properties:配置信息4. body数据*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {/* System.out.println(consumerTagconsumerTag);System.out.println(Exchangeenvelope.getExchange());System.out.println(RoutingKeyenvelope.getRoutingKey());System.out.println(propertiesproperties);*/System.out.println(bodynew String(body));System.out.println(将日志信息保存数据库.....);}};/*basicConsume(String queue, boolean autoAck, Consumer callback)参数1. queue队列名称2. autoAck是否自动确认3. callback回调对象*/channel.basicConsume(queue2Name,true,consumer);
Pub/Sub 订阅模式 在订阅模型中多了一个 Exchange 角色而且过程略有变化 P生产者也就是要发送消息的程序但是不再发送到队列中而是发给X交换机 C消费者消息的接收者会一直等待消息到来 Queue消息队列接收消息、缓存消息 Exchange交换机X。 一方面接收生产者发送的消息。另一方面知道如何处理消息例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作取决于Exchange的类型。Exchange有常见以下3种类型 Fanout广播将消息交给所有绑定到交换机的队列 Direct定向把消息交给符合指定routing key 的队列 Topic通配符把消息交给符合routing pattern路由模式 的队列 Exchange交换机只负责转发消息不具备存储消息的能力因此如果没有任何队列与 Exchange 绑定或者没有符合路由规则的队列那么消息会丢失 发布/订阅模式需要设置队列和交换机的绑定工作队列模式不需要设置实际上工作队列模式会将队 列绑 定到默认的交换机 String exchangeName test_fanout;//创建交换机/*exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, MapString, Object arguments)参数1. exchange:交换机名称2. type:交换机类型DIRECT(direct),定向FANOUT(fanout),扇形广播发送消息到每一个与之绑定队列。TOPIC(topic),通配符的方式HEADERS(headers);参数匹配3. durable:是否持久化4. autoDelete:自动删除5. internal内部使用。 一般false6. arguments参数*/channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);//创建队列String queue1Name test_fanout_queue1;String queue2Name test_fanout_queue2;channel.queueDeclare(queue1Name,true,false,false,null);channel.queueDeclare(queue2Name,true,false,false,null);//绑定队列和交换机/*queueBind(String queue, String exchange, String routingKey)参数1. queue队列名称2. exchange交换机名称3. routingKey路由键绑定规则如果交换机的类型为fanout routingKey设置为*/channel.queueBind(queue1Name,exchangeName,);channel.queueBind(queue2Name,exchangeName,);String body 日志信息张三调用了findAll方法...日志级别info...;//发送消息channel.basicPublish(exchangeName,,null,body.getBytes());//释放资源channel.close();connection.close();
Routing 路由模式 Exchange 不再把消息交给每一个绑定的队列而是根据消息的 Routing Key 进行判断只有队列的Routingkey 与消息的 Routing key 完全一致才会接收到消息
P生产者向 Exchange 发送消息发送消息时会指定一个routing keyXExchange交换机接收生产者的消息然后把消息递交给与 routing key 完全匹配的队列C1消费者其所在队列指定了需要 routing key 为 error 的消息C2消费者其所在队列指定了需要 routing key 为 info、error、warning 的消息 Routing 模式要求队列在绑定交换机时要指定 routing key消息会转发到符合 routing key 的队列。 //创建交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null);//创建队列String queue1Name test_direct_queue1;String queue2Name test_direct_queue2;channel.queueDeclare(queue1Name,true,false,false,null);channel.queueDeclare(queue2Name,true,false,false,null);//绑定队列和交换机//队列1绑定 errorchannel.queueBind(queue1Name,exchangeName,error);//队列2绑定 info error warningchannel.queueBind(queue2Name,exchangeName,info);channel.queueBind(queue2Name,exchangeName,error);channel.queueBind(queue2Name,exchangeName,warning);String body 日志信息张三调用了delete方法...出错误了。。。日志级别error...;//8. 发送消息channel.basicPublish(exchangeName,warning,null,body.getBytes());
Topics 通配符模式 红色 Queue绑定的是 usa.# 因此凡是以 usa. 开头的 routing key 都会被匹配到黄色 Queue绑定的是 #.news 因此凡是以 .news 结尾的 routing key 都会被匹配通配符规则# 匹配一个或多个词* 匹配不多不少恰好1个词例如item.# 能够匹配 item.insert.abc 或者 item.insertitem.* 只能匹配 item.insert //创建交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null);...// routing key 系统的名称.日志的级别。//需求 所有error级别的日志存入数据库所有order系统的日志存入数据库channel.queueBind(queue1Name,exchangeName,#.error);channel.queueBind(queue1Name,exchangeName,order.*);channel.queueBind(queue2Name,exchangeName,*.*);String body 日志信息张三调用了findAll方法...日志级别info...;//8. 发送消息channel.basicPublish(exchangeName,goods.error,null,body.getBytes()); Spring整合Rabbitmq
供应者
rabbitmq.properties
rabbitmq.host121.37.118.193
rabbitmq.port5672
rabbitmq.usernamedobby
rabbitmq.password123456
rabbitmq.virtual-hostdobbySpring-rabbitmq-prodicer.xml
?xml version1.0 encodingUTF-8?
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:contexthttp://www.springframework.org/schema/contextxmlns:rabbithttp://www.springframework.org/schema/rabbitxsi:schemaLocationhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd!--加载配置文件--context:property-placeholder locationclasspath:rabbitmq.properties/!-- 定义rabbitmq connectionFactory --rabbit:connection-factory idconnectionFactory host${rabbitmq.host}port${rabbitmq.port}username${rabbitmq.username}password${rabbitmq.password}virtual-host${rabbitmq.virtual-host}/!--定义管理交换机、队列--rabbit:admin connection-factoryconnectionFactory/!--定义持久化队列不存在则自动创建不绑定到交换机则绑定到默认交换机默认交换机类型为direct名字为路由键为队列的名称--!--idbean的名称namequeue的名称auto-declare:自动创建auto-delete:自动删除。 最后一个消费者和该队列断开连接后自动删除队列exclusive:是否独占durable是否持久化--rabbit:queue idspring_queue namespring_queue auto-declaretrue/!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播所有队列都能收到消息~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --!--定义广播交换机中的持久化队列不存在则自动创建--rabbit:queue idspring_fanout_queue_1 namespring_fanout_queue_1 auto-declaretrue/!--定义广播交换机中的持久化队列不存在则自动创建--rabbit:queue idspring_fanout_queue_2 namespring_fanout_queue_2 auto-declaretrue/!--定义广播类型交换机并绑定上述两个队列--rabbit:fanout-exchange idspring_fanout_exchange namespring_fanout_exchange auto-declaretruerabbit:bindingsrabbit:binding queuespring_fanout_queue_1 /rabbit:binding queuespring_fanout_queue_2//rabbit:bindings/rabbit:fanout-exchange!--rabbit:direct-exchange nameaa rabbit:bindingslt;!ndash;direct 类型的交换机绑定队列 key 路由key queue队列名称ndash;gt;rabbit:binding queuespring_queue keyxxx/rabbit:binding/rabbit:bindings/rabbit:direct-exchange--!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符*匹配一个单词#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ --!--定义广播交换机中的持久化队列不存在则自动创建--rabbit:queue idspring_topic_queue_star namespring_topic_queue_star auto-declaretrue/!--定义广播交换机中的持久化队列不存在则自动创建--rabbit:queue idspring_topic_queue_well namespring_topic_queue_well auto-declaretrue/!--定义广播交换机中的持久化队列不存在则自动创建--rabbit:queue idspring_topic_queue_well2 namespring_topic_queue_well2 auto-declaretrue/rabbit:topic-exchange idspring_topic_exchange namespring_topic_exchange auto-declaretruerabbit:bindingsrabbit:binding patternheima.* queuespring_topic_queue_star/rabbit:binding patternheima.# queuespring_topic_queue_well/rabbit:binding patternitcast.# queuespring_topic_queue_well2//rabbit:bindings/rabbit:topic-exchange!--定义rabbitTemplate对象操作可以在代码中方便发送消息--rabbit:template idrabbitTemplate connection-factoryconnectionFactory/
/beans
测试代码 RunWith(SpringJUnit4ClassRunner.class)
ContextConfiguration(locations classpath:spring-rabbitmq-producer.xml)
public class ProducerTest {//1.注入 RabbitTemplateAutowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testHelloWorld(){//2.发送消息rabbitTemplate.convertAndSend(spring_queue,hello world spring....);}/*** 发送fanout消息*/Testpublic void testFanout(){//2.发送消息rabbitTemplate.convertAndSend(spring_fanout_exchange,,spring fanout....);}/*** 发送topic消息*/Testpublic void testTopics(){//2.发送消息rabbitTemplate.convertAndSend(spring_topic_exchange,heima.hehe.haha,spring topic....);}
}SpringQueueListener
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;public class SpringQueueListener implements MessageListener {Overridepublic void onMessage(Message message) {//打印消息System.out.println(new String(message.getBody()));}
}消费者
Spring-rabbitmq-consumer.xml
?xml version1.0 encodingUTF-8?
beans xmlnshttp://www.springframework.org/schema/beansxmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexmlns:contexthttp://www.springframework.org/schema/contextxmlns:rabbithttp://www.springframework.org/schema/rabbitxsi:schemaLocationhttp://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/contexthttps://www.springframework.org/schema/context/spring-context.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit.xsd!--加载配置文件--context:property-placeholder locationclasspath:rabbitmq.properties/!-- 定义rabbitmq connectionFactory --rabbit:connection-factory idconnectionFactory host${rabbitmq.host}port${rabbitmq.port}username${rabbitmq.username}password${rabbitmq.password}virtual-host${rabbitmq.virtual-host}/bean idspringQueueListener classcom.itheima.rabbitmq.listener.SpringQueueListener/!-- 这里就需要创建多个Listener的class在此我们只测试一个SpringQueueListener--!--bean idfanoutListener1 classcom.itheima.rabbitmq.listener.FanoutListener1/bean idfanoutListener2 classcom.itheima.rabbitmq.listener.FanoutListener2/bean idtopicListenerStar classcom.itheima.rabbitmq.listener.TopicListenerStar/bean idtopicListenerWell classcom.itheima.rabbitmq.listener.TopicListenerWell/bean idtopicListenerWell2 classcom.itheima.rabbitmq.listener.TopicListenerWell2/
--rabbit:listener-container connection-factoryconnectionFactory auto-declaretruerabbit:listener refspringQueueListener queue-namesspring_queue/!-- rabbit:listener reffanoutListener1 queue-namesspring_fanout_queue_1/rabbit:listener reffanoutListener2 queue-namesspring_fanout_queue_2/rabbit:listener reftopicListenerStar queue-namesspring_topic_queue_star/rabbit:listener reftopicListenerWell queue-namesspring_topic_queue_well/rabbit:listener reftopicListenerWell2 queue-namesspring_topic_queue_well2/--/rabbit:listener-container
/beans SpringBoot整合Rabbitmq !--1. 父工程依赖--parentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.1.4.RELEASE/version/parentdependencies!--2. rabbitmq--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependency/dependenciesConfiguration
public class RabbitMQConfig {public static final String EXCHANGE_NAME boot_topic_exchange;public static final String QUEUE_NAME boot_queue;//1.交换机Bean(bootExchange)public Exchange bootExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}//2.Queue 队列Bean(bootQueue)public Queue bootQueue(){return QueueBuilder.durable(QUEUE_NAME).build();}//3. 队列和交互机绑定关系 Binding/*1. 知道哪个队列2. 知道哪个交换机3. routing key*/Beanpublic Binding bindQueueExchange(Qualifier(bootQueue) Queue queue, Qualifier(bootExchange) Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with(boot.#).noargs();}}
# 配置RabbitMQ的基本信息 ip 端口 username password..
spring:rabbitmq:host: 172.16.98.133 # ipport: 5672username: guestpassword: guestvirtual-host: /
public class ProducerTest {//1.注入RabbitTemplateAutowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSend(){rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,boot.haha,boot mq hello~~~);}
}消费者
RabbimtMQListener
Component
public class RabbimtMQListener {RabbitListener(queues boot_queue)public void ListenerQueue(Message message){//System.out.println(message);System.out.println(new String(message.getBody()));}}
