分布式消息中间件及RabbitMQ

分布式应用和集群：

　　从部署形态来看，它们都是多台机器或者多个进程部署，而且都是为了实现一个业务功能。

　　如果是一个业务被拆分成多个子业务部署在不同的服务器上，那就是分布式应用

　　如果是同一个业务部署在多台服务器上，那就是集群

分布式应用的多个子系统之间并不是完全独立的，它们需要相互通信来共同完成某个功能。系统间通信的方式有两种，一种是远程过程调用即RPC接口，

另一种是基于消息队列的方式。

　　基于消息队列的方式是指由应用中的某个系统负责发送消息，由订阅这条消息的相应系统负责接收消息。不同的系统在收到消息后进行各自系统内的业务处理。消息可以非常简单，比如只包文本字符串；也可以很复杂，比如包含字节流、字节数组，还可能嵌入对象，甚至是经序列化后的Java对象。消息生产者在发送消息后可以立即返回，由消息队列来负责消息的传递，消息发布者只管将消息发布到消息队列而不用管谁来取，消息消费者只管从消息队列中取消息而不用管是谁发布的，这样生产者和消费者都不用知道对方的存在。

消息队列使用的典型场景是异步处理，同时还可用于解耦、流量削峰、日志收集、事务最终一致性等问题。

消息队列的特点：

Broker：至少需要包含消息的发送、接收和暂存功能，另外，在不同的业务场景中，需要消息队列能解决诸如消息堆积、消息持久化、可靠投递、消息重复、严格有序、集群等各种问题。

　　1、消息堆积：消息消费者处理速度跟不上生产者发送消息的速度，造成消息堆积，所以消息队列要能够处理这种情况，比如设置阀值，将超过阀值的消息不再放入处理中

　　　　心、设置消息过期时间，以防止系统资源被耗尽导致整个消息队列不可用。

　　2、消息持久化：消息处理中心如果在接收到消息之后不做任何处理就直接转给消费者，那就无法满足流量削峰等需求。索引消息处理中心要能先把消息暂存下来，然后选择

　　　　合适的时机将消息投递给消费者。

　　3、可靠投递：可靠投递是不允许存在消息丢失的情况的

　　4、消息重复：当消息发送失败或者不知道是否发送成功时（比如超时），消息的状态时待发送，定时任务会不停的轮询所有的待发送消息，最终保证消息不会丢失，但是带

　　　　来了消息可能会重复的情况

　　5、严格有序：实际的业务场景中，会有需要按生产消息时的顺序来消费的情形，这就需要消息队列能够提供有序消息的保证。

　　6、集群：消息队列产品要提供对集群模式的支持

　　7、消息中间件：非底层操作系统软件，非业务软件，不是最终给用户使用的额，不能直接给客户端带来价值的软件统称为中间件。介于用户应用和操作系统之间的软件。

　　　　消息中间件关注于数据的发送和接收，利用高效、可靠的异步消息传递机制集成分布式系统。

　　消息协议

　　消息协议是指用于实现消息队列功能时所涉及的协议。按照是否向行业开放消息规范文档，可以将消息协议分为开放协议和私有协议。常见的开放协议有AMQP、MQTT、STOMP、XMPP等，有些特殊框架Redis、Kafka、ZeroMQ根据自身需要位严格遵循MQ规范，而是基于TCP/IP自行封装了一套协议，通过网络Socket接口进行传输，实现了MQ的功能。

这里的协议可以简单地理解成双方通信的一个约定，比如传过来一段字符流数据，其中第一个字节表示什么，第二个字节表示什么。

AMQP：Advanced Message Queuing Protocol，一般来说将AMQP协议的内容分为三个部分，基本概念、功能命令和传输层协议

　　基本概念：AMQP内部定义的各组件及组件的功能说明

　　功能命令：指该协议定义的一系列命令，应用程序可以基于这些命令来实现相应的功能

　　传输层协议：定义了数据的传输格式，消息队列的客户端可以基于这个协议与消息代理和AMQP的相关模型进行交互通信，该协议内容包括数据帧处理、信道复用、内容编码、心跳检测

　　　　　　、数据表示和错误处理等。

1、主要概念

　　Message：消息，消息服务器所处理数据的原子单元

　　Publisher：消息生产者，也是一个向交换器发布消息的客户端应用程序

　　Exchange：交换器，用来接收消息生产者所发送的消息并将这些消息路由给服务器中的队列

　　Binding：绑定，用于消息队列和交换器之间的关联，一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

　　Virtual Host：虚拟主机，它是消息队列以及相关对象的集合，是共享同一个身份验证和加密环境的独立服务器域，每个虚拟主机本质上都是一个mini版的消息服务器，拥有自己的队列、交换器、绑定和权限机制

　　Broker：消息代理，表示消息队列服务器实体，接受客户端连接，实现AMQP消息队列和路由功能的过程

　　Routing Key：路由规则，虚拟机可用它来确定如何路由一个特定消息

　　Queue：消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可被投入一个或多个队列中。消息一直在队列里面，等待消费者连接到这个队列将其取走

　　Connection：连接，可以理解成客户端和消息队列服务器之间的TCP连接

