海外营销网站建设,深圳做电商网站,seo技术手段,物流公司排名初识MQ
同步通讯和异步通讯 什么是同步通讯呢#xff1f;举个例子#xff0c;你认识了一个小姐姐#xff0c;聊的很火热#xff0c;于是你们慢慢开始打电话#xff0c;视频聊天#xff0c;这种方式就成为同步通讯#xff0c;那什么是一部通讯呢#xff0c;同样的…初识MQ
同步通讯和异步通讯 什么是同步通讯呢举个例子你认识了一个小姐姐聊的很火热于是你们慢慢开始打电话视频聊天这种方式就成为同步通讯那什么是一部通讯呢同样的你认识了多个小姐姐你和他们进行文字聊天这时候你一个人可以和多个人聊天这就是异步通讯。我们之前进行服务间调用时使用的RestTemplasteFeign就是同步调用。
同步调用的优缺点 优点时效性强可以立即得到回复就像你打视频一样 缺点假如你一个项目中存在很多业务相互之间进行调用如果你增加了新的需求此时因为原本代码是同步调用代码耦合度很高于是乎修改代码变得十分繁琐并且一个业务可能会调用很多服务只有上一个服务调用完了才能到下一个服务等待的过程中就造成了资源浪费性能下降如果当前调用的服务失败还可能会导致级联失败服务雪崩。 异步调用的优缺点
优点
1代码耦合度低因为异步调用是采取事件驱动来实现当请求进来之后到达Broker之后Broker通知各自微服务去执行服务间不在需要相互调用。
2吞吐量提升因为异步调用不像同步调用那样每个服务需要等待上游完成调用。
3故障隔离服务之间相互不进行调用即使你挂了也跟我没关系。
4流量削峰假如同时又大量请求但是你的服务处理请求能力是有限的于是Broker会净请求先拦截看服务又能力处理多少请求就拿多少请求不会一次性全部发布订阅。
缺点
1对Broker的依赖十分高对他的可靠性安全性吞吐能力要求很高万一他挂了......
2服务之间相互调用关系不清晰业务没有明显的流程线代码出问题不容易排查。 所以需要根据场景来选择同步还是异步一般大多数需要同步。
什么是MQ
MQMessageQueue中文是消息队列也就是存放消息的队列也就是时间驱动中的Broker.
常见的MQ对比 我们这里选择RabbitMQ
RabbitMQ安装
RabbitMQ是基于Erlang语言开发的开源消息中间件因此它具备强大的并发处理能力
官网地址:RabbitMQ: One broker to queue them all | RabbitMQhttps://www.rabbitmq.com/
这里我们只用Docker来安装RabbitMQ快捷方便
1拉取RabbitMQ的镜像
docker pull rabbitmq:3-management
2运行RabbitMQ容器
docker run \-e RABBITMQ_DEFAULT_USER你的账户名\-e RABBITMQ_DEFAULT_PASS你的密码\--name mq \ --hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3-management
注意账户名和密码是自己定义的15672是RabbitMQ的管理端端口5672是RabbitMQ通讯的端口。
3在浏览器输入IP地址:15672输入帐号名和密码登录 至此安装成功
我们可以看到界面有好多目录具体作用如下
channel用来操作mq的工具
exchange路由消息到队列中
queue缓存消息
virtual host虚拟主机是对queueexchange等资源的逻辑分组
MQ的整体结构
消息发送者将消息发送到交换机交换机将其路由到队列消费者从队列中取走消息 MQ中常见消息模型
大致可以分为两类
第一类基本消息队列BasicQueue工作消息队列WorkQueue他们都是最基本的消息队列 第二类发布订阅publishSubscribe根据交换机类型分为三种
Fanout Exchange广播Direct Exchange路由Topic Exchange主题 更多可参考官方文档
RabbitMQ Tutorials | RabbitMQ
RabbitMQ入门案例
我们使用RabbitMQ参考官方文档完成一个hello world案例
引入依赖 !--包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
创建一个测试类publisher用来发送消息代码如下
public class PublisherTest {Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接参数分别是主机名、端口号、vhost、用户名、密码factory.setHost(192.168.*.*);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(***);factory.setPassword(****);// 1.2.建立连接Connection connection factory.newConnection();// 2.创建通道ChannelChannel channel connection.createChannel();// 3.创建队列String queueName simple.queue;channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message hello, rabbitmq!;channel.basicPublish(, queueName, null, message.getBytes());System.out.println(发送消息成功【 message 】);// 5.关闭通道和连接channel.close();connection.close();}
}Debug执行观察
1连接工厂初始化完成之后他会创建一个连接连接上RabbitMQ 这时界面显示如下 2之后他会创建一个Channel 用于操作RabbbitMQ 3之后会创建一个我们定义好的消息队列simple.queue 4之后发送消息hellorabbitmq 5完成之后关闭连接但是消息依旧存在与消息队列中
之后创建一个测试类consumer用来接受消息代码如下
public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory new ConnectionFactory();// 1.1.设置连接参数分别是主机名、端口号、vhost、用户名、密码factory.setHost(192.168.*.*);factory.setPort(5672);factory.setVirtualHost(/);factory.setUsername(****);factory.setPassword(****);// 1.2.建立连接Connection connection factory.newConnection();// 2.创建通道ChannelChannel channel connection.createChannel();// 3.创建队列String queueName simple.queue;channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message new String(body);System.out.println(接收到消息【 message 】);}});System.out.println(等待接收消息。。。。);}
}跟publisher一样创建工厂建立连接这里需要说明的是之所以还要创建一个队列是因为在实际执行过程中发布者和消费者又可能执行顺序不一致所以我们消费者也需要创建一个队列不过这个队列只会有一个如果创建了就不再创建。之后消费者接受消息处理消息队列中消息清空。 根据上述代码我们可以看到官方的demo确实是有带你复杂繁琐实在是很不友好啊于是我们使用一种简单的方式来操作RabbitMQ
SpringAMQP
AMQPAdvanced Message Queuing Protocal高级消息队列协议是应用程序之间传递业务消息的开放标准/规范和语言和平台无关。
SpringAMQP是基于AMQP协议定义的一套API规范提供了模板用来发送和接受消息包含两部分其中spring-amop是基础抽象spring-rabbit是底层的默认实现。
SpringAMQP实现基础消息队列功能
消息发送
1引入依赖 !--AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
2在配置文件中配置RabbitMQ的信息
spring:rabbitmq:host: 192.168.121.10 #主机名port: 5672 #端口virtual-host: / #虚拟主机username: *** #用户名password: **** #密码
3使用RabbitTemplate来发送消息spring提供的发送消息的模板
这里我们编写一个测试类来测试
SpringBootTest
RunWith(SpringRunner.class)
public class AMQPTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void sendMessage(){String queueNamesimple.queue;//定义队列名String messagehello SpringAMQP!!;rabbitTemplate.convertAndSend(queueName,message);}
}可以看到消息成功发送 消息接收
依旧是先引依赖添加RabbitMQ相关配置
之后书写监听消息代码创建一个类交给spring管理定义一个方法加RabbitListener注解指定接受消息的队列可以传递数组指定多个队列
Component
public class ListenerMessage {RabbitListener(queues {simple.queue})//可以监听多个队列public void ListenerSimpleQueue(String message){System.out.println(接收到的消息是message);}
}可以看到成功接受消息 WordQueue 工作队列
Work模型-多个消费者绑定到同一个队列同一个消息只会被同一个消费者处理 我们来做一个测试在一秒内发送50条消息定义两个消费者同时处理一个消费者每秒处理50条消息另一个消费者每秒处理20个消息按照常理来说应该是多劳多得也就是处理能力强的处理更多消息弱的处理更少消息我们修改之前的代码
修改之后的消息发送代码
Testpublic void sendWorkMessage() throws InterruptedException {String queueNamesimple.queue;//定义队列名String messageWork Message__;for (int i 1; i 50; i) {rabbitTemplate.convertAndSend(queueName,messagei);Thread.sleep(20);}}
修改之后的消息接收代码
Component
public class ListenerMessage {RabbitListener(queues {simple.queue})//可以监听多个队列public void ListenerWorkSimpleQueue1(String message) throws InterruptedException {System.out.println(consumer1接收到的消息是message 时间LocalDateTime.now());Thread.sleep(20);}RabbitListener(queues {simple.queue})//可以监听多个队列public void ListenerWorkSimpleQueue2(String message) throws InterruptedException {System.err.println(consumer2接收到的消息是message 时间LocalDateTime.now());Thread.sleep(200);}
}代码执行结果显示consumer1和consumer2处理的消息是一样的并不会向我们设想的那样能力强的处理多能力弱的处理少是什么原因导致的呢
这是由于RabbitMQ的消息预取机制就是说在消息到达消息队列的时候两个消费者会分别从消息队列中一次性取完所有的消息理论上来说是无上限的所以我们需要修改机制让消费者一次性例如说取一个消息等这个消息处理完之后在去取下一个即可
在配置文件中修改如下
spring:rabbitmq:host: 192.168.101.100 #主机名port: 5672 #端口virtual-host: / #虚拟主机username: qmlx #用户名password: QMLX-9999 #密码listener:direct:prefetch: 1 #每次取一个消息取完之后在取发布Publish订阅Subscribe
发布订阅模式和之前案例的区别就是上述模型一个消息只能发送给一个consumer而发布订阅模型则是将同一个消息发送给多个消费者实现方式是假如exchange交换机 常见的exchange类型包括
Fanout广播
Direct路由
Topic话题
注意exchange只会负责消息路由而不是存储路由失败则消息丢失
发布订阅-Fanout Exchange
Fanout Exchange他会将接收到的消息路由到每一个与其绑定的queue广播模式。
实现思路
1在consumer服务中编写一个配置类声明两个队列一个交换机并将交换机绑定到队列上 Configuration
public class FanoutConfig {Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange(fanout.exchange);}//定义两个队列Beanpublic Queue fanoutQueue1(){return new Queue(fanout.queue1);}Beanpublic Queue fanoutQueue2(){return new Queue(fanout.queue2);}//将队列绑定在交换机上面Beanpublic Binding bindingQueue1(Queue fanoutQueue1,FanoutExchange fanoutExchange){//按照类型和名称传入参数return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}Beanpublic Binding bindingQueue2(Queue fanoutQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}}启动项目之后spring会自动加载交换机和队列 2编写consumer代码监听定义好的两个队列 Component
public class ListenerMessage {RabbitListener(queues {fanout.queue1})public void ListenerFanoutQueue1(String message){System.out.println(consumer1接受到的消息是:message);}RabbitListener(queues {fanout.queue2})public void ListenerFanoutQueue2(String message){System.err.println(consumer2接受到的消息是:message);}
}
3编写publisher代码发送消息
之前消息是发送到队列中现在消息发送给交换机 Testpublic void sendFanoutMessage() throws InterruptedException {String exchangeNamefanout.exchange;//定义队列名String messageFanoutExchange Message!!;rabbitTemplate.convertAndSend(exchangeName,null,message);}
可以看到两个消费者同时接受到了消息 交换机的作用
1接受publisher发送的消息
2将消息按照规则路由发送给每一个与之绑定的队列
3不能缓存信息路由失败则消息丢失
发布订阅-Direct Exchange
Direct Exchange交换机将接收到的消息根据路由规则到指定的Queue称之为路由模式routes 每一个Queue都和Exchange设定一个BindingKey 发布者发送消息时指定消息的BindingKey Exchange将消息路由到BindKey和消息的BindingKey一致的队列 实现思路
1指定消息接收者绑定交换机和队列但是需要指定BindingKey
这次直接使用注解实现不必那莫繁琐
Component
public class ListenerDirectMessage {RabbitListener(bindings QueueBinding(value Queue(name direct.queue1),exchange Exchange(name direct.exchange,type ExchangeTypes.DIRECT),key {red,blue}))public void listensterDirectQueue1(String message){System.out.println(消费者1接收到Direct的消息是message);}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name direct.exchange,type ExchangeTypes.DIRECT),key {red,blue}))public void listensterDirectQueue2(String message){System.out.println(消费者2接收到Direct的消息是message);}
}‘2定义publisher发布消息
Testpublic void sendDirectMessage() throws InterruptedException {String exchangeNamedirect.exchange;//定义队列名String messageDirectExchange Message!!;String routinfKeyblue;rabbitTemplate.convertAndSend(exchangeName,routinfKey,message);rabbitTemplate.convertAndSend(exchangeName,null,message);}
发送时需要指定routingKey即可同一个队列定义时可指定多个BindingKey
注意Direct Exchange只会发送给routingKey和BindingKey一致的队列
发布订阅-Topic Exchange
Topic Exchange:Topic Exchange和Direct key类似区别在于他的routingKey必须是多个单词的列表并且必须是以 . 分割并且Queue和Exchange指定时课使用通配符 实现思路
1实现消费者代码 Component
public class ListenerTopicMessage {RabbitListener(bindings QueueBinding(value Queue(name topic.queue1),exchange Exchange(name topic.exchange,type ExchangeTypes.TOPIC),key china.#))public void listensterDirectQueue1(String message){System.out.println(消费者1接收到topic的消息是message);}RabbitListener(bindings QueueBinding(value Queue(name topic.queue2),exchange Exchange(name topic.exchange,type ExchangeTypes.TOPIC),key #.news))public void listensterDirectQueue2(String message){System.out.println(消费者2接收到topic的消息是message);}
}2实现publisher代码
Testpublic void sendtopictMessage() throws InterruptedException {String exchangeNametopic.exchange;//定义队列名String messageTopicExchange Message!!;String routinfKeychina.news;rabbitTemplate.convertAndSend(exchangeName,routinfKey,message);//rabbitTemplate.convertAndSend(exchangeName,null,message);}
至此RabbitMQ的安装使用及其五种基本工作模式搞定
