织梦 网站搬家,网页设计代码大全下载,学生个人网页制作成品,什么是网站后台1.同步通讯和异步通讯 同步通讯#xff1a;如果举个例子来说#xff0c;同步通讯就像是两个人在打电话#xff0c;一方说的话#xff0c;能够立马传给另一方#xff0c;消息的时效性非常高#xff0c;但是相对的#xff0c;只能是给一个人通讯#xff0c;如果这个时候如果举个例子来说同步通讯就像是两个人在打电话一方说的话能够立马传给另一方消息的时效性非常高但是相对的只能是给一个人通讯如果这个时候有另一个人想和你建立联系进行通讯那么抱歉你必须得先结束当下的通讯才能和其他人通讯异步通讯如果把同步通讯比作打电话那么异步通讯就像是在发微信我发给另外一个人消息但是这个人不一定会立马看见并且回复我通信的时效性不强但是它可以同时给多方发送消息试想一下我们平时发微信的时候是可以同时给多个人发送消息并且也能够同时接收多个人发过来的消息 同步通讯——打电话 异步通讯——发微信 2.在某些场景中使用同步通讯的缺陷 比如微服务直接相互调用的feign中间件使用的就是同步调用具体来看缺陷有以下几点 这是一个简单的用户支付模块牵扯的服务用户在支付完之后后台需要调用如订单服务、仓储服务、短信服务之类的模块来完善这个支付功能由于是支付功能来直接调用的其他服务所以服务之间的耦合性较高当我们要再添加一个积分服务来让支付服务调用那么就需要修改支付服务中的代码麻烦在支付业务调用别的服务的时候由于是同步通讯所以只能一个一个调用当前服务调用完之后才能调用别的服务比如上面这个情况支付服务调用完订单服务之后才能调用仓储服务调用完仓储服务后才能调用短信服务耗时太长了假如仓储服务突然挂掉了那么支付服务在调用这个仓储服务时就会停滞在这里那么当有更多的用户来进行支付那么就有越来越多的请求卡在这一块当支付服务的资源被耗尽之后也就会挂掉像这样一条链上的服务有其中一个服务挂掉相应的也会影响别的服务跟着挂掉就是级联失效问题。 3.异步调用方案解决相关问题 还是上面的这个例子俗话说在web架构中没有什么是加一层解决不了的所以我们在支付服务和其他被调用的服务之间加一个代理人Broker 你可以理解成当我们去吃饭的时候和我们交涉的前台人员我们向前台点餐而不会直接向厨房点餐(当然路边小摊只有老板一个人做饭点餐当我没说)那么此时这种情况我们就属于客户端厨房大厨就是服务端而这个负责接待我们的前台人员就是代理人他来代替我们和厨房之间通信 当用户支付成功之后支付服务就会通知代理人BROKER说“有一个顾客已经成功下单了订单是1001”那么这个BROKER就会通知其余服务说“客户订单1001已经支付完成了你们赶紧来处理一下”此时这些服务就会来完成各自的业务。但是要注意的是被调用的服务要通过BROKER订阅事件相当于让这些服务有能力接收到BROKER发布的事件不接收到BROKER发布的事件自然是无法处理这个事件的 在整个过程中支付服务在用户支付成功之后就只会通知一下BROKER发布一个用户完成支付的事件其余的比如调用其他业务它都不再伸手了联想下现实生活当我们跟前台点完餐之后也确实不会再和后台的厨师交流什么了所以这样就解除了服务与服务之间的耦合了当支付服务被用户使用并且成功支付之后除了发布事件以外它就不再干别的了所以又能继续接待新用户的支付至于后台的别的服务什么时候去接收处理这个订单事件就不是用户和支付服务管的了这样数据的吞吐量就增高了 试想一下我们平时淘宝购物的时候我们只要支付成功了并且前台给我们回复支付成功的消息之后是不是就不会管人家后台的事情了倘若是同步通讯那我们要等到后台将这个事件处理完毕之后才会得到答复在此之间我们就会一直占着这个支付服务是不是太麻烦了由于异步通讯服务之间的依赖性不强所以当有其中一个服务挂掉了之后也不会影响其他服务的正常运行比如仓储服务挂了他并不会有影响支付服务的正常运行不用担心级联失败的问题随着业务量增加请求越来越多假如一下子来了十个订单完成的服务但是订单服务、仓储服务这些只能在同一时间处理一个事件这时BROKER就会起到一个缓冲的作用就像水库里的水坝一样他可以将事件排成队列当其他服务处理完一个事件之后再放给他们下一个事件这种操作称之为流量削峰 同样的有优点就有缺点这样加一个BROKER那么基本上中间通讯就全部依赖这个BROKER了那么就吃这个中间代理人的可靠性、安全性、吞吐能力了不得不说这个服务员是真不好干。 4.RabbitMQ MQ(MessageQueue)消息队列。用来存放消息也就是刚刚在上面提到的事件。这个MQ相当于就是事件驱动框架里的BROKER MQ的具体实现有很多种例如RabbitMQActiveMQRocketMQ和Kafka下面来重点介绍一下RabbitMQ 4.1.安装RabbitMQ 这里我们采用docker来部署安装这里我们可以使用docker命令直接从仓库拉取也可以自己上传rabbitmq的压缩包我采用的是后者安装包小伙伴可以自行上官网下载 下载好了压缩包我们直接将其拖入finalshell拖入之后会自行下载然后使用docker load命令下载rabbitmq的镜像然后使用docker run命令启动容器就ok了具体命令如下 docker run \-e RABBITMQ_DEFAULT_USERitcast \-e RABBITMQ_DEFAULT_PASS123321 \--name mq \--hostname mq1 \-p 15672:15672 \-p 5672:5672 \-d \rabbitmq:3-management 稍微看下这个docker run命令首先设置了两个环境变量一个用户名一个密码然后设置了一个主机名这个是集群部署时会用到然后映射了两个端口第一个15672是rabbitmq自己的可视化终端的端口第二个5672是事件传输的端口 访问一下rabbitmq的可视化终端 看一下rabbitMQ的服务框架 从这个框架我们可以看出消息事件的生产者将事件转发给exchange这个exchange担任一个路由转发的角色它将事件路由给不同的queue也就是存放消息的队列最后消费者再从这个队列中取出事件来处理其中要留意的是virtual host(虚拟主机)它类似docker的容器具有隔离效果每个用户绑定一个独立的虚拟主机达到互不干扰的效果 4.2.SpringAMQP AMQPAdvanced Message Queuing Protocol高级消息协议是一个进程间传递异步消息的网络协议而SpringAMQP就是基于这种协议的消息中间件框架它提供了一个简单的API来发送和接收异步、可靠的消息其主要特点有以下几点 提供监听器容器用来异步处理发送来的消息提供RabbitTemplate来接收和发送消息提供RabbitAdmin用于自动声明 queues、exchanges 和 bindings 总而言之这个SpringAMQP框架大大减少了使用RabbitMQ的成本 怎么使用这个SpringAMQP呢请往下看 4.3.一个简单的消息队列的demo 这个demo分了两个服务一个是consumer消费者另一个是publisher生产(发布)者我们要做的首先是引入SpringAMQP的依赖下面引入了一套完整的依赖包含springbootspringamqp和junit测试(这些依赖是引入在父工程里的子工程就不用引入了) !--AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!--单元测试--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdscopetest/scope/dependency 紧接着在生产者publisher服务中编写配置文件来告诉spring我们要使用rabbitmq来进行消息的发布和接收其中我们告诉了spring我们要使用的rabbitmq的地址端口用户名密码和虚拟主机名称host地址是我们虚拟机的地址port端口号是我们用docker启动rabbitmq容器设置的一般来说都是5672用户名和密码也是在那个时候设置的虚拟主机是rabbitmq默认的我们可以在可视化界面里更改这里不做演示了 spring:rabbitmq:host: 192.168.88.128port: 5672username: itcastpassword: 123321virtual-host: / 然后就是编写测试单元来发送消息 RunWith(SpringRunner.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testMessageToSimpleQueue(){String queueName simplequeue;String message hello AMQP;rabbitTemplate.convertAndSend(queueName,message);}
} 注意这是一个springboot的测试类所以要加上SpringBootTest的注解由于我们要使用注入的类(这里用autowired自动注入的RabbitTemplate)所以还要添加上RunWith的注解这个注解在Junit5中是被携带上的在Junit4中没有所以我们得单另声明下 使用RabbitTemplate类的convertAndSent方法来完成消息的发送参数有两个一个是队列的名称另一个是消息内容这里的队列的名称我是事先创建好的没有的可以去rabbitmq可视化界面创建一个 运行测试单元之后 点击队列名称我们进入到这个队列中查看一下消息内容 发现正是我们刚刚发送的消息内容 好了到这里发送消息的服务已经看完了我们再来看看接收消息的服务
和发送消息的步骤类似
优先引入springamqp的依赖不过刚刚说了我们是在父工程中引入的所以子工程就不用引入了然后就是编写配置文件和消息发送方的一样最后一步就是编写接收的代码 刚刚在上面说了springamqp的特性了它提供一个专门来监听消息的容器我们只需要把容器创建好就能接收发送来的消息了代码如下 Component
public class SpringRabbitListener {RabbitListener(queues simplequeue)public void listenSimpleQueue(String msg){System.out.println(接收到消息了: msg);}
} 创建一个消息监听类并把它注册到spring的容器中指定好监听的队列这里监听的是simplequeue这个队列创建一个接收消息的方法注意参数类型和发送的消息类型是一致的 成功接收到消息 注意Queue中存放的消息一旦被消费者处理了之后就销毁了有种阅后即焚的感觉所以这个消息被消费者接收到之后我们再看Queue队列中发现已经没有消息了 5.五种消息发布模型 基本消息队列Basic Queue工作消息队列Work Queue发布订阅模式Publish、Subscribe其中根据交换机的不同分了三种 广播Fanout Exchange路由Direct Exchange主题Topic Exchange 5.1.基本消息队列
基本消息队列就是刚刚4.3中讲的那个demo 由发布者publisher队列queue消费者consumer组成 5.2.工作消息队列
工作消息队列可以有更多的消费者来同时监听同一个队列让消息处理的效率提高 剩下的三种模式我们之后再介绍现在我们来重点讲讲这个工作消息队列的处理机制
假如现在有一种情况同一个队列绑定了两个消费者但这两个消费者处理消息的能力不一样第一个消费者处理能力强于第二个消费者我们来看看会不会处理能力强的消费者会处理更多的消息 /*** 发送50次消息总共一秒钟发完* throws InterruptedException*/Testpublic void testMessageToSimpleQueue() throws InterruptedException {String queueName simplequeue;String message hello AMQP;for (int i 0; i 50; i) {rabbitTemplate.convertAndSend(queueName,message);Thread.sleep(20);}} 消息发送方一共发送50次消息共1秒发送完 RabbitListener(queues simplequeue)public void listenSimpleQueue1(String msg){System.out.println(消费者1接收到消息了: msg LocalDateTime.now());}RabbitListener(queues simplequeue)public void listenSimpleQueue2(String msg) throws InterruptedException {System.err.println(消费者2接收到消息了: msg LocalDateTime.now());Thread.sleep(100);} 消息接收方明显可以看到接收者2的能力要差于接收者1 运行看看结果 运行结果我没有截全这里说一下总共发送了50次消息但是两个消费者是把50次消息平分成两部分每人处理25条消息但是由上面截图可以看到消费者2处理速度明显要慢些 这就奇了怪了明明消费者2的能力不够居然还是和消费者1处理一样的消息这是为什么呢其实是rabbitmq默认的消息预取机制造成的有多个消息来时Queue会把这些消息预先投递给两个消费者先不管能不能处理先一人一个把消息分完然后再各自处理分到的消息所以性能好的消费者早早就把消息处理完了而能力差的消费者就要慢些 那么我们怎么控制这个消息预期机制呢让能力强的消费者多处理一些消息 在配置中有个prefetch属性可以控制消息预取数量的上限 刚刚在上面的例子中没有设置那么就默认预取数量上限是无限来多少消息我就拿多少拿完再处理导致能力强和能力弱的消费者都能分到一样的数量。 那么我们将这个值设置成1每个消费者预取数量上限是1表示每个消费者最多预取1条消息处理完了再预取下一条消息这样处理消息数量的多少就完全看消费者的能力了 5.3.发布订阅模式
发布订阅模式
发布订阅模式引入了交换机的概念它可以控制消息发送给指定的队列框架模式图如下 交换机只负责消息的路由不负责消息的存储意味着一旦消息从交换机中发送失败之后消息就会丢失 5.3.1.广播
广播(Fanout Exchange) 刚刚在上面说了根据交换机的不同发布订阅模式分成了三种广播模式就是其中之一其功能是将消息发送给每一个与该交换机绑定的队列可以理解成up主和粉丝的关系up主一旦推出了新的视频那么就会推送给每一个关注了该up主的粉丝 该怎么实现
之前说过publisher只管发送消息其他的不管所以这里的交换机、队列的创建都在消费者中完成 创建一个配置类 Configuration
public class FanoutConfig {/*** 创建一个交换机* return*/Beanpublic FanoutExchange fanoutExchange(){//创建并指定交换机的名称return new FanoutExchange(itcast.fanout);}/*** 创建队列1* return*/Beanpublic Queue fanoutQueue1(){//创建并指定队列1的名称return new Queue(fanout.queue1);}/*** 创建队列2* return*/Beanpublic Queue fanoutQueue2(){//创建并指定队列2的名称return new Queue(fanout.queue2);}/*** 将队列1绑定到交换机上* param fanoutQueue1* param fanoutExchange* return*/Beanpublic Binding fanoutBinding1(Queue fanoutQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 将队列2绑定到交换机上* param fanoutQueue2* param fanoutExchange* return*/Beanpublic Binding fanoutBinding2(Queue fanoutQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
} 在配置类中我们创建了两个队列一个交换机还进行了两个队列和交换机的绑定其中调用了三个类FanoutExchange、Queue、Binding都将其注册到spring中spring就会帮我们自动进行装配当服务一启动之后spring就会将这些带有Bean注解的方法创建成bean然后由spring统一管理当spring容器启动之后会自动创建这些bean的对象还有一个点由Bean注解的方法其创建的bean的id是方法名 创建监听容器接收队列里的消息 //监听队列fanout.queue1RabbitListener(queues fanout.queue1)public void listenFanoutQueue1(String msg) throws InterruptedException {System.out.println(消费者1接收到fanout.queue1的消息了: msg);}//监听队列fanout.queue2RabbitListener(queues fanout.queue2)public void ListenFanoutQueue2(String msg){System.out.println(消费者2接收到fanout.queue2的消息了: msg);} 在publisher中发送消息 Test//给交换机FanoutExchange发送消息public void testMessageToFanoutExchange(){String exchangeName itcast.fanout;String message 你好每一个消费者;//指定发送的交换机key消息内容rabbitTemplate.convertAndSend(exchangeName,,message);} 在这里发送消息的方法里多了一个参数需要注意该参数是routingkey只不过这里是null在下一个模型中会讲到 测试 publish发送给交换机消息 然后交换机将消息路由给每一个与之绑定的队列之后comsumer再处理 5.3.2.路由
路由(Routing Exchange) 刚刚介绍的广播是将消息全部发送给与之绑定的队列那么这个路由就是将消息发送给指定的队列指定的准则就是刚刚在上面提到的routingkey 该怎么实现呢
方式一
1.创建一个配置类像5.3.1一样在其中创建队列和交换机并且进行绑定
Configuration
public class DirectConfig {//创建交换机Beanpublic DirectExchange directExchange(){return new DirectExchange(itcast.direct);}//创建队列queue1Beanpublic Queue queue1(){return new Queue(direct.queue1);}//创建队列queue2Beanpublic Queue queue2(){return new Queue(direct.queue2);}//将queue1绑定在交换机上Beanpublic Binding directBinding1(DirectExchange directExchange,Queue queue1){return BindingBuilder.bind(queue1).to(directExchange).with(blue);}//将queue2绑定在交换机上public Binding directBinding2(DirectExchange directExchange,Queue queue2){return BindingBuilder.bind(queue2).to(directExchange).with(yellow);}
} 需要注意的是在绑定队列和交换机的时候要添加上RoutingKey 2.创建监听容器 RabbitListener(queues direct.queue1)public void ListenDirectMessage1(String msg){System.out.println(blue接收到了消息:msg);}RabbitListener(queues direct.queue2)public void ListenDirectMessage2(String msg){System.out.println(yellow接收到了消息:msg);} 3.发送消息
Test
public void testMessageToDirectExchange1(){String exchangeName itcast.direct;String message 你好被指定的消费者;rabbitTemplate.convertAndSend(exchangeName,blue,message);
} 需要注意的是在发送的时候也要指定RoutingKey也就是指定交换机向哪一个队列发送消息 方式二
说是方式二其实也就是摒弃了在配置类中创建交换机、队列和绑定关系换成了在创建监听容器的时候使用注解的方式 RabbitListener(bindings QueueBinding(value Queue(name direct.queue1),exchange Exchange(name itcast.direct,type ExchangeTypes.DIRECT), //创建交换机并指定类型key {blue,red} //指定队列的key))public void ListenDirectQueue1(String msg){System.out.println(blue接收到消息了: msg);}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name itcast.direct,type ExchangeTypes.DIRECT),key {yellow,red}))public void ListenDirectQueue2(String msg){System.out.println(yellow接受到消息了: msg);} 这样的方式看起来有点乱但好在它在创建key的时候支持数组类型可以一次性创建多个key 5.3.3.话题
话题(Topic Exchange) TopicExchange与DirectExchange类似区别在于RoutingKeyTopicExchange的RoutingKey必须必须是多个单词的列表且以 . 隔开 比如 China.news China.vagetables 等等 这样把RoutingKey更加的细分像是一个话题一样因此得名TopicExchangeQueue与Exchange指定BindingKey时可以使用通配符 # 代指0个或者多个单词* 代指一个单词 比如当我给两个Queue分别指定Key叫做*.news和China.*然后我在发送消息的时候我指定Key是China.news那么就意味着这两个Queue都要接收到我发送的消息因为China.news对于上面两个Key都是符合的 6.消息转换器 刚刚在上面的例子中我们发送的消息都是String字符串类型的但其实发送消息的方法参数是Object也就是说它是支持发送对象的那么我们来试一试 这次发送一个Map类型的消息看看队列里接收到的消息是什么样的 我们可以看到我们发送的消息是被springamqp底层的消息转换器给序列化了想要具体了解的可以看看这个博主的文章 RabbitMQ发送对象之消息序列化(必踩坑的一个点)_rabbitmq传输对象序列化-CSDN博客 采用这种序列化工具性能不太好而且转换出来的字节太冗杂所以这里我们可以换一种消息转换器 引入依赖 !--Json的序列化工具--dependencygroupIdcom.fasterxml.jackson.dataformat/groupIdartifactIdjackson-dataformat-xml/artifactId/dependency
在配置类中创建这个序列化工具的实例并让spring托管 Beanpublic Jackson2JsonMessageConverter jsonMessageConverter(){return new Jackson2JsonMessageConverter();} 将这个json格式转换器的对象让spring托管之后在底层就会顶替掉原来默认的那种序列化工具现在我们再来发送一次消息看看 可以看到发送的消息已经被转换成Json格式了接收消息操作也是如此导入一样的依赖在配置类里将序列化工具托管给spring即可