　　Channel：信道，仅仅当创建了连接后，若客户端还是不能发送消息，则需要为连接创建一个信道。信道是一条独立的双向数据流通道，它是建立在真实的TCP连接内的虚拟连接，AMQP命令都是通过信道发送出去的，不管是发布消息、订阅队列还是接收消息，它们都是信道完成。一个连接可以包含多个信道，之所以需要信道，是因为TCP连接的建立和释放都是十分昂贵的，如果客户端的每一个线程都需要与消息服务器交互，如果每一个线程都建立了一个TCP连接，不仅浪费资源，而且操作系统也无法支持每秒建立如此多的连接

　　Consumer：消息消费者，表示一个从消息队列中取得消息的客户端应用程序

2、核心组件生命周期

（1）消息的声明周期：一条消息消息的流转过程是这样的：Publisher产生一条数据，发送到Broker，Broker中的Exchange可以被理解为一个规则表（Routing Key和Queue的映射关系-Binding），Broker收到消息后根据Routing key查询投递的目标Queue。Consumer向Broker发送订阅消息时会指定自己监听哪个Queue，当有数据到达Queue时Broker会推送数据到Consumer。

　　生产者Publisher在发布消息时可以给消息指定各种消息属性，其中有些属性有可能会被消息代理Broker所使用，而其他属性则是完全不透明的，它们只能被接收消息的应用所使用。

　　当消息到达服务器时，服务器中的交换器通常会将消息路由到服务器上的消息队列中，如果消息不能路由，则交换器会将消息丢弃或者将其返回给消息生产者，这样生产者可以选择如何来处理未路由的消息。

　　单条消息可存在于多个消息队列中，消息代理可以采用复制消息等多种方式进行处理。但是当一条消息被路由到多个消息队列中时，它在每个消息队列中都是一样的。

　　当消息到达消息队列时，消息队列会立即尝试将消息传递给消息消费者。如果传递不成功，则消息队列会存储消息（按生产者要求存储在内存或者磁盘中），并等待消费者准备好。

　　如果没有消费者，则消息队列通过AMQP将消息返回给生产者（如果需要的话）。当消息队列把消息传递给消费者后，它会从内部缓冲区中删除消息，删除动作可能是立即发生的，也可能在消费者应答已成功处理之后再删除。

　　消息消费者可选择如何及何时来应答消息，同样，消费者也可以拒绝消息（一个否定应答）。

（2）交换器的生命周期：每台AMQP服务器都预先创建了许多交换器实例，它们在服务器启动时就存在并且不能被销毁。如果你的应用程序有特殊要求，则可以选择自己创建交换器，并在完成工作后进行销毁

（3）队列的生命周期：队列分为持久化消息队列和临时消息队列。

　　持久化消息队列可被多个消费者共享，不管是否有消费者接收，它们都可以独立存在

　　临时消息队列对某个消费者是私有的，只能绑定到此消费者，当消费者断开连接时，该消息队列将被删除

Cosumer工作原理：

　　应用通过监听队列中的消息，获取queue中的message即可消费（Broker推送）

注意事项：

　　1）：没有消费者的Queue的message是无法被消费的，这个queue中的message就会一直存在

　　2）：一个Queue可以拥有多个消费者，也可以注册一个独享消费者，注册独享消费者的Queue中的消息只有指定的消费者可以消费message

　　3）：消费者消费完消息会发个发个反馈给Queue，这个Queue就会将这条message从Queue中移除，如果没有接收到反馈那么Queue就会一直

　　　　存在这条message，同时这个message如果不能被消费那么就会造成Queue中的消息堵塞

Message的主要属性

　　Content type：内容类型

　　Content encoding：内容编码

　　Routing key：路由键　　

　　Delivery mode(persistent or not)：投递模式（持久化或非持久化）

　　Message priority：消息优先权

　　Message publishing timestamp：消息发布的时间戳

　　Expiration period：消息有效期

　　Publisher application id：发布应用的ID

　　注意事项：

　　　　（1）：消息是以byte字节的形式存在

　　　　（2）：Content type可以存放一些header和argument属性（和Http Request类似）

　　　　（3）：有些内容例如中文，需要指定编码

　　　　（4）：Delivery mode设置成持久化模式可以将消息保存到硬盘，在服务器重启后会读取硬盘中未被消费的message，此举会保证

　　　　　　消息的健壮性但是会造成性能牺牲。

 3、功能命令

AMQP协议文本是分层描述的，在不同主版本中划分的层次是有一定区别的。

0-9版本共分两层：Function Layer(功能层)和Transport Layer（传输层）。

　　功能层定义了一系列命令，这些命令按功能逻辑组合成不同的类（Class），客户端应用可以利用它们来实现自己的业务功能。

　　传输层将功能层所接收的消息传递给服务器经过相应处理后再返回，处理的事情包括信道复用、帧同步、内容编码、心跳检测、数据表示和错误处理等

0-10版本则分为三层：Model Layer（模型层）、Session Layer（会话层）和Transport Layer（传输层）。

　　模型层定义了一套命令，客户端应用利用这些命令来实现业务功能

　　会话层将负责将命令从客户端应用传递给服务器，再将服务器的响应返回给客户端应用，会话层为这个传递过程提供了可靠性、同步机制和错误处理。

