网站备案要网站做才可以使用吗,网页美工设计视频,wordpress 谷歌登陆不了,建设银行网站官网巩固基础#xff0c;砥砺前行 。 只有不断重复#xff0c;才能做到超越自己。 能坚持把简单的事情做到极致#xff0c;也是不容易的。
面试题
项目上用过消息队列吗#xff1f;用过哪些#xff1f;当初选型基于什么考虑的呢#xff1f;
面试官心理分析 第一#xff0…巩固基础砥砺前行 。 只有不断重复才能做到超越自己。 能坚持把简单的事情做到极致也是不容易的。
面试题
项目上用过消息队列吗用过哪些当初选型基于什么考虑的呢
面试官心理分析 第一你知不知道你们系统里为什么要用消息队列这个东西? 不少候选人说自己项目里用了Redis、MQ但是其实他并不知道自己为什么要用这个东西。其实说白了就是为了用而用或者是别人 设计的架构他从头到尾都没思考过。 没有对自己的架构问过为什么的人一定是平时没有思考的人面试官对这类候选人印象通常很不好。因为面试官担心你进了团队之后只会木头木脑的干呆活儿不会自己思考。 第二你既然用了消息队列这个东西你知不知道用了有什么好处坏处? 你要是没考虑过这个那你盲目弄个MQ进系统里后面出了问题你是不是就自己溜了给公司留坑你要是没考虑过引入一个技术可能存 在的弊端和风险面试官把这类候选人招进来了基本可能就是挖坑型选手。就怕你干 1年挖一堆坑自己跳槽了给公司留下无穷后 第三既然你用了MQ可能是某一种MQ那么你当时做没做过调研 你别傻乎乎的自己拍脑袋看个人喜好就瞎用了一个MQ比如Kafka甚至都从没调研过业界流行的MQ到底有哪几种。每一个MQ的优点 和缺点是什么。每一个MQ没有绝对的好坏但是就是看用在哪个场景可以扬长避短利用其优势规避其劣势。 如果是一个不考虑技术选型的候选人招进了团队leader交给他一个任务去设计个什么系统他在里面用一些技术可能都没考虑过选 型最后选的技术可能并不一定合适一样是留坑。 为什么使用消息队列 其实就是问问你消息队列都有哪些使用场景然后你项目里具体是什么场景说说你在这个场景里用消息队列是什么 面试官问你这个问题期望的一个回答是说你们公司有个什么业务场景这个业务场景有个什么技术挑战如果不用MQ可能会很麻烦但 是你现在用了MQ之后带给了你很多的好处。 先说一下消息队列常见的使用场景吧其实场景有很多但是比较核心的有3个解耦、异步、削峰
消息队列有什么优缺点
缺点有以下几个: 系统可用性降低 系统引入的外部依赖越多越容易挂掉。本来你就是 A 系统调用 BCD 三个系统的接口就好了ABCD 四个系统还好好的没啥问题你偏 加个MQ进来万一MQ挂了咋整MQ一挂整套系统崩溃你不就完了如何保证消息队列的高可用可以点击这里查看。 系统复杂度提高 硬生生加个MQ进来你怎么保证消息没有重复消费怎么处理消息丢失的情况怎么保证消息传递的顺序性头大头问题一大堆痛 苦不已。 一致性问题 A系统处理完了直接返回成功了人都以为你这个请求就成功了但是问题是要是 BCD 三个系统那里BD 两个系统写库成功了结果C 系统写库失败了咋整你这数据就不一致了。 所以消息队列实际是一种非常复杂的架构你引入它有很多好处但是也得针对它带来的坏处做各种额外的技术方案和架构来规避掉做好 之后你会发现妈呀系统复杂度提升了一个数量级也许是复杂了10倍。但是关键时刻用还是得用的。
Kafka、ActiveMQ、RabbitMQ、RocketMQ有什么优缺点 综上各种对比之后有如下建议 一般的业务系统要引入 MQ最早大家都用 ActiveMQ但是现在确实大家用的不多了没经过大规模吞吐量场景的验证社区也不是很活跃, 所以大家还是算了吧我个人不推荐用这个了。| 后来大家开始用RabbitMQ但是确实erlang语言阻止了大量的Java工程师去深入研究和掌控它对公司而言几乎处于不可控的状态但是 确实人家是开源的比较稳定的支持活跃度也高。 不过现在确实越来越多的公司会去用RocketMQ确实很不错毕竟是阿里出品但社区可能有突然黄掉的风险目前RocketMQ已捐给Apache但GitHub上的活跃度其实不算高)对自己公司技术实力有绝对自信的推荐用RocketMQ否则回去老老实实用RabbitMQ吧人家有活跃的开源社区绝对不会黄。 所以中小型公司技术实力较为一般技术挑战不是特别高用RabbitMQ是不错的选择大型公司基础架构研发实力较强用RocketMQ 是很好的选择。 如果是大数据领域的实时计算、日志采集等场景用Kafka是业内标准的绝对没问题社区活跃度很高绝对不会黄何况几乎是全世界这 个领域的事实性规范。
部署是单机还是集群呢你们高可用是怎么保证的呢
如果有人问到你MQ的知识高可用是必问的。上一讲提到MQ会导致系统可用性降低。所以只要你用了MQ接下来问的一些要点肯定就 是围绕着MQ的那些缺点怎么来解决了。 要是你傻乎乎的就干用了一个MQ各种问题从来没考虑过那你就杯具了面试官对你的感觉就是只会简单使用一些技术没任何思考,马上对你的印象就不太好了。这样的同学招进来要是做个20k薪资以内的普通小弟还凑合要是做薪资20k的高工那就惨了让你设计个系统里面肯定一堆坑出了事故公司受损失团队一起背锅。 这个问题这么问是很好的因为不能问你Kafka的高可用性怎么保证ActiveMQ的高可用性怎么保证一个面试官要是这么问就显得很没水 平人家可能用的就是RabbitMQ没用过Kafka你上来问人家Kafka干什么这不是摆明了刁难人么。 所以有水平的面试官问的是MQ的高可用性怎么保证这样就是你用过哪个MQ你就说说你对那个MQ的高可用性的理解。 RabbitMQ的高可用性 RabbitMQ是比较有代表性的因为是基于主从非分布式做高可用性的我们就以RabbitMQ为例子讲解第一种 MQ的高可用性怎么实 现。
RabbitMQ有三种模式单机模式、普通集群模式、镜像集群模式。
单机模式 单机模式就是Demo级别的一般就是你本地启动了玩玩儿的没人生产用单机模式。 普通集群模式无高可用性) 普通集群模式意思就是在多台机器上启动多个RabbitMQ实例每台机器启动一个。你创建的queue只会放在一个RabbitMQ 实例上但是每个实例都同步queue的元数据元数据可以认为是queue的一些配置信息通过元数据可以找到queue所在实例。你消费的时候实际上如果连接到了另外一个实例那么那个实例会从queue所在实例上拉取数据过来。 镜像集群模式高可用性 这种模式才是所谓的RabbitMQ的高可用模式。跟普通集群模式不一样的是在镜像集群模式下你创建的queue无论是元数据还是queue 里的消息都会存在于多个实例上就是说每个RabbitMQ节点都有这个queue的一个完整镜像包含queue 的全部数据的意思。然后每次你写消息到queue的时候都会自动把消息同步到多个实例的queue上。 那么如何开启这个镜像集群模式呢其实很简单RabbitMQ有很好的管理控制台就是在后台新增一个策略这个策略是镜像集群模式的策略指定的时候是可以要求数据同步到所有节点的也可以要求同步到指定数量的节点再次创建queue的时候应用这个策略就会自动将数据同步到其他的节点上去了。 这样的话好处在于你任何一个机器宕机了没事儿其它机器节点还包含了这个queue的完整数据别的consumer都可以到其它节点上去消费数据。坏处在于第一这个性能开销也太大了吧消息需要同步到所有机器上导致网络带宽压力和消耗很重第二这么玩儿不是分布式的就没有扩展性可言了如果某个queue负载很重你加机器新增的机器也包含了这个queue的所有数据并没有办法线性扩展你的queue。你想如果这个queue的数据量很大大到这个机器上的容量无法容纳了此时该怎么办呢
有遇到过重复消费的问题吗怎么解决的呢
其实这是很常见的一个问题这俩问题基本可以连起来问。既然是消费消息那肯定要考虑会不会重复消费能不能避免重复消费或者重复消费了也别造成系统异常可以吗这个是 MQ领域的基本问题其实本质上还是问你使用消息队列郊何保证幂等性这个是你架构里要考虑的一个问题。 回答这个问题首先你别听到重复消息这个事儿就一无所知吧你先大概说一说可能会有哪些重复消费的问题。 首先比如RabbitMQ、RocketMQ、Kafka都有可能会出现消息重复消费的问题正常。因为这问题通常不是MQ自己保证的是由我们开 发来保证的。挑一个Kafka 来举个例子说说怎么重复消费吧。 Kafka 实际上有个offset 的概念就是每个消息写进去都有一个offset代表消息的序号然后consumer 消费了数据之后每隔一段时间定时定期会把自己消费过的消息的offset 提交一下表示“我已经消费过了下次我要是重启啥的你就让我继续从上次消费到的offset来继续消费吧”。 但是凡事总有意外比如我们之前生产经常遇到的就是你有时候重启系统看你怎么重启了如果碰到点着急的直接 kill 进程了再重启。 这会导致consumer有些消息处理了但是没来得及提交offset尴尬了。重启之后少数消息会再次消费一次。 举个栗子。 有这么个场景。数据 1/2/3依次进入KafkaKafka会给这三条数据每条分配一个offset代表这条数据的序号我们就假设分配的offset 依次是152/153/154。消费者从Kafka 去消费的时候也是按照这个顺序去消费。假如当消费者消费了 offset153的这条数据刚准备去提交offset 到Zookeeper此时消费者进程被重启了。那么此时消费过的数据1/2的offset并没有提交Kafka 也就不知道你已经消费了 offset153这条数据。那么重启之后消费者会找Kafka说嘿哥儿们你给我接着把上次我消费到的那个地方后面的数据继续给我传 递过来。由于之前的offset 没有提交成功那么数据1/2会再次传过来如果此时消费者没有去重的话那么就会导致重复消费。 注意新版的Kafka 已经将offset的存储从Zookeeper 转移至 Kafka brokers并使用内部位移主题 如果消费者干的事儿是拿一条数据就往数据库里写一条会导致说你可能就把数据1/2在数据库里插入了2次那么数据就错啦。 其实重复消费不可怕可怕的是你没考虑到重复消费之后怎么保证幂等性。 举个例子吧。假设你有个系统消费一条消息就往数据库里插入一条数据要是你一个消息重复两次你不就插入了两条这数据不就错了 但是你要是消费到第二次的时候自己判断一下是否已经消费过了若是就直接扔了这样不就保留了一条数据从而保证了数据的正确性。 一条数据重复出现两次数据库里就只有一条数据这就保证了系统的幂等性。 幂等性通俗点说就一个数据或者一个请求给你重复来多次你得确保对应的数据是不会改变的不能出错。 所以第二个问题来了怎么保证消息队列消费的幂等性 其实还是得结合业务来思考我这里给几个思路 。比如你拿个数据要写库你先根据主键查一下如果这数据都有了你就别插入了update一下好吧。 比如你是写Redis那没问题了反正每次都是set天然幂等性。 比如你不是上面两个场景那做的稍微复杂一点你需要让生产者发送每条数据的时候里面加一个全局唯一的id类似订单 id之类的东西然后你这里消费到了之后先根据这个id 去比如Redis里查一下之前消费过吗如果没有消费过你就处理然后这个id写Redis。如果消费过了那你就别处理了保证别重复处理相同的消息即可。 比如基于数据库的唯一键来保证重复数据不会重复插入多条。因为有唯一键约束了重复数据插入只会报错不会导致数据库中出现脏数据。当然如何保证MQ的消费是幂等性的在实际应用中需要结合具体的业务来看。
消息队列有哪些作用
1.解耦使用消息队列来作为两个系统直接的通讯方式两个系统不需要相互依赖了 2.异步系统A给消费队列发送完消息之后就可以继续做其他事情了 3.流量削峰如果使用消息队列的方式来调用某个系统那么消息将在队列中排队有消费者自己控制消费速度
死信队列是什么延时队列是什么?
1.死信队列也是一个消息队列它是用来存放那些没有成功消费的消息的通常可以用来作为消息重试 2.延时队列就是用来存放需要在指定时间被处理的元素的队列通常可以用来处理一些具有过期性操作的业务比如十分钟内未支付则取消订单
有遇到过消息丢失吗可靠性怎么保证呢
这个是肯定的用MQ有个基本原则就是数据不能多一条也不能少一条不能多就是前面说的重复消费和幂等性问题。不能少就是说 这数据别搞丢了。那这个问题你必须得考虑一下。 如果说你这个是用MQ来传递非常核心的消息比如说计费、扣费的一些消息那必须确保这个MQ传递过程中绝对不会把计费消息给弄丢。 数据的丢失问题可能出现在生产者、MQ、消费者中咱们从RabbitMQ来分析一下吧。 生产者弄丢了数据 生产者将数据发送到RabbitMQ的时候可能数据就在半路给搞丢了因为网络问题啥的都有可能。 此时可以选择用RabbitMQ提供的事务功能就是生产者发送数据之前开启RabbitMQ事务 channel.txSelect(然后发送消息如果消 息没有成功被 RabbitMQ 接收到那么生产者会收到异常报错此时就可以回滚事务 channel.txRollback(然后重试发送消息如果 收到了消息那么可以提交事务 channel.txCommit try{ 通过工厂创建连接 connection factory.newConnection(); // 获取通道 channel connection.createChannel(); //开启事务 channel.txSelect();//这里发送消息 channel.basicPublish(exchange,routingKey,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes()) // 模拟出现异常 int result 1/0; |/ 提交事务 channel.txCommit(); }catch (IOException | TimeoutException e) //捕捉异常回滚事务 channel.txRollback();
} 但是问题是RabbitMQ事务机制同步一搞基本上吞吐量会下来因为太耗性能。 所以一般来说如果你要确保说写RabbitMQ的消息别丢可以开启confirm 模式在生产者那里设置开启 confirm模式之后你每次写 的消息都会分配一个唯一的id然后如果写入了RabbitMQ中RabbitMQ会给你回传一个ack消息告诉你说这个消息ok了。如果RabbitMQ没能处理这个消息会回调你的一个nack接口告诉你这个消息接收失败你可以重试。而且你可以结合这个机制自己在内存里维护每个消息id的状态如果超过一定时间还没接收到这个消息的回调那么你可以重发。 事务机制和confirm机制最大的不同在于事务机制是同步的你提交一个事务之后会阻塞在那儿但是confirm 机制是异步的你发送 个消息之后就可以发送下一个消息然后那个消息RabbitMQ接收了之后会异步回调你的一个接口通知你这个消息接收到了。 所以一般在生产者这块避免数据丢失都是用 confirm机制的。 已经在transaction事务模式的channel是不能再设置成confirm 模式的即这两种模式是不能共存的。
客户端实现生产者 confirm 有3种方式 1.普通confirm 模式每发送一条消息后调用 waitForConfirms(方法等待服务器端confirm如果服务端返回false或者在一段时间 内都没返回客户端可以进行消息重发。 channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingkey, MessageProperties. PERSISTENT_TEXT_PLAIN, Cor if (!channel.waitForConfirms()){ |/消息发送失败 }
2.批量confirm 模式每发送一批消息后调用 waitForConfirms(方法等待服务端confirm。 channel.confirmSelect(); for (int i 0;i batchCounti) { channel.basicPubLish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties. PERSISTENT_TEXT_PLAIN, if (!channel.waitForConfirms())f //消息发送失败 } 3.异步confirm模式提供一个回调方法服务端confirm了一条或者多条消息后客户端会回调这个方法。 SortedSet confirmset Collections.synchronizedSortedtet(new TreeSet()); channel.confirmSelect(); channel.addConfirmListener(new ConfirmListener()f public void handleAck(long deliveryTag, boolean multiple) throws IOException ( ifmultiple){ confirmset.headSet(deliveryTag 1).clear(); }else { confirmset.remove(deliveryTag); } public void handleNack(long deliveryTag, boolean multiple) throws I0Exception System.out.println(“Nack, SeqNo:” deliveryTag “, multiple:” multiple)if (multiple){ confirmSet.headSet(deliveryTag 1).clear(); 1 else { confirmset.remove(deliveryTag); }
while (true){ Long nextSeqNo channel.getNextPubLishSeqNo(); channel.basicPublish(ConfirmConfig.exchangeName,ConfirmConfig.routingKey,MessageProperties.PERSISTENT_TEXT_PLAIN, confirmSet.add(nextSeqNo);}
RabbitMQ弄丢了数据
就是RabbitMQ自己弄丢了数据这个你必须开启RabbitMQ的持久化就是消息写入之后会持久化到磁盘哪怕是RabbitMQ自己挂了恢复之后会自动读取之前存储的数据一般数据不会丢。除非极其罕见的是RabbitMQ还没持久化自己就挂了可能导致少量数据丢失但是这个概率较小。 设置持久化有两个步骤 创建queue的时候将其设置为持久化。这样就可以保证RabbitMQ持久化queue的元数据但是它是不会持久化queue里的数据的。第二个是发送消息的时候将消息的 deliveryMode 设置为2。就是将消息设置为持久化的此时RabbitMQ就会将消息持久化到磁盘上去。 必须要同时设置这两个持久化才行RabbitMQ哪怕是挂了再次重启也会从磁盘上重启恢复queue恢复这个queue里的数据。 注意哪怕是你给RabbitMQ开启了持久化机制也有一种可能就是这个消息写到了RabbitMQ中但是还没来得及持久化到磁盘上结果 不巧此时RabbitMQ挂了就会导致内存里的一点点数据丢失。 所以持久化可以跟生产者那边的 confirm 机制配合起来只有消息被持久化到磁盘之后才会通知生产者ack了所以哪怕是在持久化 到磁盘之前RabbitMQ挂了数据丢了生产者收不到ack你也是可以自己重发的。 消费端弄丢了数据 RabbitMQ如果丢失了数据主要是因为你消费的时候刚消费到还没处理结果进程挂了比如重启了那么就尴尬了RabbitMQ认为你都消费了这数据就丢了。 这个时候得用RabbitMQ提供的ack 机制简单来说 就是你必须关闭RabbitMQ的自动 aick 可以通过一个api 来调用就行然后每次你自己代码里确保处理完的时候再在程序里 ack一把。这样的话如果你还没处理完不就没有ack了那RabbitMQ就认为你还没处理完这个时候RabbitMQ会把这个消费分配给别的consumer 去处理消息是不会丢的。 为了保证消息从队列中可靠地到达消费者RabbitMQ提供了消息确认机制。消费者在声明队列时可以指定noAck参数当 noAckfalse,RabbitMQ会等待消费者显式发回ack信号后才从内存和磁盘如果是持久化消息中移去消息。否则一旦消息被消费 者消费RabbitMQ会在队列中立即删除它。
大量消息在mq里积压了几个小时了还没解决
几千万条数据在MQ里积压了七八个小时从下午4点多积压到了晚上11点多。这个是我们真实遇到过的一个场景确实是线上故障了这个时候要不然就是修复consumer的问题让它恢复消费速度然后傻傻的等待几个小时消费完毕。这个肯定不能在面试的时候说吧。一个消费者一秒是1000条一秒3个消费者是3000条一分钟就是18万条。所以如果你积压了几百万到上千万的数据即使消费者恢复了也需要大概1小时的时间才能恢复过来。 一般这个时候只能临时紧急扩容了具体操作步骤和思路如下 先修复consumer的问题确保其恢复消费速度然后将现有consumer都停掉。 ·新建一个topic,partition是原来的10倍临时建立好原先10倍的queue 数量。 然后写一个临时的分发数据的consumer 程序这个程序部署上去消费积压的数据消费之后不做耗时的处理直接均匀轮询写入临时建 立好的10倍数量的queue。 接着临时征用10倍的机器来部署consumer每一批consumer 消费一个临时queue的数据。这种做法相当于是临时将queue资源和 consumer 资源扩大10倍以正常的10倍速度来消费数据。 等快速消费完积压数据之后得恢复原先部署的架构重新用原先的consumer 机器来消费消息。 MQ中的消息过期失效了 假设你用的是RabbitMQRabbtiMQ是可以设置过期时间的也就是TTL。如果消息在queue中积压超过一定的时间就会被RabbitMQ给清 理掉这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在mq里而是大量的数据会直接搞丢。 这个情况下就不是说要增加consumer 消费积压的消息因为实际上没啥积压而是丢了大量的消息。我们可以采取一个方案就是批量重导这个我们之前线上也有类似的场景干过。就是大量积压的时候我们当时就直接丢弃数据了然后等过了高峰期以后比如大家一起喝咖啡熬夜到晚上12点以后用户都睡觉了。这个时候我们就开始写程序将丢失的那批数据写个临时程序一点一点的查出来然后重新灌入mq里面去把白天丢的数据给他补回来。也只能是这样了。 假设1万个订单积压在mq里面没有处理其中1000个订单都丢了你只能手动写程序把那1000个订单给查出来手动发到mq里去再 补一次。
MQ都快写满了
如果消息积压在mq里你很长时间都没有处理掉此时导致mq都快写满了咋办这个还有别的办法吗没有谁让你第一个方案执行的太慢了你临时写程序接入数据来消费消费一个丢弃一个都不要了快速消费掉所有的消息。然后走第二个方案到了晚上再补数据吧。 对于RocketMQ官方针对消息积压问题提供了解决方案。 1.提高消费并行度 绝大部分消息消费行为都属于10密集型即可能是操作数据库或者调用RPC这类消费行为的消费速度在于后端数据库或者外系统的吞吐量通过增加消费并行度可以提高总的清费吞吐量但是并行度增加到一定程度反而会下降。所以应用必须要设置合理的并行度。如下有几种修改消费并行度的方法 同一个ConsumerGroup下通过增加Consumer 实例数量来提高并行度需要注意的是超过订阅队列数的Consumer 实例无效。可以通过 加机器或者在已有机器启动多个进程的方式。提高单个Consumer的消费并行线程通过修改参数consumeThreadMin. consumeThreadMax实现。 2.批量方式消费 某些业务流程如果支持批量方式消费则可以很大程度上提高消费吞吐量例如订单扣款类应用一次处理一个订单耗时1s一次处理 10个订单可能也只耗时2s这样即可大幅度提高消费的吞吐量通过设置consumer 的consumeMessageBatchMaxSize 返个参数默认是1即一次只消费一条消息例如设置为N那么每次消费的消息数小于等于N。 3.跳过非重要消息 发生消息堆积时如果消费速度一直追不上发送速度如果业务对数据要求不高的话可以选择丢弃不重要的消息。例如当某个队列的消息 数堆积到100000条以上则尝试丢弃部分或全部消息这样就可以快速追上发送消息的速度。示例代码如下 public ConsumeConcurrentlyStatus consumeMessage( Listmsgs ConsumeConcurrentlyContext context)long offset msgs.get(0).getQueue0ffset();String maxoffset msgs.get(0).getProperty(Message.PROPERTY_MAX_OFFSET);Long diff Long.parseLong(max0ffset)- offset; if (diff 100000 //TODO 消息堆积情况的特殊处理 return ConsumeConcurrentyStatus.CONSUME SUCCESS // TODO正常消费过程 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; } 4.优化每条消息消费过程举例如下某条消息的消费过程如下: •根据消息从DB查询【数据1] 根据消息从DB查询【数据2】复杂的业务计算•向DB插入【数据 3]向DB插入【数据 4】 这条消息的消费过程中有 4次与 DB的交互如果按照每次 5ms 计算那么总共耗时 20ms假设业务计算耗时 5ms那么总过耗时25ms,所以如果能把4次DB交互优化为2次那么总耗时就可以优化到15ms即总体性能提高了40%。所以应用如果对时延敏感的话可以把DB部署在SSD硬盘相比于SCSI磁盘前者的RT会小很多。
如果让你写一个消息队列该如何进行架构设计说一下你的思路。
其实聊到这个问题一般面试官要考察两块: 你有没有对某一个消息队列做过较为深入的原理的了解或者从整体了解把握住一个消息队列的架构原理。 看看你的设计能力给你一个常见的系统就是消息队列系统看看你能不能从全局把握一下整体架构设计给出一些关键点出来。说实话问类似问题的时候大部分人基本都会蒙因为平时从来没有思考过类似的问题大多数人就是平时埋头用从来不去思考背后的一些东西。类似的问题比如如果让你来设计一个Spring 框架你会怎么做如果让你来设计一个Dubbc框架你会怎么做如果让你来设计一个MyBatis 框架你会怎么做 其实回答这类问题说白了不求你看过那技术的源码起码你要大概知道那个技术的基本原理、核心组成部分、基本架构构成然后参照一 些开源的技术把一个系统设计出来的思路说一下就好。 比如说这个消息队列系统我们从以下几个角度来考虑一下: 首先这个mq得支持可伸缩性吧就是需要的时候快速扩容就可以增加吞吐量和容量那怎么搞设计个分布式的系统呗参照一下kafka的设计理念broker-topic-partition每个partition放一个机器就存一部分数据。如果现在资源不够了简单啊给topic增加partition然后做数据迁移增加机器不就可以存放更多数据提供更高的吞吐量了 其次你得考虑一下这个mq的数据要不要落地磁盘吧那肯定要了落磁盘才能保证别进程挂了数据就丢了。那落磁盘的时候怎么落啊 顺序写这样就没有磁盘随机读写的寻址开销磁盘顺序读写的性能是很高的这就是kafka的思路。 其次你考虑一下你的mq的可用性啊这个事儿具体参考之前可用性那个环节讲解的kafka的高可用保障机制。多副本-leader follower-broker 挂了重新选举leader即可对外服务。 能不能支持数据0丢失啊可以的参考我们之前说的那个kafka数据零丢失方案。 mq 肯定是很复杂的面试官问你这个问题其实是个开放题他就是看看你有没有从架构角度整体构思和设计的思维以及能力。确实这个问题 可以刷掉一大批人因为大部分人平时不思考这些东西。
ActiveMQ
消息队列(第一节)
1.消息队列产品有好多种kafka、rabbitMQ 、rocketMQ 、activeMQ 等
在学习这些产品时都需要从以下几个方面来着手
1常用的API 如何发送接收消息
2如何实现MQ高可用
3MQ的集群和容错机制
4MQ的持久化
5MQ如何延迟和定时发送消息如何保证消息有序
6MQ的签收机制
7这些MQ如何和Spring、SpringBoot 整合
8这些消息队列有什么不同使用场景有那些差异
9他们是用哪些语音开发的
kafkajava、scale、rabbitMQ(erlang)、rocketMQjava、activeMQjava2.电商业务中的秒杀模块的操作
读取订单、库存检查、库存冻结、余额查询、余额冻结、订单生成、余额扣减、库存扣减、生成流水、余额解冻、库从解冻
3. activeMQ 的两个端口 61616 后台端口8161 web页面端口
4.查看后台程序是否存活
ps -ef|grep activemq | grep -v activemq
netstart -anp|grep 61616
lsof -i:616165.linux关闭防火墙命令
1查看防火墙状态
service iptables status
systemctl status firewalld
2)暂时关闭防火墙
systemctl stop firewalld
service iptables stop
3)永久关闭防火墙
systemctl disable firewalld
service iptables off
4)重启防火墙
systemctl enable firewalld
service iptables restart
5)查看版本
forewalld -cmd -version6.消息队列工作流程
1创建连接工厂
2连接工厂创建连接得到连接
3连接创建session
4session创建消息生产者或者消息消费者
5消息生产者组装消息并发送 activeMQ 使用场景(第二节)
问题引入
1.在什么情况下使用消息中间件
2.为什么要使用消息中间件解耦 系统之间接口耦合太高
异步 同步操作太费时间,例如 注册发送邮件 XXX
消峰 双十一 春运等高并发场景 activeMQ 官网地址
activeMQ Java简单实现(第三节)
两种通讯方式
点对点队列
订阅发布主题pom.xml
propertiesproject.build.sourceEncodingUTF-8/project.build.sourceEncodingmaven.compiler.source1.8/maven.compiler.sourcemaven.compiler.target1.8/maven.compiler.target/propertiesdependencies!--activemq所需要的jar包--dependencygroupIdorg.apache.activemq/groupIdartifactIdactivemq-all/artifactIdversion5.15.9/version/dependencydependencygroupIdorg.apache.xbean/groupIdartifactIdxbean-spring/artifactIdversion3.16/version/dependency!-- 下面是通用jar包--dependencygroupIdorg.slf4j/groupIdartifactIdslf4j-api/artifactIdversion1.7.25/version/dependency!-- https://mvnrepository.com/artifact/ch.qos.logback/logback-classic --dependencygroupIdch.qos.logback/groupIdartifactIdlogback-classic/artifactIdversion1.2.3/versionscopetest/scope/dependency!-- https://mvnrepository.com/artifact/org.projectlombok/lombok --dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.16.18/versionscopeprovided/scope/dependency!-- https://mvnrepository.com/artifact/junit/junit --dependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.12/versionscopetest/scope/dependency/dependencies队列生产者
package com.ttzz.activemq;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;
/*** 创建生产者* Description: * author: tangzhong* date: 2021年3月15日 下午6:31:41*/
public class ActiveMQProduceByQueue {public static String url tcp://localhost:61616;public static String queueName myQueue;public static void main(String[] args) throws JMSException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.start();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地 Queue Queue queue session.createQueue(queueName);//5. 创建生产者MessageProducer messageProducer session.createProducer(queue);//6. 发送消息for (int i 0; i 4; i) {TextMessage textMessage session.createTextMessage(queue…… i );messageProducer.send(textMessage);}//关闭资源messageProducer.close();session.close();connection.close();System.out.println(OOKK);}
}
队列消费者
有两种接收方式
同步阻塞
异步非阻塞package com.ttzz.activemq;import java.io.IOException;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;import org.apache.activemq.ActiveMQConnectionFactory;/*** 创建消费者* Description: * author: tangzhong* date: 2021年3月15日 下午6:32:57*/
public class ActiveMQConsumerByQueue {public static String url tcp://localhost:61616;public static String queueName myQueue;public static void main(String[] args) throws JMSException, IOException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.start();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地 Queue Queue queue session.createQueue(queueName);//5. 创建消费者MessageConsumer messageConsumer session.createConsumer(queue);//使用同步阻塞的方式
// while(true) {
// TextMessage textMessage (TextMessage) messageConsumer.receive();
// if(textMessage!null) {
// System.out.println(****消费者接收到消息:textMessage.getText());
// } else {
// break;
// }
// System.out.println(textMessage.getText());
// }//使用异步非阻塞的方式 监听器 messageConsumer.setMessageListener(new MessageListener() {public void onMessage(Message arg0) {TextMessage textMessage (TextMessage) arg0;if(textMessage!null) {try {System.out.println(****消费者接收到消息:textMessage.getText());} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}} try {System.out.println(textMessage.getText());} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}}});System.in.read(); //保证控制台不关//关闭资源messageConsumer.close();session.close();connection.close();System.out.println(OOKK2);}
}
主题生产者
package com.ttzz.activemq;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQProduceByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.start();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地 Topic Topic queue session.createTopic(topicName);//5. 创建生产者MessageProducer messageProducer session.createProducer(queue);//6. 发送消息for (int i 0; i 4; i) {TextMessage textMessage session.createTextMessage(myTopic…… i );messageProducer.send(textMessage);}//关闭资源messageProducer.close();session.close();connection.close();System.out.println(OOKK);}
}
主题消费者
package com.ttzz.activemq;import java.io.IOException;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQConsumerByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException, IOException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.start();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地TopicTopic queue session.createTopic(topicName);//5. 创建消费者MessageConsumer messageConsumer session.createConsumer(queue);//使用同步阻塞的方式
// while(true) {
// TextMessage textMessage (TextMessage) messageConsumer.receive();
// if(textMessage!null) {
// System.out.println(****消费者接收到消息:textMessage.getText());
// } else {
// break;
// }
// System.out.println(textMessage.getText());
// }//使用异步非阻塞的方式 监听器 messageConsumer.setMessageListener(new MessageListener() {public void onMessage(Message arg0) {TextMessage textMessage (TextMessage) arg0;if(textMessage!null) {try {System.out.println(****消费者接收到消息:textMessage.getText());} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}} try {System.out.println(textMessage.getText());} catch (JMSException e) {// TODO Auto-generated catch blocke.printStackTrace();}}});System.in.read(); //保证控制台不关//关闭资源messageConsumer.close();session.close();connection.close();System.out.println(OOKK2);}
}
消费者的三种情况 /*** 1. 先生成只启动一个消费者第1个消费者能消费吗 能* 2. 先生成先启动一个消费者再启动一个消费者第2个消费者能消费吗 no* 3. 先启动两个消费者然后再启动生成着第二个消费者可以消费吗 Y 采用轮询的方式进行消费*/topic 简介
前提
1.先启动消费者 然后再启动生成者只有订阅了才能接收到订阅的消息生成者
package com.ttzz.activemq;import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQProduceByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地 Topic Topic topic session.createTopic(topicName);//5. 创建生产者MessageProducer messageProducer session.createProducer(topic);messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);connection.start();//6. 发送消息for (int i 0; i 4; i) {TextMessage textMessage session.createTextMessage(myTopic…… i );textMessage.setStringProperty(自定义消息的key, 自定义消息的value);messageProducer.send(textMessage);}//关闭资源messageProducer.close();session.close();connection.close();System.out.println(OOKK);}
}
消费者
package com.ttzz.activemq;import java.io.IOException;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQConsumerByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException, IOException {// 1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);// 2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.setClientID(消费者2);System.out.println(消费者2);// 3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 4. 创建目的地TopicTopic topic session.createTopic(topicName);// 5. 创建消费者TopicSubscriber topicSubscriber session.createDurableSubscriber(topic, remark...); // 创建持久化的订阅connection.start();Message message topicSubscriber.receive();while (message ! null) {TextMessage textMessage (TextMessage) message;System.out.println(textMessage.getText());message topicSubscriber.receive();}session.close();connection.close();System.out.println(OOKK2);}
}
先启动两个消费者然后再启动生产者 消费者控制台 消费者1 消费者2 MQ界面
JMS 规范以及消息特性
JMS规范是什么
它是JavaEE体系中的一项Message Service常用消息中间件比较
JMS组成和特点
JMS provider
实现jms接口的消息中间件JMS Producer 、JMS Constomer JMS Message 消息头
1)jms destination 消息目的地 队列或者主题
2)jms deviverymode 持久化方式
3)jms expiration 消息过期时间
4)jms 优先级 1到4是普通消息 5-9是加急消息
5)消息id 唯一识别每个消息的标识是有MQ自己生成消息头之destination 当然也可以通过消息进行设置
四种重载目的地消息优先级存活时间是否持久化 消息的目的地队列 和 主题
持久性 消息的过期时间 默认是永不过期的
消息体
发送的消息类型有哪些
StringMessage MapMessage ByteMessage StringMessage ObjectMessage 五中类型要求发送的消息体和接受的消息体要求类型一致。要求发送的消息体和接受的消息体要求类型一致。
自定义的消息属性
自定义的消息属性能有什么还用呢
去重、识别、重点标注等 TextMessage textMessage session.createTextMessage(myTopic…… i );
messageProducer.send(textMessage);
textMessage.setStringProperty(自定义消息的key, 自定义消息的value);如何保证消息的可靠性
消息的可靠性可以从以下四个方面来回答
1消息的持久性
2消息的事务特性
3消息的签收机制
4消息持久化
队列消息的持久性 验证1 设置消息为非持久性然后生产消息服务器不关闭再去消费消息 消息被正常消费 验证2 设置消息为非持久性然后生产消息服务器关闭再去消费消息 生成出的消息 服务器关闭之后再去消费消息 消息丢失不能被消费。 刚刚生成的消息被丢失了 验证3设置消息为持久性然后生成消息。这个是刚刚生成的消息。 关闭消息服务器再次重新启动消息消息依旧存在 去消费消息。消息成功被消费
topic消息的持久性
对于topic消息持久性没有多大意义因为在topic模式中需要先启动消费者然后再启动生产者如果设置了消息持久性但是还没有启动动消费者则这些消息就会被丢失不能被消费者消费
设置持久化的topic生成者
package com.ttzz.activemq;import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQProduceByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException {//1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);//2. 创建连接Connection connection activeMQConnectionFactory.createConnection();//3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//4. 创建目的地 Topic Topic topic session.createTopic(topicName);//5. 创建生产者MessageProducer messageProducer session.createProducer(topic);messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);connection.start();//6. 发送消息for (int i 0; i 4; i) {TextMessage textMessage session.createTextMessage(myTopic…… i );textMessage.setStringProperty(自定义消息的key, 自定义消息的value);messageProducer.send(textMessage);}//关闭资源messageProducer.close();session.close();connection.close();System.out.println(OOKK);}
}
topic 消费者
package com.ttzz.activemq;import java.io.IOException;import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;import org.apache.activemq.ActiveMQConnectionFactory;public class ActiveMQConsumerByTopic {public static String url tcp://localhost:61616;public static String topicName myTopic;public static void main(String[] args) throws JMSException, IOException {// 1.获取工厂ActiveMQConnectionFactory activeMQConnectionFactory new ActiveMQConnectionFactory(url);// 2. 创建连接Connection connection activeMQConnectionFactory.createConnection();connection.setClientID(消费者1);System.out.println(topic消费者1);// 3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 4. 创建目的地TopicTopic topic session.createTopic(topicName);// 5. 创建消费者TopicSubscriber topicSubscriber session.createDurableSubscriber(topic, remark...); // 创建持久化的订阅connection.start();Message message topicSubscriber.receive();while (message ! null) {TextMessage textMessage (TextMessage) message;System.out.println(textMessage.getText());message topicSubscriber.receive();}session.close();connection.close();System.out.println(OOKK2);}
}
验证1启动一个消费者 Active Durable Topic Subscribers 处于激活状态的持久化的topic消费者 Offline Durable Topic Subscribers处于离线状态的持久化的topic消费者 启动持久化的topic生成者 topic消费者消费消息 消息服务器 验证2将消费者1 关闭启动消费者2 消费者1处于离线状态启动生成者消费 消费者2能够正常消费消息 再次启动消费者1消费者1也能正常消费消息
消息的事务特性
事务主要是针对生产者签收主要针对消费者 //3.创建会话// 第一个参数 是否开启开启事务// 第二个参数 是签收模式Session session connection.createSession(false, Session.AUTO_ACKNOWLEDGE);验证1 事务设置为 false执行发送消息消息自动到服务器 因为设置了消息的持久性关闭服务器再次重启启动该消息依旧存在
验证2 事务设置为 true执行发送消息
看看服务器有没有收到消息。服务器中没有收到刚刚发送的消息。因为没有做消息的提交操作 提交事务 消息入队 事务对于多个消息同时发送能够保证原子性 session.rollback();
为了验证的需要需要重启的时候删除持久化的消息
操作在配置文件activemq.xml的broker字段添加deleteAllMessagesOnStartup“true” 可以看到持久化的消息被删除
生成者开启事务将消息发送到消息服务器 看到消息 消费者消费掉消息 再次启动一个消费者发现消息已经被消费说明消息不能被重新消费 验证2设置消费者开启事务但是没有提交事务【消息被重复消费】。第一次消费者1正常消费消息 但是在服务器看到消息没有被消费 再次启动另一个消费者发现消息可以被多次消费
一个有趣的现象
生成者以事务的方式将生成者发送到服务器 消费者开启事务进行消费但是没有提交事务。保证控制到不灭。再次启动2号消费者发现不能重复消费 如果去掉 System.in.read();设置消费者4秒之后没有消息自动关闭。则启动2号消费者可以重复消费。 原因呢哈哈哈哈
消息的签收
非事务的签收有三种
1自动签收 Session.AUTO_ACKNOWLEDGE 2手动签收 Session.CLIENT_ACKNOWLEDGE 3允许重复消息 Session.DUPS_OK_ACKNOWLEDGE 这个我没有验证通过 对生产者而言如果开启了事务则签收机制可以随便选择事务的优先级高于签收机制 验证1生产者未开启事务采用自动签收的方式将消息发送到服务器 消费者采用手动签收发现消息可以重复消费。 消息发送到服务器运行消费者程序消息没有被消费掉 开启消息签收机制后消息不能重复消费
事务模式下的签收
生产者开启事务签收模式为自动签收将消息发送到服务器 消费者开启事务采用手动签收模式但是消息没有使用ack机制。 消息仍然被消费掉
验证 生产者开启事务签收模式为自动签收将消息发送到服务器 消费者开启事务采用手动签收模式消息使用ack机制。但是没有commit。消息能被重复消费 结论 在事务性会话中当一个事物被成功提交则消息被自动签收。如果事物回滚则消息会被再次传送。 非事物性会话中消息何时被确认取决于创建会话时的应答模式acknowledgement
activeMQ两种模式比较
1)工作模式上来说主题采用订阅发布模式如果没有订阅者消息就会被丢弃如果有多个订阅者则
就会被多个订阅者接收队列采用一对一的模式如果当前消息没有消费者则该
消息也不会丢弃如果有多个消费者那么该消息只能被一个消费者消费同时要求
消费者发送ack确认信息
2)从有无状态上来看主题是无状态的队列会默认在服务器上以文件的形式保存
也可以配置DB存储
3)从消息传递的完整性来看主题如果没有订阅者则消息会丢弃而队列不会
4) 处理效率主题会随着订阅者的增多效率减低而队列不会activemq传输协议
http://activemq.apache.org/configuring-version-5-transports.html
activemq 默认采用tcp协议 在网络传输之前序列化数据为字节流 open wire tcp协议的优点 可靠性高稳定性强效率高采用字节流的方式传递高效性、可用性支持所有平台
nio协议 autonio
// public static String url tcp://localhost:61616;
// public static String url nio://localhost:61618;public static String url autonio://localhost:5671;
