java中的消息队列_java 消息队列

消息队列：可以看做是一个存储消息的容器，它是分布式系统中的重要组件之一。
目的是：
1、为了通过异步处理来提高系统的性能来减少系统响应的时间

一般的步骤是客户端发起请求给服务端，服务端在请求给数据库，数据库在返回结果给服务端，最后客户端在接收服务端的响应。此时运用消息队列先是客户端发送请求给服务端，服务端发送消息给消息队列，客户端接收服务端的响应，在消息队列请求数据库。因此在使用消息队列时需要考虑业务的流程进行异步处理。

2、削峰/限流

将短时间高并发产生的事务消息存储在消息队列中，然后后端服务再根据自己能力慢慢消费这些消息，避免后端服务被击垮。

3、降低系统耦合性。

模块之间如果不存在直接调用，那么新增模块修改模块时对其他模块影响就比较小，对系统的扩展性更好。消息队列运用在：生产者也就是客户端发送消息到消息队列中，接受者也就是服务端处理消息就直接从消息队列中获取，不与其他系统耦合，这样也提高了系统的扩展性。另外为了避免消息队列服务器发生宕机而造成消息丢失，会将成功发送到消息队列的消息存储在消息生产者服务器上，等消息真正被消费者服务器处理后才删除消息，在消息队列服务器宕机时生产者服务器会选择分布式消息队列服务器集群中的其他服务器发送消息。

使用消息队列带来的一些问题：

系统可用性降低：系统在某种程度上可用性降低，会担心消息丢失问题
系统复杂性提高：加入消息队列你需要担心消息没有被重复消费、需要处理消息被丢失的情况、保证消息被传递的顺序性。
一致性问题：异步的确可以提高响应速度，但是消费者未正确消费可能会导致数据不一致问题。

JMS 与AMQP:
JMS是java的消息服务，JMS的客户端之间可以通过JMS服务进行异步的消息传输。JMS API时一个消息服务的标准、规范。ActiveMQ就是基于JMS规范实现的。
JMS两种消息模型：

1、点到点（P2P）模型
队列模型，生产者发送消息给消息队列，一条消息只能被一个消费者使用，未被消费的消息在队列中保留直到被消费或超时。
2、发布订阅模型
使用主题Topic作为消息通信载体，类似于广播模式；发布者发布一条消息，该消息通过主题传递给所有订阅者，在一条消息广播之后才订阅的用户时收不到该条信息的。

AMQP:
一个提供统一消息服务的应用层标准高级消息队列协议（二进制应用层协议），是应用层协议的开放标准，为面向消息的中间件设计，兼容JMS。基于此协议的客户端可与消息中间件可传递消息，并不受客户端/中间件同产品，不同开发语言等条件限制。RabbitMQ就是基于AMQP协议实现的。

总结：AMQP为消息定义了线路层的协议，而JMS所定义的是API规范，在Java体系中，多个客户端可以通过JMS进行交互，不需要应用修改代码，但是其跨平台的支持较差。kafka唯一一点劣势是有可能消息重复消费。

Kafka:
分布式流式处理平台，关于流平台具有三个关键功能：

1、消息队列：发布和订阅消息流，这个功能类似于消息队列，这也是Kafka归类为消息队列的原因。
2、容错的持久方式存储记录消息流：Kafka会将消息持久化到磁盘，有效避免消息丢失的风险。
3、流失处理平台：在消息发布的时候进行处理，Kafka提供了一个完整的流失处理类库。

应用场景：

1、消息队列：建立实时流数据通道，以可靠的在系统或应用程序之间获取数据
2、数据处理：构建实时的流数据处理程序来转换或处理数据流。

与其消息队列相比优势：

1、性能，设计中大量使用了批量处理和异步的思想，最高可以每秒处理千万级别的消息。
2、兼容性，与周边生态系统兼容度最好

队列模型、发布订阅模型
RocketMQ的消息模型跟Kafka基本是完全一样的，唯一区别是Kafka中没有队列这个概念，与之对应的是分区Partition.

Producer(生产者)产生消息的一方，Consumer（消费者）消费消息的一方，Broker（代理）可以看作是一个独立的Kafka实例。多个KafkaBroker组成一个Kafka Cluster.

Kafka的多副本机制：

Kafka为分区（Partition）引入了多副本（Replica）机制。分区中的多个副本之间会有一个叫leader的，其他副本称为follower.发送的消息会被发送到leader副本，然后follower副本才能从leader副本中拉去消息进行同步。生产者和消费者只与leader副本交互，其他副本只是leader副本的拷贝，它们的存在只是为了保证消息存储的安全性。当Leader副本发生故障时从follower中选举出一个leader。