　　传输层负责提供帧处理、信道复用、错误检测和数据表示

4、消息的数据格式

　　所有的消息数据都被组织成各种类型的帧（Frame）。帧可以携带协议方法和其他信息，所有帧都有同样的格式，都有一个帧头（header，7个字节）、任意大小的负载（payload）和

一个检测错误的结束帧（Frame-end）字节组成。

　　其中帧头包括一个type字段、一个channel字段和一个size字段；帧负载的格式依赖帧类型（type）

　　要读取一个帧需要三步、

　　1、读取帧头，检查帧类型和通道（channel）

　　2、根据帧类型读取帧负载并进行处理

　　3、读取结束帧字节

AMQP定义了如下帧类型。

　　type=1,"METHOD":方法帧；

　　type=2,"HEADER":内容头帧；

　　type=3,"BODY":内容体帧；

　　type=4,"HEARTBEAT":心跳帧。

通道编号为0的代表全局链接中的所有帧，1~65535代表特定通道的帧。size字段是指帧负载的大小，它的数值不包括结束帧字节。

AMQP使用结束帧来检测错误客户端和服务端实现引起的错误。

　　JMS

　　Java Message Service，即Java消息服务应用程序接口，是Java平台中面向消息中间件的一套规范的Java API接口。用于在两个应用程序之间或分布式系统中发送消息，进行异步通信。

JMS不是消息队列协议中的一种，更不是消息队列产品，它是与具体平台无关的API，目前市面上的绝大多数消息中间件厂商都支持JMS接口规范。换句话说，你可以使用JMS API来连

接支持AMQP、STOMP等协议的消息中间件产品。在这一点上和JDBC很像。

　　RabbitMQ

　　RabbitMQ是一个由Erlang语言开发的基于AMQP标准的开源实现。RabbitMQ最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

其具体特点包括：

　　1、保证可靠性（Reliability）。RabbitMQ使用一些机制来保证可靠性，如持久化、传输确认、发布确认等

　　2、具有灵活的路由（Flexible Routing）功能。在消息进入队列之前，是通过Exchange（交换器）来路由消息的。对于典型的路由功能，RabbitMQ已经提供了一些内置的Exchange来

　　　　实现。针对更复杂的路由功能，可以将多个Exchange绑定在一起，也可以通过插件机制来实现自己的Exchange。

　　3、支持消息集群（Clustering）。多台RabbitMQ服务器可以组成一个集群，形成一个逻辑Broker。

　　4、具有高可用性（Highly Available）。队列可以在集群中的机器上进行镜像，使得在部分节点出现问题的情况下仍然可用。

　　5、支持多种协议（Multi-protocol）除了支持AMQP之外，还通过插件的方式支持其他消息队列协议，如STOMP、MQTT等

　　RabbitMQ的整体架构图：

　　

交换器：

不同类型的交换器分发消息的策略也不同，Direct、Fanout、Topic等

Driect交换器：

　　Direct交换器基于消息中的路由键将消息投递到对应的消息队列中，Direct交换器是完全匹配，单播的模式。

　　1、每一个消息队列根据routing key 为K绑定到交换器上

　　2、当一个到达Direct 交换器中的消息的routing key 为R时，交换器根据路由key R找到Binding中的路由key K和R相等的队列，将消息投递到这个

　　　　队列中

　　

Fanout交换器：

　　Fanout交换器路由消息到所有的与该交换器绑定的消息队列中，每个队列都会得到这个消息的一个拷贝。忽略掉Routing key。

　　

Topic交换器

　　Topic交换器根据消息中的Routing key和队列与绑定到交换器中的Routing key的模式进行匹配。消息中的Routing key与模式匹配的话就可以将消息分发到队列中，因此可以分发到一个或多个队列中。

Topic交换器通常用于各种发布/订阅模式的变体和消息的多播路由。Topic交换器将路由键和绑定键的字符串切分成单词，这些单词之间用点”.“隔开，且Topic交换器会识别两个通配符，#和*，#匹配0或多个单词，*匹配刚好一个单词。

 <!--RabbitMQ-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>4.1.0</version>
        </dependency>

 生产者：

package com.yang.spbo.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * rabbitmq生产者
 * 〈功能详细描述〉
 *
 * @author 17090889
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        // 设置rabbitmq地址
        connectionFactory.setHost("localhost");
        // 虚拟主机
        connectionFactory.setVirtualHost("/");
        // 建立到代理服务器的连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();
        // 声明direct交换器
        String exchangeName = "hello-exchange";
        channel.exchangeDeclare(exchangeName, "direct", true);
        // 路由键
        String routingKey = "testRoutingKey";
        // 发布消息
        byte[] messageBodyBytes = "quit".getBytes();
        channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);
        // 关闭信道和连接
        channel.close();
        connection.close();
    }
}

消费者：

消息消费者通过不断循环等待服务器推送消息，一旦有消息过来，就在控制台输出消息的相关内容。

