都 2020 年&＃xff0c;应该没有项目不用消息队列的吧&＃xff1f;消息队列解决了哪些问题&＃xff0c;条件反射应该都能回答“异步消息”、“应用解耦”和“流量削峰”和。

消息队列使用场景

1、异步处理

比如 &＃64;Python大星 之前做的基金项目&＃xff0c;当用户购买成功后&＃xff0c;有如下几个操作&＃xff1a;

① 保存订单信息&＃xff1b;② 短信通知用户&＃xff1b;③ 当购买金额大的时候邮件通知销售人员&＃xff1b;④ 用户积分增加&＃xff0c;可兑换奖品或者优惠券。

同步处理这些请求&＃xff0c;用户体验差。当我们对接口的 Rt(ResponseTime 响应时间)有要求时&＃xff0c;需要异步处理比如短信、邮件和积分问题。

2、应用解耦

比如线下活动购物 POS 机消费&＃xff0c;如果消费满足一定条件&＃xff0c;就进行微信推送引导用户参与抽奖。当多部门协助时&＃xff0c;其他部门负责 POS 机消费信息推送&＃xff0c;本部门负责接收信息处理相应的业务逻辑。

3、流量削锋

一般在秒杀或团抢活动中使用广泛。用户的请求&＃xff0c;服务器接收后&＃xff0c;首先写入消息队列。假如消息队列长度超过最大数量&＃xff0c;则直接抛弃用户请求或跳转到错误页面。秒杀业务根据消息队列中的请求信息&＃xff0c;再做后续处理。

MQ 选型

面试中&＃xff0c;考官经常会问你技术选型的问题&＃xff0c;比如消息中间件&＃xff0c;市场上种类繁多&＃xff0c;如何选择一个合适的呢&＃xff1f;

咨询了很多开发小伙伴&＃xff0c;有的用 kafka、有的用 rocketMQ、还有的用 rabbitMQ。没有好坏之分&＃xff0c;只有适不适合。

图 1

根据图 1 中&＃xff0c;我们可以综合几个方面来选型&＃xff1a;

① 社区活跃性。

万一哪天这个 mq 社区没人维护&＃xff0c;出现 bug&＃xff0c; 怎么办&＃xff0c;重新再选一个 MQ 吗&＃xff1f;从图中可以看出 ActiveMQ 活跃度很低。

② 单机吞吐量。

目前 ActiveMQ、RabbitMQ 吞吐量在万级&＃xff0c;而 RocketMQ 和 kafka 在 10 万级&＃xff1b;

③ 时效性

ActiveMQ、RocketMQ 和 kafka 延迟在 ms 级别&＃xff0c;RabbitMQ 延迟在 ms 级别&＃xff1b;

④ 可用性

ActiveMQ、RabbitMQ 基于主从复制&＃xff0c;高可用&＃xff1b; RocketMQ 和 kafka 非常高&＃xff0c;分布式架构&＃xff1b;

⑤ 消息可靠性

RocketMQ 有较低概率丢失数据&＃xff0c;RabbitMQ 基本不丢&＃xff0c;RocketMQ 和 kafka 可以做到 0 丢失&＃xff1b;

⑥ 功能扩展

kafka 功能较简单&＃xff0c;日志采集 nb&＃xff0c;其他是三个不分上下。

kafka

把消息放到【队列】里&＃xff0c;我们叫【消息生产者】 producer&＃xff0c;把 消息从队列里取出来&＃xff0c;我们叫【消息消费者 】consumer。

1、consumer 如何区分 producer 里的消息&＃xff1f;&＃xff1f;&＃xff1f;

通常&＃xff0c;我们在使用消息队列时&＃xff0c;n to n&＃xff0c;多个 producer 和 consumer&＃xff0c;如何进行区分&＃xff0c;这里使用了 topic 的概念&＃xff0c;给 队列取个响亮的名字。

2、kafka 如何保证吞吐量&＃xff1f;&＃xff1f;&＃xff1f;

从上面图 1 中我们看出&＃xff0c;kafka 在 topic 从几十到几百个时&＃xff0c;吞吐量会大幅度下降&＃xff0c;尽量保证 topic 数量不要过多。

比如说&＃xff1a;n 个 producer 往 topic 里塞消息&＃xff0c; n 个 consumer 从 topic 里取消息。

为了提高吞吐量&＃xff0c;kafka 使用 partion 分区的概念(和 Mysql 分区类似)

3、kafka 如何保证高可用

当然是集群了&＃xff0c;用它&＃xff0c;用它&＃xff0c;用它。

在 kafka 集群中&＃xff0c;我们把每个 kafka 服务器叫“broker”。

比如说我们这 3 台 broker&＃xff0c;每台里有 3 个 partion&＃xff0c;当 producer 生产消息时&＃xff0c;分别进入 3 台 broker 的一个分区中(作为主分区)&＃xff0c;每台 broker 里剩余的 2 个 partion 只做备份 其他 2 个 broker 里的消息(备份分区)。 你品&＃xff0c;你品&＃xff0c;你细品。当某个 broker 挂了&＃xff0c;选举其他 broker 里的 partion 来做主分区&＃xff0c;实现高可用。

4、消费者如何保证消费消息

假设多个消费者 consumer 组团去消费分区里的消息&＃xff0c;一个 consumer 对应 一个 分区&＃xff0c;假设有一个 consumer 挂了&＃xff0c;必然另一个 consumer 要去接盘&＃xff0c;但是我得知道挂掉的 consumer 消费进度。

kafka 在这里使用了 offset 的概念&＃xff0c;在以前版本的 Kafka&＃xff0c;这个 offset 是由 Zookeeper 来管理的&＃xff0c;后来 Kafka 开发者认为 Zookeeper 不合适大量的删改操作&＃xff0c;于是把 offset 在 broker 以内部 topic (__consumer_offsets) 的方式来保存起来。

消息队列引发的问题

1、如何保证消息的幂等性&＃xff1f;&＃xff1f;&＃xff1f;

① 数据库层面使用唯一性索引&＃xff1b;

② 每次先查再保存到数据库&＃xff1b;

2、如何保证消息的可靠性传输&＃xff1f;&＃xff1f;&＃xff1f;

集群

3、如何保证消息的顺序性&＃xff1f;&＃xff1f;&＃xff1f;

生产者在写的时候&＃xff0c;其实可以指定一个 key&＃xff0c;比如说我们指定了某个订单 id 作为 key&＃xff0c;那么这个订单相关的数据&＃xff0c;一定会被分发到同一个 partition 中去&＃xff0c;而且这个 partition 中的数据一定是有顺序的。

写 N 个内存 queue&＃xff0c;具有相同 key 的数据都到同一个内存 queue&＃xff1b;然后对于 N 个线程&＃xff0c;每个线程分别消费一个内存 queue 即可&＃xff0c;这样就能保证顺序性。

&＃64;Python大星 | 文