多分区以及多副本机制好处：
Kafka通过给特定Topic指定多个Patition,而各个Partition可以分布在不同的Broker上，方便提供较好的并发能力（负载均衡）。
Partition可以指定对于的Replica数，极大提供消息存储的安全性，提高了容灾能力，不过也增加了所需要的存储空间。

如果保证Kafka中消息消费的顺序，方法：

1、1个Topic只对应一个Partition.
2、发送消息的时候指定key/Partition.

Kafka如何保证消息不丢失：

生产者调用send方法发送消息之后，消息可能因为网络问题并没有发送过去。可以用get()方法调用结果，但是这样也让它变成了同步操作。还可以为Producer的retries（重试次数）设置一个合理的值。
消费者丢失消息的情况：手动关闭偏移量offset使每次真正消费完消息之后在手动提交offset.但是该方法会带来消息重复消费的问题。
Kafka弄丢消息：设置acks = all本来默认值为1，当配置为all时代表所有副本都要接收到该消息之后，该消息才算真正成功被发送。设置replication.factor>=3,保证每个分区至少有3个副本，设置min.insync.replicas>1,消息至少要写入到2个副本才算成功发送，一般设置replication.factor = min.insync.replicas+1;

Kafka如何保证消息不重复消费：

服务端侧已经消费的数据没有成功提交offset
kafka侧由于服务端处理业务时间长或者网络链接等等原因让kafka认为服务假死，出发了分区rebalance。
解决方案：消费消息服务做幂等校验。将enable.auto.commit参数设置为false，关闭自动提交，手动提交offset。

RocketMQ:

一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式的特点。采用Java语言开发的分布式消息系统

对于主题模型的实现来说每个消息中间件的底层设计都是不一样的，比如Kafka中的分区，RocketMQ中的队列，RabbitMQ中的Exchange。主题模型/发布订阅模型就是一个标准，这些中间件都是照着这个标准去实现而已。

Producer Group 生产者组,Consumer Group 消费者组,Topic 主题,RocketMQ通过使用一个Topic中配置多个队列并且每个队列维护每个消费者的消息位置实现了主题模式/发布订阅模式。

在RocketMQ中使用的是事务消息加上事务反查机制来解决事务问题的。

RabbitMQ：

采用Erlang语言实现AMQP的消息中间件，具体特点可靠性、灵活的路由、扩展性、高可用性、支持多种协议、多语言客户端、易用的管理界面、插件机制

消息 一般由2部分组成：消息头或者说Label和消息体payLoad,消息体是不透明的，而消息头由一系列的可选属性组成，属性包括routing-key路由键、priority相对于其他消息的优先权、delivery-mode该消息可能需要持久性存储。生产者把消息发给RabbitMQ后，RabbitMQ会根据消息头把消息发送给感兴趣的消费者。消息不是直接投递到Queue消息队列中的，必须还经过交换器Exchange，它一般会制定一个路由键，用来指定这个消息的路由规则，经过它过后再分配到队列中。如果路由不到，或许会返回给生产者，或许被丢弃掉。交换器由4种类型，不同类型对应着不同的路由策略：direct默认、fanout、topic、headers，RabbitMQ中通过Binding绑定将交换器和消息队列关联起来。

交换器SpringAMQP-FanoutExchange的作用：
1、接收publisher发送的消息
2、将消息按照规则路由到与之绑定的队列
3、不能缓存消息，路由失败，消息丢失
4、FanoutExchange的会将消息路由到每个绑定的队列
声明队列、交换机、绑定关系的Bean是什么
1、Queue
2、FanoutExchange
3、Binding

交换机SpringAMQP-DirectExchange发布订阅：会将接收到的消息根据路由规则路由到指定的Queue，因此称为路由模式（routes)
1、每个Queue都与Exchange设置一个BindingKey
2、发布者发送消息时，指定消息的RoutingKey
3、Exchange将消息路由到BindingKey与消息RoutingKey一致的队列
4、如果多个队列具有相同的RoutingKey,则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机有哪些常用注解
@Queue
@Exchange
1
2
3

交换机SpringAMQP-TopExchange发布订阅：与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且与.分割

Queue与Exchange指定BindingKey时可以使用通配符：
#：代指0或多个单词
*：代指一个单词
1
2
3

SpringAMQP:
SpringAMQP如何发送消息：
1、引入amqp的starter依赖
2、配置RabbitMQ依赖
3、利用RabbitTemplate的convertAndSend方法
SpringAMQP如何接收消息：
1、引入amqp的starter依赖
2、配置RabbitMQ地址
3、定义类，添加@Component注解
4、类中声明方法，添加@RabbitListener注解，方法参数就是消息
注意：消息一旦消费就会从队列中删除，RabbitMQ没有消息回溯功能