package com.yang.spbo.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * rabbitmq消费者
 * 〈功能详细描述〉
 *
 * @author 17090889
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        // 设置rabbitmq地址
        connectionFactory.setHost("localhost");
        // 虚拟主机
        connectionFactory.setVirtualHost("/");
        // 建立到代理服务器的连接
        Connection connection = connectionFactory.newConnection();
        // 创建信道,不能修改
        final Channel channel = connection.createChannel();
        // 声明direct交换器
        String exchangeName = "hello-exchange";
        channel.exchangeDeclare(exchangeName, "direct", true);
        // 声明队列
        String queueName = channel.queueDeclare().getQueue();
        // 路由键
        String routingKey = "testRoutingKey";
        // 将交换器和队列根据路由key绑定起来
        channel.queueBind(queueName, exchangeName, routingKey);
        while (true) {
            // 消费消息
            boolean autoAck = false;
            String consumerTag = "";
            channel.basicConsume(queueName, autoAck, consumerTag, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String routingKey = envelope.getRoutingKey();
                    String contentType = properties.getContentType();
                    System.out.println("消费的路由键：" + routingKey);
                    System.out.println("消费的内容类型：" + contentType);
                    long deliveryTag = envelope.getDeliveryTag();
                    // 确认消息
                    channel.basicAck(deliveryTag, false);
                    String bodyStr = new String(body, "UTF-8");
                    System.out.println("消费的消息体内容：" + bodyStr);
                }
            });
        }
    }
}

使用消息队列可以使之前同步调用的代码改成了异步处理的方式。

Spring整合RabbitMQ

   <!--spring rabbitmq-->
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
            <version>2.0.2.RELEASE</version>
        </dependency>

同时在Spring配置文件中配置连接信息、监听器、队列名称、交换器，rabbitTemplate以及rabbitAdmin等

同时自定义类实现MessageListener

@Service
public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        String messageBody = new String(message.getBody());
    }
}

基于RabbitMQ的消息推送：

　　以浏览器推送接收消息为例，以前，浏览器中的推送功能都是通过轮询来实现的。所谓轮询是指以特定时间间隔（如每隔1s）由浏览器向服务器发送请求，然后服务器返回最新的数据给浏览器。但这种模式的缺点是浏览器需要不断地向服务器发出请求。每次请求中的绝大部分数据都是相同的，里面包含的有效数据只是很小的一部分，这会导致占用很多的带宽，而且不断地连接将大量消耗服务器资源。

　　所以，为了改善这种情况，H5定义了WebSocket，能够实现浏览器与服务器之间全双工通信。其优点有两个：

　　一是服务器与客户端之间交换的标头信息很小；

　　二是服务器可以主动传送数据给客户端；

　　目前的主流浏览器都已支持WebSocket，而服务端消息队列选用RabbitMQ，则是因为RabbitMQ有丰富的第三方插件，用户可以在AMQP协议的基础上自己扩展应用。针对WebSocket通信RabbitMQ提供了Web STOMP插件，它是一个实现了STOMP协议的插件，它是一个实现了STOMP协议的插件，可以将该插件理解为WebSocket与STOMP协议间的桥接，目的是让浏览器能够使用RabbitMQ，当RabbitMQ启用了Web STOMP插件后，浏览器就可以使用WebSocket与之通信了。

　　当有新消息需要发布时，系统后台将消息数据发送到RabbitMQ中，再通过WebSocket将数据推送给浏览器。

　　再js中消费消息。可以在github上下载stomp.js。

消息保存：

　　对队列中的消息的保存方式有disk和RAM两种。

　　disk即写入磁盘，也就是持久化，在发生宕机时，消息数据可以在系统重启之后恢复。

　　采用disk方式，消息数据会被保存在以.rdq后缀命名的文件中，当文件达到16M时会重新生成一个新的文件，当文件中已经被删除的消息比例大于阀值时会触发文件合并操作，以提高磁盘利用率。

　　采用RAM方式，只是在RAM保存内部数据库表数据，而不会保存消息，消息存储索引。队列索引和其他节点等数据，所以必须在启动时从集群中其他节点同步原来的消息数据，这也意味着集群中必须包含至少一个disk方式的节点。

　　消息持久化包括Queue、Message、Exchange持久化三部分。durable：持久的

　　Queue持久化：通过设置queueDeclare方法中的durable参数设置为true；

　　Message持久化：通过设置basePublish方法中的BasicProperties中的deliveryMode为2；

　　Exchange持久化：在声明Exchange时使用支持durable入参的方法，设置为true；

如何保证消息不会丢失？

消息确认模式（生产者确认消息投递到消息队列中）：

　　在默认情况下，生产者把消息发送出去以后，Broker不会返回任何消息给生产者。也就是说，生产者不知道消息有没有到达Broker。如果在消息到达Broker前发生了宕机或者Broker接收到消息在将消息写入磁盘时发生了宕机，那么消息就会丢失。而生产者并不知道消息的情况？

RabbitMQ提供了两种解决方式：1、通过AMQP协议中的事务机制　　2、把信道设置成确认模式

　　AMQP中的事务机制将把信道设置成事务模式后，生产者和Broker之间会有一种发送/响应机制，生产者需要同步等待Broker的执行结果，在性能上会降低消息服务的吞吐量，所以一般采用性能更好的发送方确认方式来保障消息投递，将信道设置为确认模式之后，在该信道上发布的所有消息都会被分配一个唯一ID，一旦消息被投递到所有匹配的队列中，该信道就会向生产者发送确认消息，在确认消息中包含了之前的唯一ID，从而让生产者知道消息已到达目的队列。确认模式最大的优势是异步，生产者可继续发送消息。

