1

消息队列–经验小结

 2 years ago
source link: https://zyun.360.cn/blog/?p=2153
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
neoserver,ios ssh client

现代开发中，随处可见消息队列的身影，对于跨服务、跨语言甚至同一个业务的不同模块之间都能见到消息队列大显身手，下面就消息队列做一个应用场景、设计思路、技术实现等诸多方面做一个汇总。

1、常见的消息队列的中间件及其特点

中间件名称吞吐量特点RabbitMQ万级别社区比较活跃,功能较完备，不是分布式的，延迟低，采用erlang语言开发。ActiveMQ万级别技术成熟，会丢失消息，维护越来越少RocketMQ十万级别阿里开源，功能完备，分布式，事务支持，消息可靠，消息堆积及回溯kafka十万到百万级别功能简单，适合大数据、日志采集等，分布式的，每个partition对应一个消费者Qbus（kafka）十万级别公司自研，基于kafka定制，消息持久，分布式  

2、应用场景

  • 2.1、异步处理
  • 2.2、流量削峰
  • 2.3、模块解耦
  • 2.4、数据同步
  • 2.5、重试补偿
  • 2.6、发布订阅
  • 2.7、高并发缓冲

3、设计思路

  • 3.1 推拉模型(批量)

3.1.1 push:消息实时性高，消费者使用简单(来一个处理一个),推送速率难以适应消费速率，增加broker压力和复杂度。

3.1.2 pull:优根据需要拉取，可以批量拉取，broker相对轻松，无消息时sleep指数递增，消息延迟，资源忙等。

3.1.3 长轮询

客户端应该存在一个一直循环的程序，不停的向服务端发送获取消息请求。服务器接收到客户端请求之后，首先查看是否有数据，如果有数据则直接返回，如果没有则保持连接，等待获取数据，服务端获取数据之后，会通知之前的请求连接来获取数据，然后返回给客户端；wait+timeout。正常情况下，客户端会马上接收到服务端的数据，或者等待一段时间获取到数据；如果一直获取不到数据，会有超时处理；在获取数据或者超时处理之后会关闭连接，然后再次发起长轮询请求。RocketMQ 和 Kafka 都是利用“长轮询”来实现拉模式(推拉结合，兼容推的优点)。一句话说就是消费者和 Broker 相互配合，拉取消息请求不满足条件的时候 hold 住，避免了多次频繁的拉取动作，当消息一到就提醒返回。

  • 3.2 存储方案优劣

内存 > 文件系统 > 分布式kv > 分布式文件系统 > 数据库

  • 3.3 消费模型

单播vs多播，集群vs广播，1对1vs1对n，队列vsTopic

  • 3.4 队列特性

3.4.1 消息丢失–引入消息确认机制，生产者->队列，队列->消费者，队列持久化

3.4.2 消息重复–唯一key，数据库唯一主键，redis唯一，分布式ID，队列messageID

3.4.3 消息顺序–单线程生产者+单线程消费者+每个消费线程对应一个单独队列, 排除消息丢失的情况，可以做到严格有序

4、引入消息队列的缺点

  • 4.1 复杂性增加：重复消费、丢失消息、消费乱序
  • 4.2 数据一致性：

4.2.1 TCC补偿事务：三个接口，保证强一致性，对业务侵入比较大。

4.2.2 消息队列+本地消息表：本地事务+状态表+MQ，最终一致性。

4.2.3 支持事务消息的消息队列：提供类似XA功能，提供消息提交，消息回查功能check

  • 4.3 可用性：如果消息队列崩了，会导致其他服务都崩了

5、常见问题

  • 5.1 如何保证消息不会被重复消费（幂等性）

原因：这里有个偏移量的概念，消息者每次消费了之后都会进行提交，用来告诉MQ我消费道哪了，不过提交的时候并不是说消费一条就马上提交，而是消息了一定的量才提交，如果这时候消息了一些还没提交，然后挂了。之后MQ就会重复发这些被消费的数据给它，导致重复消息。

解决方法：

a、强校验(MySql)：写数据的时候，根据主键查一下是否有该数据了，如果有，则只需要更新一下。

b、如果是写redis的字符串，则没事，天然幂等性 c、弱校验(Redis)：每次发消息的时候加一个全局的id，然后消费者每次消费之后存放再redis的set里，每次要消费的时候查询消费过没有。

  • 5.2 如何保证消息按照顺序执行

Rabbitmq：

原因：一个queue多个消费者导致顺序出错。

解决方法：一个queue对应一个消费者，或者消费者内部弄一些内存队列，然后分发给不同层的woker处理。

Kafka：

原因：虽然一个partition只有一个消费者，但消费者内部多个线程消费，导致不一致。

解决方法：

a、单线程；

b、或者消费者内部维护一些队列，然后不同的key存放道不同的队列，之后让一个线程只负责一个队列。

  • 5.3 如何保证消息不丢失

RabbitMQ:

场景1：再生产者发消息过程中网络丢失了。

解决方法:

a、可以用rabbitmq提供的事物机制来解决了，通过channel.txSelect来开启，不过事务机制是同步的，性能太差；

b、通过消费者自己发送确认，即confirm机制来解决。该函数接收成功会回调发送ack，否则发送nack。

场景2：MQ自己弄丢失了，比如重启。

解决方法：

a、通过持久化来解决；

b、如果还没持久化就挂了，则还是会丢失，所以可以和confirm机制进行配合。

场景3：消费者弄丢了

解决方法：关闭自动提交，而是自己来提交，等消息处理完毕再提交。

Kafka：

场景1：消费端弄丢

解决方法：关闭自动提交即可，当然这会导致消息重复消费，自己保证幂等性即可。

场景2：kafka弄丢了（例如主leader挂了，但是数据还没有同步，此时重新选举follwer为主节点，然后同步数据，导致丢失）

解决方法：只有等leader全部同步成功之后才进行确认。

  • 5.4 消息积压了几千万怎么办

·先修复现有的consumer，然后停止工作。

·新建一个topic，之后弄10个partition和10个consumer。

·弄一个消息分发程序，把原来的消息分发到10个queue上。

·弄10个消费者来消费这10个queue。

·相当于用了10倍的速度，等消费完之后再恢复到原来的配置。

  • 5.5 消息过期了怎么办？

·不要设置过期时间

·重新批量导入过期的消息

  • 5.6 MQ写满了怎么办？

·只能丢消息了，一边丢一边记录这些消息，之后等高峰期过了再来补这些数据

总结

以上是作者日常开发中零零散散总结的经验，有些具体的实施方案没有罗列出来，因为具体的实施方案是根据业务场景灵活制定的。

以上方案可能不适合某些具体场景，望读者还能依据实际，从现实出发，制定适合自身业务的实施方案，部分内容可能描述不够准确或有遗漏，望读者提出意见和建议。


report

Recommend

About Joyk


Aggregate valuable and interesting links.
Joyk means Joy of geeK