消费者回执（消费者成功消费消息）：

　　在实际应用中可能出现消费者接收到消息，但是还没有处理完就发生宕机的情况，这也会导致消息丢失。为避免这种情况，可以要求消费者在消费完消息之后发送一个回执给RabbitMQ服务器，RabbitMQ服务器在收到回执之后再将消息从其队列中删除，如果没有收到回执并且检测到消费者与RabbitMQ服务器的连接断开了，则由RabbitMQ服务器负责把消息发送给其他消费者。如果没有断开，RabbitMQ是不会把消息发送给其他消费者的。

　　1、两种消息回执模式

　　　　1）、自动回执：当Broker成功发送消息给消费者后就会立即把此消息从队列中删除，而不等待消费者回送确认消息

　　　　2）、手动回执：当Broker发送消息给消费者后不会立即把此消息删除，而是等待消费者回执的确认消息后才会删除。消费者收到消息并处理完成后需要向Broker显式发送ACK指令，如果消费者因为意外崩溃而没有发送ACK指令，那么Broker就会把该消息转发给其他消费者，如果此时没有其他消费者，那么Broker会缓存消息。

　　2、拒绝消息

　　　　当消费者处理消息失败或者当前不能处理消息时，可以给Broker发送一个拒绝消息的指令，并且可要求Broker将该消息丢弃或者重新放入队列中。

　　3、消息预取

　　　　为了消费者负载均衡，可以设置预取数量限制每个消费者在收到下一个确认回执前一次可以接收多少条消息。

如何处理消息不被重复消费呢？

　　首先要知道消息为什么会被重复消费，大多是由于网络不通导致，消费者的确认消息没有传送到消息队列，导致消息队列不知道消息已经被消费了，再次

将该消息分发给其他消费者。所以解决的思路有下面几种：

　　1）：如果消息是做数据库的插入操作，给这个消息做一个唯一的主键，那么就算出现重复消费的情况，就会导致主键冲突，避免数据库出现脏数据。

　　2）：判重表，将消费过处理成功的消息存入判重表中，每次消费处理前先去判重表查询是否已消费过

　　2）：如果你拿到这个消息做Redis的set操作，不用解决，因为无论你set几次结果都是一样的，set操作本来就算幂等操作

　　3）：如果上面两种情况都不行，准备一个第三方服务方来做消费记录。以Redis为例，给消息分配一个全局id，只要消费过该消息，将<id,Message>以KV

　　　　形式写入Redis。那消费者开始消费前，先去Redis中查询 有没有消费记录即可。

　　总之，解决思路就是，如果消息重复消费不会带来问题，那大可不用理会，如果有问题，要对消费过的消息做记录（数据库或者缓存），再次消费前查询是否已经被消费。

或者两次操作做互斥操作，使只有一次操作能成功执行。有些情况需要考虑使用分布式锁

如何保证消息顺序消费？ 

　　通过算法，将需要保持先后顺序的消息放在同一个消息队列中，然后只用一个消费者去消费该队列。

　　1、RabbitMQ：如果存在多个消费者，那么就让每个消费者对应一个queue，然后把要发送的数据全部放到一个queue，这样就能保证所有

　　的数据只到达一个消费者从而保证每个数据到达数据库都是顺序的。

　　（拆分多个queue，每个queue一个consumer。或者就是一个queue但是对应一个consumer，然后这个consumer内部用内存队列做排队，然后分发给底层不同的worker来处理）。

　　2、Kafka写入partition时指定一个key，例如订单id，那么消费者从partition中取出数据的时候肯定是有序的，当开启多个线程的时候可能

数据不一致，这时候就需要内存队列，将相同的hash过的数据放在一个内存队列里，这样就能保证一条线程对应一个内存队列的数据写入数据

库的时候顺序性的，从而卡伊开启多条线程对应多个内存队列。

　　（Kafka：一个topic，一个partition，一个consumer，内部单线程消费，写N个内存queue，然后N个线程分别消费一个内存queue即可）。

流控机制：

　　RabbitMQ可以对内存和磁盘的使用量设置阀值，当达到阀值后生产者将被阻塞，直到对应资源的使用恢复正常。除了设置这两个阀值之外，RabbitMQ还用流控（Flow Control）机制来确保稳定性。

　　Kafka

　　特点：

　　1、同时为发布和订阅提供搞吞吐量。Kafka的设计目标是以时间复杂度为O(1)的方式提供消息持久化能力的，即使对TB级别以上数据也能保证常数时间的访问性能，即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输（一般消息处理是百万级，使用Partition实现机器间的并行处理）

　　2、消息持久化。将消息持久化到磁盘，因此可用于批量消费，例如ETL以及实时应用程序。通过将数据持久化到磁盘以及复制可以防止数据丢失

　　3、分布式。支持服务器间的消息分区及分布式消费，同时保证每个Partition内的消息顺序传输。同时保证每个Partition内的消息顺序传输。其内部的Producer、Broker和Consumer都是分布式架构，这更易于向外扩展。

　　4、 消费消息采用Pull模式。消息被处理的状态是在Consumer端维护的，而不是由服务端维护，Broker无状态，Consumer自己保存offset。

　　5、支持Online和Offline场景，同时支持离线数据处理和实时数据处理。

基本概念：

　　Broker：Kafka集群中的一台或多台服务器

　　Topic：主题，发布到Kafka的每条消息都有一个类别，这个类别就被称为Topic（物理上，不同Topic的消息分开存储；逻辑上，虽然一个Topic的消息被保存在一个或多个Broker上，但用户只需要指定消息的Topic即可生产或消费数据，而不必关心数据存于何处）

　　Partition：物理上的Topic分区，一个Topic可以分为多个Partition，每个partition都是一个有序的队列。Partition中的每条消息都会被分配一个有序的ID（offset），它唯一地标识分区中的每个记录。每个Partition只能被一个消费组中的一个消费者消费。

　　Producer：消息和数据的生产者，可以理解为向Kafka发送消息的客户端

　　Consumer：消息和数据的消费者，可以理解为从Kafka取消息的客户端，通过与Kafka集群建立长连接的方式，不断的从集群中拉去消息

　　Consumer Group（消费组）：每个消费者都属于一个特定的消费组（可为每个消费者指定组名，若不指定组名，则属于默认的组）。这是Kafka用来实现一个Topic的广播（发送给所有的消费者）和单播（发送给任意一个消费者）的手段。一个Topic可以由多个消费组。但对每个消费组，只会把消息发送给该组中的一个消费者。如果要实现广播，只要每个消费者都有一个独立的消费组就可以了；如果要实现单播，只要所有的消费者都在同一个消费者组中就行。

　　      

　　一个典型的Kafka集群中包含若干生产者、若干Broker（Kafka支持水平扩展，一般Broker数量越多集群吞吐量越大）、若干消费者组以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置、选举leader，以及当消费者组发生变化时进行Rebalance（再均衡）。生产者使用推模式将消息发布到Broker，消费者使用拉模式从Broker订阅并消费消息。

　　

　　创建一个Topic时，可以指定分区数目，分区数越多，其吞吐量越大，但是需要的资源也越多，也会带来更高的不可用性。

　　生产者在向kafka集群发送消息的分区策略：

　　1、可以指定分区，则消息投递到指定的分区

　　2、如果没有指定分区，但是消息的key不为空，则基于key的哈希值来选择一个分区

　　3、如果既没有指定分区，且消息的key也为空，则用轮询的方式选择一个分区

也就是一条消息只会发送到一个分区中。

　　对于一个group而言，消费者的数量不应该多余分区的数量，因为在一个group中，每个分区最多只能绑定到一个消费者上，只能被一个消费组中的一个消费者消费。而一个消费者可以消费多个分区。因此，若一个消费组中的消费者数量大于分区数量的话，多余的消费者将不会收到消息（没有分区可以消费）。

　　实例：

        <!--kafka-->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.1</version>
        </dependency>

消息生产者：

package com.yang.spbo.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;

/**
 * Kafka消息生产者
 * 〈功能详细描述〉
 *
 * @author 17090889
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class ProducerSample {
    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        // Kafka集群，多台服务器地址之间用逗号隔开
        props.put("bootstrap.servers", "localhost:9092");
        // 消息的序列化类型
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 消息的反序列化类型
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // zookeeper集群地址，提供了基于Zookeeper的集群服务器自动感知功能，可以动态从Zookeeper中读取Kafka集群配置信息
        props.put("zk.connect", "127.0.0.1:2181");
        String topic = "test-topic";
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        // 发送消息
        producer.send(new ProducerRecord<String, String>(topic, "idea-key2", "javaMesage1"));
        producer.close();
    }
}

ProducerRecord构造：

public ProducerRecord(String topic, Integer partition, K key, V value) {
        this(topic, partition, (Long)null, key, value, (Iterable)null);
    }

    public ProducerRecord(String topic, K key, V value) {
        this(topic, (Integer)null, (Long)null, key, value, (Iterable)null);
    }

    public ProducerRecord(String topic, V value) {
        this(topic, (Integer)null, (Long)null, (Object)null, value, (Iterable)null);
    }

topic和value是必填的，如果指定了partition，那么消息会被发送至指定的partition；如果没有指定partition但指定了key，那么消息会按照hash(key)发送至指定的partition；如果既没有指定partition，也没有指定key，那么消息会按照round-robin模式发送至每一个partition。

消息消费者：

package com.yang.spbo.kafka;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**
 * Kafka消息消费者
 * 〈功能详细描述〉
 *
 * @author 17090889
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本] （可选）
 */
public class ConsumerSample {
    public static void main(String[] args) {
        String topic = "test-topic";
        Properties props = new Properties();
        // Kafka集群，多台服务器地址之间用逗号隔开
        props.put("bootstrap.servers", "localhost:9092");
        // 消费组ID
        props.put("group.id", "test_group1");
        // Consumer的offset是否自动提交
        props.put("enable.auto.commit", "true");
        // 自动提交offset到zk的时间间隔，时间单位是毫秒
        props.put("auto.commit.interval.ms", "1000");
        // 消息的反序列化类型
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        // 订阅的话题
        consumer.subscribe(Arrays.asList(topic));
        // Consumer调用poll方法来轮询Kafka集群的消息，一直等到Kafka集群中没有消息或者达到超时时间100ms为止
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                System.out.println(record.partition() + record.offset());
                System.out.println(record.key());
                System.out.println(record.value());
            }
        }
    }
}

Kafka应用：

　　1、用户行为数据采集

　　2、基于Kafka的日志收集：

　　　　集群方式部署的应用，日志文件有多个存放地址，需要快速定位日志问题就比较繁琐，那么就需要一个统一的日志平台来管理项目中产生的日志文件。

　　　　各个应用系统在输出日志时利用拥有高吞吐量的Kafka作为数据缓冲平台，将日志统一输出到Kafka，再通过Kafka以统一接口服务的方式开放给消费者。

　　　　现在很多公司做的统一日志平台就是收集重要的系统日志几种到Kafka中，然后再导入Elasticsearch、HDFS、Storm等具体日志数据的消费者中，用于进行实时搜索分析、离线统计、数据备份、大数据分析等。引入log4j和Kafka的集成包kafka-log4j-appender，在日志配置文件中配置kafka信息即可

　　3、基于Kafka的流量削峰

　　　　比如秒杀带来的流量高峰的场景，为了保证系统高可用，加入消息队列作为信息流的缓冲，从而缓解短时间内产生的高流量带来的压垮整个应用的问题，这就叫流量削峰。

　　　　比如：秒杀场景，将商品的基本信息和库存使用缓存预热保存在Redis中，然后从Redis中读取，系统后台收到秒杀下单请求先从Redis中预减库存，如果库存

　　不足，返回秒杀失败；如果库存充足，则将请求的业务数据放入消息队列Kafka中排队，之后请求立即返回页面。消息队列的消费者在收到消息后取得业务数据，

　　执行后续的生成订单、扣减数据库和写消息操作。

Kafka分区：

　　在使用Kafka作为消息队列时，不管是发布还是订阅都需要指定主题topic，在这里的主题是一个逻辑上的概念，实际上Kafka的基本存储单元是分区Partition，在一个Topic中会有一个或多个Partition，不同的Partition可位于不同的服务器节点上，物理上一个Partition对应一个文件夹。（分区是Topic私有的，所有的Topic之间不共享分区）

　　站在生产者和Broker的角度，对不同Partition的写操作时完全并行的，但对消费者而言，其并发数则取决于Partition的数量。

　　所以在实际的项目中需要配置合适的Partition数量，而这个数值需要根据所设计的系统吞吐量来推算。假设p是生产者写入单个Partition的最大吞吐量，c表示消费者从单个Partitin消费的最大吞吐量，系统需要的目标吞吐量为t，那么Partition的数量应取t/p和t/c之间的大者。而且Partition的值要大于或等于消费组中消费者的数量。

Kafka集群复制（1.0保证消息不丢失的策略）

　　Kafka使用了zookeeper实现了去中心化的集群功能，简单地讲，其运行机制是利用zookeeper维护集群成员的信息，每个Broker实例都会被设置一个唯一的标识符，Broker在启动时会通过创建临时节点的方式把自己的唯一标识符注册到zookeeper中，Kafka中的其他组件会监视Zookeeper里的/broker/ids路径，所以当集群中有Broker加入或退出时其他组件就会收到通知。

　　虽然Kafka有集群功能，但是在0.8版本之前一直存在一个严重的问题，就是一旦某个Broker宕机，该Broker上的所有Partition数据就不能被消费了，生产者也不能把数据存放在这些Partition中了，显然不满足高可用设计。

　　为了让Kafka集群中某些节点不能继续提供服务的情况下，集群对外整体依然可用，即生产者可继续发送消息，消费者可继续消费消息，所以需要提供一种集群间数据的复制机制。在Kafka中是通过使用Zookeeper提供的leader选举方式来实现数据复制方案的，其基本原理是：首先在Kafka集群中的所有节点中选举出一个leader，其他副本作为follower，所有的写操作都先发给leader，然后再由leader把消息发给follower。

　　复制方案使Kafka集群可以在部分节点不可用的情况下还能保证Kafka的整体可用性。Kafka中的复制操作也是针对分区的。一个分区有多个副本，副本被保存在Broker上，每个Broker都可以保存上千个属于不同主题和分区的副本。副本有两种类型：leader副本（每个分区都会有）和follower副本（除了leader副本之外的其他副本）。为了保证一致性，所有的生产者和消费者的请求都会经过leader。而follower不处理客户端的请求，它的职责是从leader处复制消息数据，使自己和leader的状态保持一致，如果leader节点宕机，那么某个follower就会被选为leader继续对外提供服务。

　　Kafka保证消息不丢失的方案

一、消息发送

　　1、消息发送确认：消息数据是存储在分区中的，而分区又可能有多个副本，所以一条消息被发送到Broker之后何时算投递成功呢？Kafka提供了三种模式：

　　　　1）：不等Broker确认，消息被发送出去就认为是成功的。这种方式延迟最小，但是不能保证消息已经被成功投递到Broker

　　　　2）：由leader确认，当leader确认接收到消息就认为投递是成功的，然后由其他副本通过异步方式拉取

　　　　3）：由所有的leader和follower都确认接收到消息才认为是成功的。采用这种方式投递的可靠性最高，但相对会损伤性能

　　　　// 生产者消息发送确认模式，0表示第一种，1表示第二种，all表示第三种
        props.put("acks", "1");

　　2、消息重发：Kafka为了高可用性，生产者提供了自动重试机制。当从Broker接收到的是临时可恢复的异常时，生产者会向Broker重发消息，但不能无限

　　次重发，如果重发次数达到阀值，生产者将不再重试并返回错误。

　　　　 // 消息发送重试次数
        props.put("retries", "10");
        // 重试间隔时间，默认100ms，设置时需要知道节点恢复所用的时间，要设置的比节点恢复所用时间长
        props.put("retry.backoff.ms", "1000");

二、消息消费

　　从设计上来说，由于Kafka服务端并不保存消息的状态，所以在消费消息时就需要消费者自己去做很多事情，消费者每次调用poll方法时，该方法总是返回

由生产者写入Kafka中但还没有被消费者消费的消息。Kafka在设计上有一个不同于其他JMS队列的地方是生产者的消息并不需要消费者确认，而消息在分区中

又是顺序排列的，那么必然就可以通过一个偏移量offset来确定每一条消息的位置，偏移量在消息消费的过程中起着很重要的作用。

　　更新分区当前位置的操作叫做提交偏移量，Kafka中有个叫做_consumer_offset的特殊主题用来保存消息在每个分区的偏移量，消费者每次消费时都会往

这个主题中发送消息，消息包含每个分区的偏移量。如果消费者崩溃或者有新的消费者加入消费组从而触发再均衡操作，再均衡之后该分区的消费者若不是之前

那个，那么新的消费者如何得知该分区的消息已经被之前的消费者消费到哪个位置了呢？这种情况下，就提现了偏移量的用处。为了能继续之前的工作，新的消

费者需要读取每个分区最后一次提交的偏移量，然后再从偏移量开始继续往下消费消息。

偏移量提交方式：

　　1、自动提交

　　Kafka默认会定期自动提交偏移量，提交的默认时间间隔是5000ms，但可能存在提交不及时导致再均衡之后重复消费的情况

        // Consumer的offset是否自动提交
        props.put("enable.auto.commit", "true");
        // 自动提交offset到zk的时间间隔，时间单位是毫秒
        props.put("auto.commit.interval.ms", "1000");

　　2、手动提交

　　先关闭消费者的自动提交配置，然后使用commitSync方法提交偏移量。

　　　　// 关闭自动提交
        props.put("enable.auto.commit", "false");
　 　　// Consumer调用poll方法来轮询Kafka集群的消息，一直等到Kafka集群中没有消息或者达到超时时间100ms为止
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                System.out.println(record.partition() + record.offset());
                System.out.println(record.key());
                System.out.println(record.value());
            }
            // 手动提交最新的偏移量
            consumer.commitSync();
        }

commitSync方法会提交由poll返回的最新偏移量，所以在处理完记录后要确保调用了commitSync方法，否则还是会发生重复处理的问题。

　　3、异步提交

　　使用commitSync方法提交偏移量有一个不足之处，就是该方法在Broker对提交请求做出回应前是阻塞的，要等待回应。因此，采用这种方式每提交一次偏移量就

等待一次限制了消费端的吞吐量，因此Kafka提供了异步提交的方式【consumer.commitAsync();】，消费者只管发送提交请求，而不需要等待Broker的立即回应。

但commitSync方法在成功提交之前如碰到无法恢复的错误之前会一直重试，而commitAsync并不会，因为为了避免异步提交的偏移量被覆盖。

　　Kafka高吞吐量的原因？

　　1）、顺序读写

　　　　Kafka的消息是不断追加到文件中的，这个特性使Kafka可以充分利用磁盘的顺序读写性能。顺序读写不需要磁盘磁头的寻道时间，只需很少的扇区

　　　　旋转时间，所以速度远快于随机读写

　　2）、零拷贝

　　　　在Linux Kernel2.2之后出现了一种叫做“零拷贝（zero-copy）”系统调用机制，就是跳过“用户缓冲区”的拷贝，建立一个磁盘空间和内存的直接映射，

　　　　数据不再复制到“用户缓冲区”

　　3）、分区

　　　　kafka中的topic中的内容可以分在多个分区（partition）存储，每个partition又分为多个段segment，所以每次操作都是针对一小部分做操作，很轻便，

　　　　并且增加并行操作的能力

　　4）、批量发送

　　　　Kafka允许进行批量发送消息，Productor发送消息的时候，可以将消息缓存在本地，等到了固定条件发送到kafka

　　　　（1）：等消息条数到固定条数

　　　　（2）：一段时间发送一次

　　5）、数据压缩

　　　　Kafka还支持对消息集合进行压缩，Producer可以通过GZIP或Snappy格式对消息集合进行压缩，压缩的好处就是减少传输的数据量，减轻

　　　　对网络传输的压力。

　　　　批量发送和数据压缩一起使用，单条做数据压缩的话，效果不太明显。