概念和基本架构Kafka介绍Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多生产者、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。主要应用场景是：日志收集系统和消息系统。Kafka主要设计目标如下：以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。同时支持离线数据处理和实时数据处理。支持在线水平扩展有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。对于消息中间件，消息分推拉两种模式。Kafka只有消息的拉取，没有推送，可以通过轮询实现消息的推送。Kafka在一个或多个可以跨越多个数据中心的服务器上作为集群运行。Kafka集群中按照主题分类管理，一个主题可以有多个分区，一个分区可以有多个副本分区。每个记录由一个键，一个值和一个时间戳组成。Kafka具有四个核心API：Producer API：允许应用程序将记录流发布到一个或多个Kafka主题。Consumer API：允许应用程序订阅一个或多个主题并处理为其生成的记录流。Streams API：允许应用程序充当流处理器，使用一个或多个主题的输入流，并生成一个或多个输出主题的输出流，从而有效地将输入流转换为输出流。Connector API：允许构建和运行将Kafka主题连接到现有应用程序或数据系统的可重用生产者或使用者。例如，关系数据库的连接器可能会捕获对表的所有更改。Kafka优势高吞吐量：单机每秒处理几十上百万的消息量。即使存储了许多TB的消息，它也保持稳定的性能。高性能：单节点支持上千个客户端，并保证零停机和零数据丢失。持久化数据存储：将消息持久化到磁盘。通过将数据持久化到硬盘以及replication防止数据丢失。零拷贝顺序读，顺序写利用Linux的页缓存分布式系统，易于向外扩展。所有的Producer、Broker和Consumer都会有多个，均为分布式的。无需停机即可扩展机器。多个Producer、Consumer可能是不同的应用。可靠性 - Kafka是分布式，分区，复制和容错的。客户端状态维护：消息被处理的状态是在Consumer端维护，而不是由server端维护。当失败时能自动平衡。支持online和offline的场景。支持多种客户端语言。Kafka支持Java、.NET、PHP、Python等多种语言。Kafka应用场景日志收集：一个公司可以用Kafka可以收集各种服务的Log，通过Kafka以统一接口服务的方式开放给各种Consumer；消息系统：解耦生产者和消费者、缓存消息等；用户活动跟踪：Kafka经常被用来记录Web用户或者App用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到Kafka的Topic中，然后消费者通过订阅这些Topic来做实时的监控分析，亦可保存到数据库；运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；流式处理：比如Spark Streaming和Storm。基本架构消息和批次Kafka的数据单元称为消息。可以把消息看成是数据库里的一个“数据行”或一条“记录”。消息由字节数组组成。消息有键，键也是一个字节数组。当消息以一种可控的方式写入不同的分区时，会用到键。为了提高效率，消息被分批写入Kafka。批次就是一组消息，这些消息属于同一个主题和分区。把消息分成批次可以减少网络开销。批次越大，单位时间内处理的消息就越多，单个消息的传输时间就越长。批次数据会被压缩，这样可以提升数据的传输和存储能力，但是需要更多的计算处理。模式消息模式（schema）有许多可用的选项，以便于理解。如JSON和XML，但是它们缺乏强类型处理能力。Kafka的许多开发者喜欢使用Apache Avro。Avro提供了一种紧凑的序列化格式，模式和消息体分开。当模式发生变化时，不需要重新生成代码，它还支持强类型和模式进化，其版本既向前兼容，也向后兼容。数据格式的一致性对Kafka很重要，因为它消除了消息读写操作之间的耦合性。主题和分区Kafka的消息通过主题进行分类。主题可比是数据库的表或者文件系统里的文件夹。主题可以被分为若干分区，一个主题通过分区分布于Kafka集群中，提供了横向扩展的能力。生产者和消费者生产者创建消息。消费者消费消息。一个消息被发布到一个特定的主题上。生产者在默认情况下把消息均衡地分布到主题的所有分区上：直接指定消息的分区根据消息的key散列取模得出分区轮询指定分区。消费者通过偏移量来区分已经读过的消息，从而消费消息。消费者是消费组的一部分。消费组保证每个分区只能被一个消费者使用，避免重复消费。broker和集群一个独立的Kafka服务器称为broker。broker接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。broker为消费者提供服务，对读取分区的请求做出响应，返回已经提交到磁盘上的消息。单个broker可以轻松处理数千个分区以及每秒百万级的消息量。每个集群都有一个broker是集群控制器（自动从集群的活跃成员中选举出来）。控制器负责管理工作：将分区分配给broker监控broker集群中一个分区属于一个broker，该broker称为分区首领。一个分区可以分配给多个broker，此时会发生分区复制。分区的复制提供了消息冗余，高可用。副本分区不负责处理消息的读写。核心概念Producer生产者创建消息。该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。一般情况下，一个消息会被发布到一个特定的主题上。默认情况下通过轮询把消息均衡地分布到主题的所有分区上。在某些情况下，生产者会把消息直接写到指定的分区。这通常是通过消息键和分区器来实现的，分区器为键生成一个散列值，并将其映射到指定的分区上。这样可以保证包含同一个键的消息会被写到同一个分区上。生产者也可以使用自定义的分区器，根据不同的业务规则将消息映射到分区。Consumer消费者读取消息。消费者订阅一个或多个主题，并按照消息生成的顺序读取它们。消费者通过检查消息的偏移量来区分已经读取过的消息。偏移量是另一种元数据，它是一个不断递增的整数值，在创建消息时，Kafka会把它添加到消息里。在给定的分区里，每个消息的偏移量都是唯一的。消费者把每个分区最后读取的消息偏移量保存在Zookeeper或Kafka上，如果消费者关闭或重启，它的读取状态不会丢失。消费者是消费组的一部分。群组保证每个分区只能被一个消费者使用。如果一个消费者失效，消费组里的其他消费者可以接管失效消费者的工作，再平衡，分区重新分配。Broker一个独立的Kafka 服务器被称为broker。broker 为消费者提供服务，对读取分区的请求作出响应，返回已经提交到磁盘上的消息。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。broker 是集群的组成部分。每个集群都有一个broker 同时充当了集群控制器的角色（自动从集群的活跃成员中选举出来）。控制器负责管理工作，包括将分区分配给broker 和监控broker。在集群中，一个分区从属于一个broker，该broker 被称为分区的首领。Topic每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。物理上不同Topic的消息分开存储。主题就好比数据库的表，尤其是分库分表之后的逻辑表。Partition主题可以被分为若干个分区，一个分区就是一个提交日志。消息以追加的方式写入分区，然后以先入先出的顺序读取。无法在整个主题范围内保证消息的顺序，但可以保证消息在单个分区内的顺序。Kafka 通过分区来实现数据冗余和伸缩性。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。ReplicasKafka 使用主题来组织数据，每个主题被分为若干个分区，每个分区有多个副本。那些副本被保存在broker上，每个broker可以保存成百上千个属于不同主题和分区的副本。副本有以下两种类型：首领副本每个分区都有一个首领副本。为了保证一致性，所有生产者请求和消费者请求都会经过这个副本。跟随者副本首领以外的副本都是跟随者副本。跟随者副本不处理来自客户端的请求，它们唯一的任务就是从首领那里复制消息，保持与首领一致的状态。如果首领发生崩溃，其中的一个跟随者会被提升为新首领。Offset生产者Offset消息写入的时候，每一个分区都有一个offset，这个offset就是生产者的offset，同时也是这个分区的最新最大的offset。有些时候没有指定某一个分区的offset，这个工作kafka帮我们完成。消费者Offset这是某一个分区的offset情况，生产者写入的offset是最新最大的值是12，而当Consumer A进行消费时，从0开始消费，一直消费到了9，消费者的offset就记录在9，Consumer B就纪录在了11。等下一次他们再来消费时，他们可以选择接着上一次的位置消费，当然也可以选择从头消费，或者跳到最近的记录并从“现在”开始消费。副本Kafka通过副本保证高可用。副本分为首领副本(Leader)和跟随者副本(Follower)。跟随者副本包括同步副本和不同步副本，在发生首领副本切换的时候，只有同步副本可以切换为首领副本。AR分区中的所有副本统称为AR（Assigned Repllicas）。AR=ISR+OSRISR所有与leader副本保持一定程度同步的副本（包括Leader）组成ISR（In-Sync Replicas），ISR集合是AR集合中的一个子集。消息会先发送到leader副本，然后follower副本才能从leader副本中拉取消息进行同步，同步期间内follower副本相对于leader副本而言会有一定程度的滞后。前面所说的“一定程度”是指可以忍受的滞后范围，这个范围可以通过参数进行配置。OSR与leader副本同步滞后过多的副本（不包括leader）副本，组成OSR（Out-Sync Relipcas）。在正常情况下，所有的follower副本都应该与leader副本保持一定程度的同步，即AR=ISR，OSR集合为空。HWHW是High Watermak的缩写， 俗称高水位，它表示了一个特定消息的偏移量（offset），消费之只能拉取到这个offset之前的消息。LEOLEO是Log End Offset的缩写，它表示了当前日志文件中下一条待写入消息的offset。Kafka安装与配置Java环境为前提上传jdk-8u261-linux-x64.rpm到服务器并安装：rpm -ivh jdk-8u261-linux-x64.rpm配置环境变量vim /etc/profileexport JAVA_HOME=/usr/java/jdk1.8.0_261-amd64 export PATH:$PATH:$JAVA_HOME/bin# 生效 source /etc/profile # 验证 java -versionZookeeper的安装配置上传zookeeper-3.4.14.tar.gz到服务器解压到/opttar -zxf zookeeper-3.4.14.tar.gz -C /opt cd /opt/zookeeper-3.4.14/conf # 复制zoo_sample.cfg命名为zoo.cfg cp zoo_sample.cfg zoo.cfg # 编辑zoo.cfg文件 vim zoo.cfg修改Zookeeper保存数据的目录，dataDirdataDir=/var/test/zookeeper/data编辑/etc/profile设置环境变量ZOO_LOG_DIR，指定Zookeeper保存日志的位置；ZOOKEEPER_PREFIX指向Zookeeper的解压目录；将Zookeeper的bin目录添加到PATH中：export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14 export PATH=$PATH:$ZOOKEEPER_PREFIX/bin export ZOO_LOG_DIR=/var/test/zookeeper/log使配置生效source /etc/profile6、验证Kafka的安装与配置上传kafka_2.12-1.0.2.tgz到服务器并解压tar -zxf kafka_2.12-1.0.2.tgz -C /opt配置环境变量并生效vim /etc/profileexport KAFKA_HOME=/opt/kafka_2.12-1.0.2 export PATH=$PATH:$KAFKA_HOME/bin配置/opt/kafka_2.12-1.0.2/config中的server.properties文件Kafka连接Zookeeper的地址，此处使用本地启动的Zookeeper实例，连接地址是localhost:2181，后面的myKafka是Kafka在Zookeeper中的根节点路径：启动ZookeeperzkServer.sh start确认Zookeeper的状态启动Kafka进入Kafka安装的根目录，执行如下命令：kafka-server-start.sh config/server.properties启动成功，可以看到控制台输出的最后一行的started状态：查看Zookeeper的节点此时Kafka是前台模式启动，要停止，使用Ctrl+C。如果要后台启动，使用命令：kafka-server-start.sh -daemon config/server.properties查看Kafka的后台进程：ps aux | grep kafka停止后台运行的Kafka：生产与消费kafka-topics.sh用于管理主题# 列出现有的主题 kafka-topics.sh --list --zookeeper localhost:2181/myKafka # 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。 kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_1 --partitions 1 --replication-factor 1 # 查看分区信息 kafka-topics.sh --zookeeper localhost:2181/myKafka --list # 查看指定主题的详细信息 kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_1 # 删除指定主题 kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_1kafka-console-producer.sh用于生产消息# 开启生产者 kafka-console-producer.sh --topic topic_1 --broker-list localhost:9020kafka-console-consumer.sh用于消费消息# 开启消费者 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1 # 开启消费者方式二，从头消费，不按照偏移量消费 kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_1 --from-beginningKafka开发实战消息的发送与接收生产者主要的对象有：KafkaProducer，ProducerRecord。其中KafkaProducer是用于发送消息的类，ProducerRecord类用于封装Kafka的消息。KafkaProducer的创建需要指定的参数和含义：参数说明bootstrap.servers配置生产者如何与broker建立连接。该参数设置的是初始化参数。如果生产者需要连接的是Kafka集群，则这里配置集群中几个broker的地址，而不是全部，当生产者连接上此处指定的broker之后，在通过该连接发现集群中的其他节点。key.serializer要发送信息的key数据的序列化类。设置的时候可以写类名，也可以使用该类的Class对象。value.serializer要发送消息的alue数据的序列化类。设置的时候可以写类名，也可以使用该类的Class对象。acks默认值：all。acks=0：生产者不等待broker对消息的确认，只要将消息放到缓冲区，就认为消息已经发送完成。该情形不能保证broker是否真的收到了消息，retries配置也不会生效。发送的消息的返回的消息偏移量永远是-1。acks=1：表示消息只需要写到主分区即可，然后就响应客户端，而不等待副本分区的确认。在该情形下，如果主分区收到消息确认之后就宕机了，而副本分区还没来得及同步该消息，则该消息丢失。acks=all：首领分区会等待所有的ISR副本分区确认记录。该处理保证了只要有一个ISR副本分区存活，消息就不会丢失。这是Kafka最强的可靠性保证，等效于acks=-1retriesretries重试次数。当消息发送出现错误的时候，系统会重发消息。跟客户端收到错误时重发一样。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1，否则在重试此失败消息的时候，其他的消息可能发送成功了其他参数可以从org.apache.kafka.clients.producer.ProducerConfig中找到。后面的内容会介绍到。消费者生产消息后，需要broker端的确认，可以同步确认，也可以异步确认。同步确认效率低，异步确认效率高，但是需要设置回调对象。生产者：package com.test.kafka.demo.producer; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.HashMap; import java.util.Map; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class MyProducer1 { public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException { Map < String, Object > configs = new HashMap < > (); // 设置连接Kafka的初始连接用到的服务器地址 // 如果是集群，则可以通过此初始连接发现集群中的其他broker configs.put("bootstrap.servers", "node1:9092"); // 设置key的序列化器 configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"); // 设置value的序列化器 configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); configs.put("acks", "1"); KafkaProducer < Integer, String > producer = new KafkaProducer < Integer, String > (configs); // 用于封装Producer的消息 ProducerRecord < Integer, String > record = new ProducerRecord < Integer, String > ( "topic_1", // 主题名称 0, // 分区编号，现在只有一个分区，所以是0 0, // 数字作为key "message 0" // 字符串作为value ); // 发送消息，同步等待消息的确认 producer.send(record).get(3 _000, TimeUnit.MILLISECONDS); // 关闭生产者 producer.close(); } }生产者2：package com.test.kafka.demo.producer; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.HashMap; import java.util.Map; public class MyProducer2 { public static void main(String[] args) { Map < String, Object > configs = new HashMap < > (); configs.put("bootstrap.servers", "node1:9092"); configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"); configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer < Integer, String > producer = new KafkaProducer < Integer, String > (configs); ProducerRecord < Integer, String > record = new ProducerRecord < Integer, String > ( "topic_1", 0, 1, "test message 2" ); // 使用回调异步等待消息的确认 producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { System.out.println( "主题：" + metadata.topic() + "\n" + "分区：" + metadata.partition() + "\n" + "偏移量：" + metadata.offset() + "\n" + "序列化的key字节：" + metadata.serializedKeySize() + "\n" + "序列化的value字节：" + metadata.serializedValueSize() + "\n" + "时间戳：" + metadata.timestamp() ); } else { System.out.println("有异常：" + exception.getMessage()); } } }); // 关闭连接 producer.close(); } }生产者3：package com.test.kafka.demo.producer; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import java.util.HashMap; import java.util.Map; public class MyProducer3 { public static void main(String[] args) { Map < String, Object > configs = new HashMap < > (); configs.put("bootstrap.servers", "node1:9092"); configs.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"); configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer < Integer, String > producer = new KafkaProducer < Integer, String > (configs); for (int i = 100; i < 200; i++) { ProducerRecord < Integer, String > record = new ProducerRecord < Integer, String > ( "topic_1", 0, i, "test message " + i ); // 使用回调异步等待消息的确认 producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { System.out.println( "主题：" + metadata.topic() + "\n" + "分区：" + metadata.partition() + "\n" + "偏移量：" + metadata.offset() + "\n" + "序列化的key字节：" + metadata.serializedKeySize() + "\n" + "序列化的value字节：" + metadata.serializedValueSize() + "\n" + "时间戳：" + metadata.timestamp() ); } else { System.out.println("有异常：" + exception.getMessage()); } } }); } // 关闭连接 producer.close(); } }消息消费流程：消费者：package com.test.kafka.demo.consumer; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; import java.lang.reflect.Array; import java.util.*; import java.util.regex.Pattern; public class MyConsumer1 { public static void main(String[] args) { Map < String, Object > configs = new HashMap < > (); // 指定bootstrap.servers属性作为初始化连接Kafka的服务器。 // 如果是集群，则会基于此初始化连接发现集群中的其他服务器。 configs.put("bootstrap.servers", "node1:9092"); // key的反序列化器 configs.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"); // value的反序列化器 configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); configs.put("group.id", "consumer.demo"); // 创建消费者对象 KafkaConsumer < Integer, String > consumer = new KafkaConsumer < Integer, String > (configs); // final Pattern pattern = Pattern.compile("topic_\\d"); final Pattern pattern = Pattern.compile("topic_[0-9]"); // 消费者订阅主题或分区 // consumer.subscribe(pattern); // consumer.subscribe(pattern, new ConsumerRebalanceListener() { final List < String > topics = Arrays.asList("topic_1"); consumer.subscribe(topics, new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection < TopicPartition > partitions) { partitions.forEach(tp - > { System.out.println("剥夺的分区：" + tp.partition()); }); } @Override public void onPartitionsAssigned(Collection < TopicPartition > partitions) { partitions.forEach(tp - > { System.out.println(tp.partition()); }); } }); // 拉取订阅主题的消息 final ConsumerRecords < Integer, String > records = consumer.poll(3 _000); // 获取topic_1主题的消息 final Iterable < ConsumerRecord < Integer, String >> topic1Iterable = records.records("topic_1"); // 遍历topic_1主题的消息 topic1Iterable.forEach(record - > { System.out.println("========================================"); System.out.println("消息头字段：" + Arrays.toString(record.headers().toArray())); System.out.println("消息的key：" + record.key()); System.out.println("消息的偏移量：" + record.offset()); System.out.println("消息的分区号：" + record.partition()); System.out.println("消息的序列化key字节数：" + record.serializedKeySize()); System.out.println("消息的序列化value字节数：" + record.serializedValueSize()); System.out.println("消息的时间戳：" + record.timestamp()); System.out.println("消息的时间戳类型：" + record.timestampType()); System.out.println("消息的主题：" + record.topic()); System.out.println("消息的值：" + record.value()); }); // 关闭消费者 consumer.close(); } }SpringBoot Kafkapom.xml文件<properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka-test</artifactId> <scope>test</scope> </dependency> </dependencies>application.propertiesspring.application.name=springboot-kafka-02 server.port=8080 # 用于建立初始连接的broker地址 spring.kafka.bootstrap-servers=node1:9092 # producer用到的key和value的序列化类 spring.kafka.producer.keyserializer=org.apache.kafka.common.serialization.IntegerSerializer spring.kafka.producer.valueserializer=org.apache.kafka.common.serialization.StringSerializer # 默认的批处理记录数 spring.kafka.producer.batch-size=16384 # 32MB的总发送缓存 spring.kafka.producer.buffer-memory=33554432 # consumer用到的key和value的反序列化类 spring.kafka.consumer.keydeserializer=org.apache.kafka.common.serialization.IntegerDeserializer spring.kafka.consumer.valuedeserializer=org.apache.kafka.common.serialization.StringDeserializer # consumer的消费组id spring.kafka.consumer.group-id=spring-kafka-02-consumer # 是否自动提交消费者偏移量 spring.kafka.consumer.enable-auto-commit=true # 每隔100ms向broker提交一次偏移量 spring.kafka.consumer.auto-commit-interval=100 # 如果该消费者的偏移量不存在，则自动设置为最早的偏移量 spring.kafka.consumer.auto-offset-reset=earliestDemo02SpringbootApplication.javapackage com.test.kafka.demo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class Demo02SpringbootApplication { public static void main(String[] args) { SpringApplication.run(Demo02SpringbootApplication.class, args); } }KafkaConfig.javapackage com.test.kafka.demo.config; import org.apache.kafka.clients.admin.NewTopic; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class KafkaConfig { @Bean public NewTopic topic1() { return new NewTopic("ntp-01", 5, (short) 1); } @Bean public NewTopic topic2() { return new NewTopic("ntp-02", 3, (short) 1); } }KafkaSyncProducerController.javapackage com.test.kafka.demo.controller; import org.apache.kafka.clients.producer.ProducerRecord; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.concurrent.ExecutionException; @RestController public class KafkaSyncProducerController { @Autowired private KafkaTemplate template; @RequestMapping("send/sync/{message}") public String sendSync(@PathVariable String message) { ListenableFuture future = template.send(new ProducerRecord < Integer, String > ("topic-spring-02", 0, 1, message)); try { // 同步等待broker的响应 Object o = future.get(); SendResult < Integer, String > result = (SendResult < Integer, String > ) o; System.out.println( result.getRecordMetadata().topic() + result.getRecordMetadata().partition() + result.getRecordMetadata().offset() ); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } return "success"; } }KafkaAsyncProducerControllerpackage com.test.kafka.demo.controller; import org.apache.kafka.clients.producer.ProducerRecord; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @RestController public class KafkaAsyncProducerController { @Autowired private KafkaTemplate < Integer, String > template; @RequestMapping("send/async/{message}") public String asyncSend(@PathVariable String message) { ProducerRecord < Integer, String > record = new ProducerRecord < Integer, String > ("topic-spring-02", 0, 3, message); ListenableFuture < SendResult < Integer, String >> future = template.send(record); // 添加回调，异步等待响应 future.addCallback(new ListenableFutureCallback < SendResult < Integer, String >> () { @Override public void onFailure(Throwable throwable) { System.out.println("发送失败: " + throwable.getMessage()); } @Override public void onSuccess(SendResult < Integer, String > result) { System.out.println("发送成功：" + result.getRecordMetadata().topic() + "\t" + result.getRecordMetadata().partition() + "\t" + result.getRecordMetadata().offset()); } }); return "success"; } }MyConsumer.javapackage com.test.kafka.demo.consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; import java.util.Optional; @Component public class MyConsumer { @KafkaListener(topics = "topic-spring-02") public void onMessage(ConsumerRecord < Integer, String > record) { Optional < ConsumerRecord < Integer, String >> optional = Optional.ofNullable(record); if (optional.isPresent()) { System.out.println( record.topic() + "\t" + record.partition() + "\t" + record.offset() + "\t" + record.key() + "\t" + record.value()); } } }服务端参数配置$KAFKA_HOME/config/server.properties文件中的配置。zookeeper.connect该参数用于配置Kafka要连接的Zookeeper集群的地址。它的值是一个字符串，使用逗号分隔Zookeeper的多个地址。Zookeeper的单个地址是host:port形式的，可以在最后添加Kafka在Zookeeper中的根节点路径。如：zookeeper.connect=node2:2181,node3:2181,node4:2181/myKafkalisteners用于指定当前Broker向外发布服务的地址和端口。与advertised.listeners配合，用于做内外网隔离。内外网隔离配置：listener.security.protocol.map监听器名称和安全协议的映射配置。比如，可以将内外网隔离，即使它们都使用SSL。listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SSL每个监听器的名称只能在map中出现一次。inter.broker.listener.name用于配置broker之间通信使用的监听器名称，该名称必须在advertised.listeners列表中。inter.broker.listener.name=EXTERNALlisteners用于配置broker监听的URI以及监听器名称列表，使用逗号隔开多个URI及监听器名称。如果监听器名称代表的不是安全协议，必须配置listener.security.protocol.map。每个监听器必须使用不同的网络端口。advertised.listeners需要将该地址发布到zookeeper供客户端使用，如果客户端使用的地址与listeners配置不同。可以在zookeeper的get /myKafka/brokers/ids/<broker.id>中找到。在IaaS环境，该条目的网络接口得与broker绑定的网络接口不同。如果不设置此条目，就使用listeners的配置。跟listeners不同，该条目不能使用0.0.0.0网络端口。advertised.listeners的地址必须是listeners中配置的或配置的一部分。典型配置：broker.id该属性用于唯一标记一个Kafka的Broker，它的值是一个任意integer值。当Kafka以分布式集群运行的时候，尤为重要。最好该值跟该Broker所在的物理主机有关的，如主机名为host1.test.com，则broker.id=1，如果主机名为192.168.100.101，则broker.id=101等等。log.dir通过该属性的值，指定Kafka在磁盘上保存消息的日志片段的目录。它是一组用逗号分隔的本地文件系统路径。如果指定了多个路径，那么broker 会根据“最少使用”原则，把同一个分区的日志片段保存到同一个路径下。broker会往拥有最少数目分区的路径新增分区，而不是往拥有最小磁盘空间的路径新增分区。
RabbitMQ介绍、概念、基本架构RabbitMQ介绍RabbitMQ，俗称“兔子MQ”（可见其轻巧，敏捷），是目前非常热门的一款开源消息中间件，不管是互联网行业还是传统行业都广泛使用（最早是为了解决电信行业系统之间的可靠通信而设计）。高可靠性、易扩展、高可用、功能丰富等支持大多数（甚至冷门）的编程语言客户端。RabbitMQ遵循AMQP协议，自身采用Erlang（一种由爱立信开发的通用面向并发编程的语言）编写。RabbitMQ也支持MQTT等其他协议。RabbitMQ具有很强大的插件扩展能力，官方和社区提供了非常丰富的插件可供选择：https://www.rabbitmq.com/community-plugins.htmlRabbitMQ整体逻辑架构RabbitMQ Exchange类型RabbitMQ常用的交换器类型有： fanout 、direct 、topic 、headers 四种。Fanout会把所有发送到该交换器的消息路由到所有与该交换器绑定的队列中，如图：Directdirect类型的交换器路由规则很简单，它会把消息路由到那些BindingKey和RoutingKey完全匹配的队列中，如下图：、Topictopic类型的交换器在direct匹配规则上进行了扩展，也是将消息路由到BindingKey和RoutingKey相匹配的队列中，这里的匹配规则稍微不同，它约定：BindingKey和RoutingKey一样都是由"."分隔的字符串；BindingKey中可以存在两种特殊字符“”和“#”，用于模糊匹配，其中""用于匹配一个单词，"#"用于匹配多个单词（可以是0个）。Headersheaders类型的交换器不依赖于路由键的匹配规则来路由信息，而是根据发送的消息内容中的headers属性进行匹配。在绑定队列和交换器时指定一组键值对，当发送的消息到交换器时，RabbitMQ会获取到该消息的headers，对比其中的键值对是否完全匹配队列和交换器绑定时指定的键值对，如果匹配，消息就会路由到该队列。headers类型的交换器性能很差，不实用。RabbitMQ数据存储存储机制RabbitMQ消息有两种类型：持久化消息和非持久化消息。这两种消息都会被写入磁盘。持久化消息在到达队列时写入磁盘，同时会内存中保存一份备份，当内存吃紧时，消息从内存中清除。这会提高一定的性能。非持久化消息一般只存于内存中，当内存压力大时数据刷盘处理，以节省内存空间。RabbitMQ存储层包含两个部分：队列索引和消息存储。队列索引：rabbit_queue_index索引维护队列的落盘消息的信息，如存储地点、是否已被给消费者接收、是否已被消费者ack等。每个队列都有相对应的索引。索引使用顺序的段文件来存储，后缀为.idx，文件名从0开始累加，每个段文件中包含固定的segment_entry_count 条记录，默认值是16384。每个index从磁盘中读取消息的时候，至少要在内存中维护一个段文件，所以设置queue_index_embed_msgs_below 值得时候要格外谨慎，一点点增大也可能会引起内存爆炸式增长。消息存储：rabbit_msg_store消息以键值对的形式存储到文件中，一个虚拟主机上的所有队列使用同一块存储，每个节点只有一个。存储分为持久化存储（msg_store_persistent）和短暂存储（msg_store_transient）。持久化存储的内容在broker重启后不会丢失，短暂存储的内容在broker重启后丢失。store使用文件来存储，后缀为.rdq，经过store处理的所有消息都会以追加的方式写入到该文件中，当该文件的大小超过指定的限制（file_size_limit）后，将会关闭该文件并创建一个新的文件以供新的消息写入。文件名从0开始进行累加。在进行消息的存储时，RabbitMQ会在ETS（Erlang Term Storage）表中记录消息在文件中的位置映射和文件的相关信息。channel.queueDeclare("q.demo1", true, false, false, null); channel.exchangeDeclare("ex.demo1", BuiltinExchangeType.DIRECT, true, false, null); channel.queueBind("q.demo1", "ex.demo1", "key.demo1", null); Map<String, Object> headers = new HashMap<>(); headers.put("mykey", "myvalue"); AMQP.BasicProperties props = new AMQP.BasicProperties.Builder() .contentType("text/plain") .contentEncoding("utf-8") .deliveryMode(2) .headers(headers) .build(); // 4096个数字 String message = "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "1234567890123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890" + "123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456"; // message = message.substring(0, message.length() - 25); // message = message.substring(0, message.length() - 38); message = message.substring(0, message.length() - 45); // message = message.substring(0, message.length() - 50); // message = message.substring(0, message.length() - 100); channel.basicPublish("ex.demo1", "key.demo1", props, message.getBytes("UTF-8"));消息（包括消息头、消息体、属性）可以直接存储在index中，也可以存储在store中。最佳的方式是较小的消息存在index中，而较大的消息存在store中。这个消息大小的界定可以通过queue_index_embed_msgs_below 来配置，默认值为4096B。当一个消息小于设定的大小阈值时，就可以存储在index中，这样性能上可以得到优化。一个完整的消息大小小于这个值，就放到索引中，否则放到持久化消息文件中。rabbitmq.conf中的配置信息：## Size in bytes below which to embed messages in the queue index. ## Related doc guide: https://rabbitmq.com/persistence-conf.html ## # queue_index_embed_msgs_below = 4096 ## You can also set this size in memory units ## # queue_index_embed_msgs_below = 4kb如果消息小于这个值，就在索引中存储，如果消息大于这个值就在store中存储：大于这个值的消息存储于msg_store_persistent目录中的.rdq文件中：小于这个值的消息存储于.idx索引文件中：读取消息时，先根据消息的ID（msg_id）找到对应存储的文件，如果文件存在并且未被锁住，则直接打开文件，从指定位置读取消息内容。如果文件不存在或者被锁住了，则发送请求由store进行处理。删除消息时，只是从ETS表删除指定消息的相关信息，同时更新消息对应的存储文件和相关信息。在执行消息删除操作时，并不立即对文件中的消息进行删除，也就是说消息依然在文件中，仅仅是标记为垃圾数据而已。当一个文件中都是垃圾数据时可以将这个文件删除。当检测到前后两个文件中的有效数据可以合并成一个文件，并且所有的垃圾数据的大小和所有文件（至少有3个文件存在的情况下）的数据大小的比值超过设置的阈值garbage_fraction（默认值0.5）时，才会触发垃圾回收，将这两个文件合并，执行合并的两个文件一定是逻辑上相邻的两个文件。合并逻辑：锁定这两个文件先整理前面的文件的有效数据，再整理后面的文件的有效数据将后面文件的有效数据写入到前面的文件中更新消息在ETS表中的记录删除后面文件队列结构通常队列由rabbit_amqqueue_process和backing_queue这两部分组成，rabbit_amqqueue_process负责协议相关的消息处理，即接收生产者发布的消息、向消费者交付消息、处理消息的确认（包括生产端的confirm和消费端的ack）等。backing_queue是消息存储的具体形式和引擎，并向rabbit_amqqueue_process提供相关的接口以供调用。如果消息投递的目的队列是空的，并且有消费者订阅了这个队列，那么该消息会直接发送给消费者，不会经过队列这一步。当消息无法直接投递给消费者时，需要暂时将消息存入队列，以便重新投递。rabbit_variable_queue.erl 源码中定义了RabbitMQ队列的4种状态：alpha：消息索引和消息内容都存内存，最耗内存，很少消耗CPUbeta：消息索引存内存，消息内存存磁盘gama：消息索引内存和磁盘都有，消息内容存磁盘delta：消息索引和内容都存磁盘，基本不消耗内存，消耗更多CPU和I/O操作消息存入队列后，不是固定不变的，它会随着系统的负载在队列中不断流动，消息的状态会不断发送变化。持久化的消息，索引和内容都必须先保存在磁盘上，才会处于上述状态中的一种gama状态只有持久化消息才会有的状态。在运行时，RabbitMQ会根据消息传递的速度定期计算一个当前内存中能够保存的最大消息数量（target_ram_count），如果alpha状态的消息数量大于此值，则会引起消息的状态转换，多余的消息可能会转换到beta、gama或者delta状态。区分这4种状态的主要作用是满足不同的内存和CPU需求。对于普通没有设置优先级和镜像的队列来说，backing_queue的默认实现是rabbit_variable_queue，其内部通过5个子队列Q1、Q2、delta、Q3、Q4来体现消息的各个状态。消费者获取消息也会引起消息的状态转换。当消费者获取消息时首先会从Q4中获取消息，如果获取成功则返回。如果Q4为空，则尝试从Q3中获取消息，系统首先会判断Q3是否为空，如果为空则返回队列为空，即此时队列中无消息。如果Q3不为空，则取出Q3中的消息；进而再判断此时Q3和Delta中的长度，如果都为空，则可以认为 Q2、Delta、 Q3、Q4 全部为空，此时将Q1中的消息直接转移至Q4，下次直接从Q4 中获取消息。如果Q3为空，Delta不为空，则将Delta的消息转移至Q3中，下次可以直接从Q3中获取消息。在将消息从Delta转移到Q3的过程中，是按照索引分段读取的，首先读取某一段，然后判断读取的消息的个数与Delta中消息的个数是否相等，如果相等，则可以判定此时Delta中己无消息，则直接将Q2和刚读取到的消息一并放入到Q3中，如果不相等，仅将此次读取到的消息转移到Q3。这里就有两处疑问，第一个疑问是：为什么Q3为空则可以认定整个队列为空？试想一下，如果Q3为空，Delta不为空，那么在Q3取出最后一条消息的时候，Delta 上的消息就会被转移到Q3这样与 Q3 为空矛盾；如果Delta 为空且Q2不为空，则在Q3取出最后一条消息时会将Q2的消息并入到Q3中，这样也与Q3为空矛盾；在Q3取出最后一条消息之后，如果Q2、Delta、Q3都为空，且Q1不为空时，则Q1的消息会被转移到Q4，这与Q4为空矛盾。其实这一番论述也解释了另一个问题：为什么Q3和Delta都为空时，则可以认为 Q2、Delta、Q3、Q4全部为空？通常在负载正常时，如果消费速度大于生产速度，对于不需要保证可靠不丢失的消息来说，极有可能只会处于alpha状态。对于持久化消息，它一定会进入gamma状态，在开启publisher confirm机制时，只有到了 gamma 状态时才会确认该消息己被接收，若消息消费速度足够快、内存也充足，这些消息也不会继续走到下一个状态。为什么消息的堆积导致性能下降？在系统负载较高时，消息若不能很快被消费掉，这些消息就会进入到很深的队列中去，这样会增加处理每个消息的平均开销。因为要花更多的时间和资源处理“堆积”的消息，如此用来处理新流入的消息的能力就会降低，使得后流入的消息又被积压到很深的队列中，继续增大处理每个消息的平均开销，继而情况变得越来越恶化，使得系统的处理能力大大降低。应对这一问题一般有3种措施：增加prefetch_count的值，即一次发送多条消息给消费者，加快消息被消费的速度。采用multiple ack，降低处理 ack 带来的开销流量控制安装和配置RabbitMQ安装环境：虚拟机软件：VMWare 15.1.0操作系统：CentOS Linux release 7.7.1908Erlang：erlang-23.0.2-1.el7.x86_64RabbitMQ：rabbitmq-server-3.8.4-1.el7.noarchRabbitMQ的安装需要首先安装Erlang，因为它是基于Erlang的VM运行的。RabbitMQ需要的依赖：socat和logrotate，logrotate操作系统中已经存在了，只需要安装socat就可以了。RabbitMQ与Erlang的兼容关系详见：https://www.rabbitmq.com/which-erlang.html安装依赖：yum install socat -y安装Erlangerlang-23.0.2-1.el7.x86_64.rpm下载地址：https://github.com/rabbitmq/erlang-rpm/releases/download/v23.0.2/erlang-23.0.2-1.el7.x86_64.rpm首先将erlang-23.0.2-1.el7.x86_64.rpm上传至服务器，然后执行下述命令：rpm -ivh erlang-23.0.2-1.el7.x86_64.rpm安装RabbitMQrabbitmq-server-3.8.4-1.el7.noarch.rpm下载地址：https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.5/rabbitmq-server-3.8.5-1.el7.noarch.rpm首先将rabbitmq-server-3.8.4-1.el7.noarch.rpm上传至服务器，然后执行下述命令：rpm -ivh rabbitmq-server-3.8.4-1.el7.noarch.rpm启用RabbitMQ的管理插件rabbitmq-plugins enable rabbitmq_management开启RabbitMQsystemctl start rabbitmq-server或者rabbitmq-server或者后台启动rabbitmq-server -detached添加用户rabbitmqctl add_user root 123456给用户添加权限给root用户在虚拟主机"/"上的配置、写、读的权限rabbitmqctl set_permissions root -p / ".*" ".*" ".*"给用户设置标签rabbitmqctl set_user_tags root administrator用户的标签和权限：column1column2(None)没有访问management插件的权限management可以使用消息协议做任何操作的权限，加上：可以使用AMQP协议登录的虚拟主机的权限；查看它们能登录的所有虚拟主机中所有队列、交换器和绑定的权限；查看和关闭它们自己的通道和连接的权限；查看它们能访问的虚拟主机中的全局统计信息，包括其他用户的活动policymaker所有management标签可以做的，加上：在它们能通过AMQP协议登录的虚拟主机上，查看、创建和删除策略以及虚拟主机参数的权限monitoring所有management能做的，加上：列出所有的虚拟主机，包括列出不能使用消息协议访问的虚拟主机的权限；查看其他用户连接和通道的权限；查看节点级别的数据如内存使用和集群的权限；查看真正的全局所有虚拟主机统计数据的权限administrator所有policymaker和monitoring能做的，加上：创建删除虚拟主机的权限；查看、创建和删除用户的权限；查看、创建和删除权限的权限；关闭其他用户连接的权限打开浏览器，访问http://<安装了CentOS的VMWare虚拟机IP地址>:15672使用刚才创建的用户登录：RabbitMQ常用操作命令# 前台启动Erlang VM和RabbitMQ rabbitmq-server # 后台启动 rabbitmq-server -detached # 停止RabbitMQ和Erlang VM rabbitmqctl stop # 查看所有队列 rabbitmqctl list_queues # 查看所有虚拟主机 rabbitmqctl list_vhosts # 在Erlang VM运行的情况下启动RabbitMQ应用 rabbitmqctl start_app rabbitmqctl stop_app # 查看节点状态 rabbitmqctl status # 查看所有可用的插件 rabbitmq-plugins list # 启用插件 rabbitmq-plugins enable <plugin-name> # 停用插件 rabbitmq-plugins disable <plugin-name> # 添加用户 rabbitmqctl add_user username password # 列出所有用户： rabbitmqctl list_users # 删除用户： rabbitmqctl delete_user username # 清除用户权限： rabbitmqctl clear_permissions -p vhostpath username # 列出用户权限： rabbitmqctl list_user_permissions username # 修改密码： rabbitmqctl change_password username newpassword # 设置用户权限： rabbitmqctl set_permissions -p vhostpath username ".*" ".*" ".*" # 创建虚拟主机: rabbitmqctl add_vhost vhostpath # 列出所以虚拟主机: rabbitmqctl list_vhosts # 列出虚拟主机上的所有权限: rabbitmqctl list_permissions -p vhostpath # 删除虚拟主机: rabbitmqctl delete_vhost vhost vhostpath # 移除所有数据，要在 rabbitmqctl stop_app 之后使用: rabbitmqctl resetRabbitMQ工作流程详解生产者发送消息的流程生产者连接RabbitMQ，建立TCP连接( Connection)，开启信道（Channel）生产者声明一个Exchange（交换器），并设置相关属性，比如交换器类型、是否持久化等生产者声明一个队列井设置相关属性，比如是否排他、是否持久化、是否自动删除等生产者通过bindingKey （绑定Key）将交换器和队列绑定（ binding ）起来生产者发送消息至RabbitMQ Broker，其中包含routingKey （路由键）、交换器等信息相应的交换器根据接收到的routingKey 查找相匹配的队列。如果找到，则将从生产者发送过来的消息存入相应的队列中。如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者关闭信道。关闭连接。消费者接收消息的过程消费者连接到RabbitMQ Broker ，建立一个连接(Connection ) ，开启一个信道(Channel) 。消费者向RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数， 以及做一些准备工作等待RabbitMQ Broker 回应并投递相应队列中的消息， 消费者接收消息。消费者确认( ack) 接收到的消息。RabbitMQ 从队列中删除相应己经被确认的消息。关闭信道。关闭连接。案例Hello World一对一的简单模式。生产者直接发送消息给RabbitMQ，另一端消费。未定义和指定Exchange的情况下，使用的是AMQP default这个内置的Exchange。HelloWorldSender.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; Rabbitmq是一个消息broker：接收消息，传递给下游应用*术语：Producing就是指发送消息，发送消息的程序是ProducerQueue指的是RabbitMQ内部的一个组件，消息存储于queue中。queue使用主机的内存和磁盘存储，收到内存和磁盘空间的限制可以想象为一个大的消息缓冲。很多Producer可以向同一个queue发送消息，很多消费者可以从同一个queue消费消息。Consuming就是接收消息。一个等待消费消息的应用程序称为Consumer*生产者、消费者、队列不必在同一台主机，一般都是在不同的主机上的应用。一个应用可以同时是生产者和消费者。**/public class HelloWorldSender { private static String QUEUE_NAME = "hello";public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); try (Connection conn = factory.newConnection(); Channel channel = conn.createChannel()) { channel.queueDeclare(QUEUE_NAME, false, false, true, null); String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); } catch (Exception e) { } }}HelloWorldReceiver.javapackage com.test.rabbitmq.demo;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;public class HelloWorldReceiver { private final static String QUEUE_NAME = "hello"; public static void main(String[] argv) throws Exception { // 连接工厂 ConnectionFactory factory = new ConnectionFactory(); // 设置服务器主机名或IP地址 factory.setHost("node1"); // 设置Erlang的虚拟主机名称 factory.setVirtualHost("/"); // 设置用户名 factory.setUsername("root"); // 设置密码 factory.setPassword("123456"); // 设置客户端与服务器的通信端口，默认值为5672 factory.setPort(5672); // 获取连接 Connection connection = factory.newConnection(); // 从连接获取通道 Channel channel = connection.createChannel(); // 声明一个队列 // 第一个参数是队列名称，第二个参数false表示在rabbitmq-server重启后就没有了 // 第三个参数表示该队列不是一个排外队列，否则一旦客户端断开，队列就删除了 // 第四个参数表示该队列是否自动删除，true表示一旦不使用了，系统删除该队列 // 第五个参数表示该队列的参数，该参数是Map集合，用于指定队列的属性 // channel.queueDeclare(QUEUE_NAME, false, false, true, null); channel.queueDeclare(QUEUE_NAME, false, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 消息的推送回调函数 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; /**使用服务器生成的consumerTag启动本地，非排他的使用者。启动一个仅提供了basic.deliver和basic.cancel AMQP方法（对大多数情形够用了）第一个参数：队列名称autoAck – true 只要服务器发送了消息就表示消息已经被消费者确认; false服务端等待客户端显式地发送确认消息deliverCallback – 服务端推送过来的消息回调函数cancelCallback – 客户端忽略该消息的回调方法Returns:服务端生成的consumerTag */ channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {}); }}## Connection 和Channel关系 生产者和消费者，需要与RabbitMQ Broker 建立TCP连接，也就是Connection 。一旦TCP连接建立起来，客户端紧接着创建一个AMQP 信道（Channel），每个信道都会被指派一个唯一的ID。信道是建立在Connection 之上的虚拟连接，RabbitMQ 处理的每条AMQP 指令都是通过信道完成的。 ![image.png](/upload/2020/10/image-15c6963326284240a593a71f5539bad4.png) 为什么不直接使用TCP连接，而是使用信道？ RabbitMQ 采用类似NIO的做法，复用TCP 连接，减少性能开销，便于管理。 当每个信道的流量不是很大时，复用单一的Connection 可以在产生性能瓶颈的情况下有效地节省TCP连接资源。 当信道本身的流量很大时，一个Connection 就会产生性能瓶颈，流量被限制。需要建立多个Connection，分摊信道。具体的调优看业务需要。 信道在AMQP中是一个很重要的概念，大多数操作都是在信道这个层面进行的。channel.exchangeDeclarechannel.queueDeclarechannel.basicPublishchannel.basicConsume// ...RabbitMQ相关的API与AMQP紧密相连，比如channel.basicPublish对应AMQP的Basic.Publish命令。 # RabbitMQ工作模式详解 官网地址：https://www.rabbitmq.com/getstarted.htm ## Work Queue 生产者发消息，启动多个消费者实例来消费消息，每个消费者仅消费部分信息，可达到负载均衡的效果。 ![image](/upload/2020/10/image-2f9e162208c941b68a3907e9532ccaf1.png) NewTask.javapackage com.test.rabbitmq.demo;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class NewTask { private static final String QUEUE_NAME = "";private static final String[] works = { "hello.", "hello..", "hello...", "hello....", "hello.....", "hello......", "hello.......", "hello........", "hello.........", "hello.........." };public static void main(String[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); try (Connection conn = factory.newConnection(); Channel channel = conn.createChannel()) { channel.queueDeclare("taskQueue", false, false, false, null); for (String work : works) { // BuiltinExchangeType.DIRECT // BuiltinExchangeType.FANOUT // BuiltinExchangeType.HEADERS // BuiltinExchangeType.TOPIC // channel.exchangeDeclare("ex1", BuiltinExchangeType.DIRECT); // 将消息路由到taskQueue队列 channel.basicPublish("", "taskQueue", null, work.getBytes("UTF-8")); // channel.basicPublish("", "hello", MessageProperties.PERSISTENT_TEXT_PLAIN, work.getBytes()); System.out.println(" [x] Sent '" + work + "'");} } catch (Exception e) { e.printStackTrace(); } }}Worker.javapackage com.test.rabbitmq.demo;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Worker { private static final String TASK_QUEUE_NAME = "taskQueue";public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); // true表示不需要手动确认消息，false表示需要手动确认消息： channel.basicAck(xxx, yyy); Boolean autoAck = true; Connection conn = factory.newConnection(); Channel channel = conn.createChannel(); // 预取指定个数的消息。此处每次获取一个消息 // channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String task = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + task + "'"); try { doWork(task);} catch (InterruptedException e) { e.printStackTrace();} finally { System.out.println(" [x] Done"); // 手动确认消息 // channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} }; channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null); channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {}); }private static void doWork(String task) throws InterruptedException { System.out.println("task = " + task); for (char ch : task.toCharArray()) { if (ch == '.') Thread.sleep(1000); } }}## 发布订阅模式 使用fanout类型交换器，routingKey忽略。每个消费者定义生成一个队列并绑定到同一个Exchange，每个消费者都可以消费到完整的消息。 消息广播给所有订阅该消息的消费者。 在RabbitMQ中，生产者不是将消息直接发送给消息队列，实际上生产者根本不知道一个消息被发送到哪个队列。 生产者将消息发送给交换器。交换器非常简单，从生产者接收消息，将消息推送给消息队列。交换器必须清楚地知道要怎么处理接收到的消息。应该是追加到一个指定的队列，还是追加到多个队列，还是丢弃。规则就是交换器类型。 ![image.png](/upload/2020/10/image-9727b65f24634bc5be7a2fb2fc1d8f63.png) 交换器的类型前面已经介绍过了： direct 、topic 、headers 和fanout 四种类型。发布订阅使用fanout。创建交换器，名字叫logs：channel.exchangeDeclare("logs", "fanout");fanout 交换器很简单，从名字就可以看出来（用风扇吹出去），将所有收到的消息发送给它知道的所有的队列。列出RabbitMQ的交换器，包括了amq.* 的和默认的（未命名）的交换器。rabbitmqctl list_exchanges未命名交换器在前面的那里中我们没有指定交换器，但是依然可以向队列发送消息。这是因为我们使用了默认的交换器。channel.basicPublish("", "hello", null, message.1 getBytes());第一个参数就是交换器名称，为空字符串。直接使用routingKey向队列发送消息，如果该routingKey指定的队列存在的话。现在，向指定的交换器发布消息：channel.basicPublish("logs", "", null, message.getBytes());临时队列前面我们使用队列的名称，生产者和消费者都是用该名称来发送和接收该队列中的消息。首先，我们无论何时连接RabbitMQ的时候，都需要一个新的，空的队列。我们可以使用随机的名字创建队列，也可以让服务器帮我们生成随机的消息队列名字。其次，一旦我们断开到消费者的连接，该队列应该自动删除。String queueName = channel.queueDeclare().getQueue();上述代码我们声明了一个非持久化的、排他的、自动删除的队列，并且名字是服务器随机生成的。queueName一般的格式类似：amq.gen-JzTY20BRgKO-HjmUJj0wLg绑定在创建了消息队列和fanout 类型的交换器之后，我们需要将两者进行绑定，让交换器将消息发送给该队列。channel.queueBind(queueName, "logs", "");此时，logs交换器会将接收到的消息追加到我们的队列中。可以使用下述命令列出RabbitMQ中交换器的绑定关系：rabbitmqctl list_bindings发布订阅模式的整体代码如下：public class EmitLog { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String message = argv.length < 1 ? "info: Hello World!" : String.join(" ", argv); channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); } } }import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.DeliverCallback; public class ReceiveLogs { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {}); } }当消费者启动起来之后，命令rabbitmqctl list_bindings列出绑定关系：消息的推拉：实现RabbitMQ的消费者有两种模式，推模式（Push）和拉模式（Pull）。实现推模式推荐的方式是继承 DefaultConsumer 基类，也可以使用Spring AMQP的 SimpleMessageListenerContainer。推模式是最常用的，但是有些情况下推模式并不适用的，比如说：由于某些限制，消费者在某个条件成立时才能消费消息 需要批量拉取消息进行处理 实现拉模式 RabbitMQ的Channel提供了 basicGet 方法用于拉取消息。路由模式使用direct类型的Exchange，发N条消费并使用不同的routingKey，消费者定义队列并将队列、routingKey、Exchange绑定。此时使用direct模式Exchagne必须要routingKey完全匹配的情况下消息才会转发到对应的队列中被消费。上一个模式中，可以将消息广播到很多接收者。现在我们想让接收者只接收部分消息，如，我们通过直接模式的交换器将关键的错误信息记录到log文件，同时在控制台正常打印所有的日志信息。绑定上一模式中，交换器的使用方式：channel.queueBind(queueName, EXCHANGE_NAME, "");绑定语句中还有第三个参数：routingKey：channel.queueBind(queueName, EXCHANGE_NAME, "black");bindingKey的作用与具体使用的交换器类型有关。对于fanout类型的交换器，此参数设置无效，系统直接忽略。direct交换器分布式系统中有很多应用，这些应用需要运维平台的监控，其中一个重要的信息就是服务器的日志记录。我们需要将不同日志级别的日志记录交给不同的应用处理。如何解决？使用direct交换器如果要对不同的消息做不同的处理，此时不能使用fanout类型的交换器，因为它只会盲目的广播消息。我们需要使用direct 类型的交换器。direct 交换器的路由算法很简单：只要消息的routingKey和队列的bindingKey对应，消息就可以推送给该队列。上图中的交换器X 是direct 类型的交换器，绑定的两个队列中，一个队列的bindingKey是orange，另一个队列的bindingKey是black和green。如此，则routingKey是orange的消息发送给队列Q1，routingKey是black和green的消息发送给Q2队列，其他消息丢弃。多重绑定上图中，我们使用direct类型的交换器X，建立了两个绑定：队列Q1根据bindingKey的值black绑定到交换器X，队列Q2根据bindingKey的值black绑定到交换器X；交换器X会将消息发送给队列Q1和队列Q2。交换器的行为跟fanout的行为类似，也是广播。在案例中，我们将日志级别作为routingKey。EmitLogsDirect.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class EmitLogsDirect { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); String servrity = null; // 声明direct类型的交换器logs channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT); for (int i = 0; i < 100; i++) { switch (i % 3) { case 0: servrity = "info"; break; case 1: servrity = "warn"; break; case 2: servrity = "error"; break; default: System.err.println("log错误，程序退出"); System.exit(-1); } String logStr = "这是 【" + servrity + "】 的消息"; channel.basicPublish("direct_logs", servrity, null, logStr.getBytes("UTF-8")); } } }ReceiveErrorLogsDirect.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveErrorLogsDirect { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT); String queueName = channel.queueDeclare().getQueue(); // 将logs交换器和queueName队列通过bindingKey：error绑定 channel.queueBind(queueName, "direct_logs", "error"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(queueName, deliverCallback, consumerTag -> {}); } }ReceiveWarnInfoLogsDirect.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveWarnInfoLogsDirect { public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare("direct_logs", BuiltinExchangeType.DIRECT); String queueName = channel.queueDeclare().getQueue(); // 将logs交换器和queueName队列通过bindingKey：warn绑定 channel.queueBind(queueName, "direct_logs", "warn"); // 将logs交换器和queueName队列通过bindingKey：info绑定 channel.queueBind(queueName, "direct_logs", "info"); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(queueName, deliverCallback, consumerTag -> {}); } }主题模式使用topic类型的交换器，队列绑定到交换器、bindingKey时使用通配符，交换器将消息路由转发到具体队列时会根据消息routingKey模糊匹配，比较灵活。上个模式中，我们通过direct类型的交换器做到了根据日志级别的不同，将消息发送给了不同队列的。这里有一个限制，加入现在我不仅想根据日志级别划分日志消息，还想根据日志来源划分日志，怎么做？比如，我想监听cron服务发送的error消息，又想监听从kern服务发送的所有消息。此时可以使用RabbitMQ的主题模式（Topic）。要想topic 类型的交换器，routingKey就不能随便写了，它必须得是点分单词。单词可以随便写，生产中一般使用消息的特征。如：“stock.usd.nyse”，“nyse.vmw”，“quick.orange.rabbit”等。该点分单词字符串最长255字节。bindingKey也必须是这种形式。topic类型的交换器背后原理跟direct 类型的类似：只要队列的bindingKey的值与消息的routingKey匹配，队列就可以收到该消息。有两个不同：*(star)匹配一个单词匹配0到多个单词上图中，我们发送描述动物的消息。消息发送的时候指定的routingKey 包含了三个词，两个点。第一个单词表示动物的速度，第二个是颜色，第三个是物种：..。创建三个绑定：Q1绑定到" .orange. "Q2绑定到" ..rabbit "和" lazy.# "。Q1关注orange颜色动物的消息Q2关注兔子的消息，以及所有懒的动物消息如果不能匹配，就丢弃消息。如果发送的消息routingKey是" lazy.orange.male.rabbit "，则会匹配最后一个绑定。如果在topic类型的交换器中bindingKey使用# 则就是fanout类型交换器的行为。如果在topic类型的交换器中bindingKey中不使用* 和# 则就是direct类型交换器的行为。EmitLogTopic.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.Random; import java.util.concurrent.TimeoutException; public class EmitLogTopic { private static final String EXCHANGE_NAME = "topic_logs"; private static final String[] SPEED = { "lazy", "quick", "normal" }; private static final String[] COLOR = { "black", "orange", "red", "yellow", "blue", "white", "pink" }; private static final String[] SPECIES = { "dog", "rabbit", "chicken", "horse", "bear", "cat" }; private static final Random RANDOM = new Random(); public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String message = null; String routingKey = null; String speed = null; String color = null; String species = null; for (int i = 0; i < 10; i++) { speed = SPEED[RANDOM.nextint(SPEED.length)]; color = COLOR[RANDOM.nextint(COLOR.length)]; species = SPECIES[RANDOM.nextint(SPECIES.length)]; message = speed + "-" + color + "-" + species; routingKey = speed + "." + color + "." + species; channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes()); } System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'"); } }EmitLogTopic1.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.Random; import java.util.concurrent.TimeoutException; public class EmitLogTopic1 { private static final String EXCHANGE_NAME = "topic_logs"; private static final String[] SPECIES = { "dog", "rabbit", "chicken", "horse", "bear", "cat" }; private static final Random RANDOM = new Random(); public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String message = null; String routingKey = null; String speed = null; String species = null; for (int i = 0; i < 10; i++) { speed = "lazy"; species = SPECIES[RANDOM.nextint(SPECIES.length)]; message = speed + "-" + species; routingKey = speed + "." + species; channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes()); } System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'"); } }ReceiveLogTopic.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveLogsTopic { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, "*.*.rabbit"); DeliverCallback callback = (consumerTag, message) -> { System.out.println("*.*.rabbit 匹配到的消息：" + new String(message.getBody(), "UTF-8")); }; channel.basicConsume(queueName, true, callback, consumerTag -> {}); } }ReceiveLogTopic1.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveLogsTopic1 { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*"); DeliverCallback callback = (consumerTag, message) -> { System.out.println("*.orange.* 匹配到的消息：" + new String(message.getBody(), "UTF-8")); }; channel.basicConsume(queueName, true, callback, consumerTag -> {}); } }ReceiveLopTopic2.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveLogsTopic2 { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, "lazy.*.*"); DeliverCallback callback = (consumerTag, message) -> { System.out.println("lazy.*.* 匹配到的消息：" + new String(message.getBody(), "UTF-8")); }; channel.basicConsume(queueName, true, callback, consumerTag -> {}); } }ReceiveLogTopic3.javapackage com.test.rabbitmq.demo; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveLogsTopic3 { private static final String EXCHANGE_NAME = "topic_logs"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("node1"); factory.setVirtualHost("/"); factory.setUsername("root"); factory.setPassword("123456"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, "lazy.#"); DeliverCallback callback = (consumerTag, message) -> { System.out.println("lazy.# 匹配到的消息：" + new String(message.getBody(), "UTF-8")); }; channel.basicConsume(queueName, true, callback, consumerTag -> {}); } }Spring整合RabbitMQspring-amqp是对AMQP的一些概念的一些抽象，spring-rabbit是对RabbitMQ操作的封装实现。主要有几个核心类RabbitAdmin、RabbitTemplate、SimpleMessageListenerContainer等RabbitAdmin类完成对Exchange，Queue，Binding的操作，在容器中管理了RabbitAdmin类的时候，可以对Exchange，Queue，Binding进行自动声明。RabbitTemplate类是发送和接收消息的工具类。SimpleMessageListenerContainer是消费消息的容器。目前比较新的一些项目都会选择基于注解方式，而比较老的一些项目可能还是基于配置文件的。基于配置文件的整合创建maven项目配置pom.xml，添加rabbit的spring依赖<dependencies> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.2.7.RELEASE</version> </dependency> </dependencies>rabbit-context.xml<?xml version="1.0" encoding="UTF-8" ?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"> <rabbit:connection-factory id="connectionFactory" host="node1" virtual-host="/" username="root" password="123456" port="5672" /> <rabbit:template id="amqpTemplate" connectionfactory="connectionFactory" /> <rabbit:admin id="rabbitAdmin" connectionfactory="connectionFactory" /> <rabbit:queue name="myqueue" /> <rabbit:direct-exchange name="direct.biz.ex" autodeclare="true" auto-delete="false" durable="false"> <rabbit:bindings> <rabbit:binding queue="myqueue" key="dir.ex"> </rabbit:binding> </rabbit:bindings> </rabbit:direct-exchange> </beans>Application.javapackage com.test.rabbitmq.demo; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.GenericXmlApplicationContext; 使用spring xml配置的方式发送接接收消息*/public class App { public static void main(String[] args) { AbstractApplicationContext context = new GenericXmlApplicationContext("classpath:/rabbit-context.xml"); AmqpTemplate template = context.getBean(AmqpTemplate.class); for (int i = 0; i < 1000; i++) { // 第一个参数是路由key，第二个参数是消息 template.convertAndSend("dir.ex", "foo" + i); } // 主动从队列拉取消息 String foo = (String) template.receiveAndConvert("myqueue"); System.out.println(foo); context.close(); }}启动RabbitMQ之后，直接运行即可。创建maven项目配置pom.xml，添加rabbit的spring依赖<dependencies> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>2.2.7.RELEASE</version> </dependency> </dependencies>添加配置类RabbitConfiguration.javapackage com.test.rabbitmq.demo; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitConfiguration { @Bean public com.rabbitmq.client.ConnectionFactory rabbitFactory() { com.rabbitmq.client.ConnectionFactory rabbitFactory = new com.rabbitmq.client.ConnectionFactory(); rabbitFactory.setHost("node1"); rabbitFactory.setVirtualHost("/"); rabbitFactory.setUsername("root"); rabbitFactory.setPassword("123456"); rabbitFactory.setPort(5672); return rabbitFactory; } @Bean public ConnectionFactory connectionFactory(com.rabbitmq.client.ConnectionFactory rabbitFactory) { ConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitFactory); return connectionFactory; } @Bean public AmqpAdmin amqpAdmin(ConnectionFactory factory) { AmqpAdmin amqpAdmin = new RabbitAdmin(factory); return amqpAdmin; } @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory factory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(factory); return rabbitTemplate; } @Bean public Queue queue() { Queue myqueue = new Queue("myqueue"); return myqueue; } }主入口类App.javapackage com.test.rabbitmq.demo; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.support.AbstractApplicationContext; 使用spring的注解方式发送和接收消息*/public class SpringAnnotationDemo { public static void main(String[] args) { AbstractApplicationContext context = new AnnotationConfigApplicationContext(RabbitConfiguration.class); AmqpTemplate template = context.getBean(AmqpTemplate.class); template.convertAndSend("myqueue", "foo"); String foo = (String) template.receiveAndConvert("myqueue"); System.out.println(foo); context.close(); }}添加starter依赖<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>application.properties中添加连接信息spring.application.name=springboot_rabbitmq spring.rabbitmq.host=node1 spring.rabbitmq.virtual-host=/ spring.rabbitmq.username=root spring.rabbitmq.password=123456 spring.rabbitmq.port=5672主入口类package com.test.rabbitmq.demo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class RabbitqmDemo { public static void main(String[] args) { SpringApplication.run(RabbitqmDemo.class, args); } }RabbitConfig类package com.test.rabbitmq.demo.config; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitConfig { @Bean public Queue myQueue() { return new Queue("myqueue"); } @Bean public Exchange myExchange() { // new Exchange() // return new TopicExchange("topic.biz.ex", false, false, null); // return new DirectExchange("direct.biz.ex", false, false, null); // return new FanoutExchange("fanout.biz.ex", false, false, null); // return new HeadersExchange("header.biz.ex", false, false, null); // 交换器名称，交换器类型（），是否是持久化的，是否自动删除，交换器属性Map集合 // return new CustomExchange("custom.biz.ex", ExchangeTypes.DIRECT, false, false, null); return new DirectExchange("myex", false, false, null); } @Bean public Binding myBinding() { // 绑定的目的地，绑定的类型：到交换器还是到队列，交换器名称，路由key，绑定的属性 // new Binding("", Binding.DestinationType.EXCHANGE, "", "", null); // 绑定的目的地，绑定的类型：到交换器还是到队列，交换器名称，路由key，绑定的属性 // new Binding("", Binding.DestinationType.QUEUE, "", "", null); // 绑定了交换器direct.biz.ex到队列myqueue，路由key是direct.biz.ex return new Binding("myqueue", Binding.DestinationType.QUEUE, "myex", "direct.biz.ex", null); } }使用RestController发送消息package com.test.rabbitmq.demo; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @RestController public class HelloController { @Autowired private AmqpTemplate rabbitTemplate; @RequestMapping("/send/{message}") public String sendMessage(@PathVariable String message) { rabbitTemplate.convertAndSend("myex", "direct.biz.ex", message); return "ok"; } }使用监听器，用于推消息package com.test.rabbitmq.demo; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Component public class HelloConsumer { @RabbitListener(queues = "myqueue") public void service(String message) { System.out.println("消息队列推送来的消息：" + message); } }
架构详解storage server：存储服务器（又称存储节点或数据服务器），文件和文件属性（meta data）都保存到存储服务器上。Storage server直接利用OS的文件系统调用管理文件。Storage server（后简称storage）以组（卷，group或volume）为单位组织，一个group内包含多台storage机器，数据互为备份，存储空间以group内容量最小的storage为准，所以建议group内的多个storage尽量配置相同，以免造成存储空间的浪费。以group为单位组织存储能方便的进行应用隔离、负载均衡、副本数定制（group内storage server数量即为该group的副本数），比如将不同应用数据存到不同的group就能隔离应用数据，同时还可根据应用的访问特性来将应用分配到不同的group来做负载均衡；缺点是group的容量受单机存储容量的限制，同时当group内有机器坏掉时，数据恢复只能依赖group内的其它机器，使得恢复时间会很长。group内每个storage的存储依赖于本地文件系统，storage可配置多个数据存储目录，比如有10块磁盘，分别挂载在/data/disk1-/data/disk10 ，则可将这10个目录都配置为storage的数据存储目录。storage接受到写文件请求时，会根据配置好的规则，选择其中一个存储目录来存储文件。为了避免单个目录下的文件数太多，在storage第一次启动时，会在每个数据存储目录里创建2级子目录，默认每级256个，总共65536个文件，新写的文件会以hash的方式被路由到其中某个子目录下，然后将文件数据直接作为一个本地文件存储到该目录中。group：组，也可称为卷。同组内服务器上的文件是完全相同的，同一组内的storage server之间是对等的，文件上传、删除等操作可以在任意一台storage server上进行。meta data：文件相关属性，键值对（ Key Value Pair）方式，如：width: 1118。tracker server：跟踪服务器，主要做调度工作，起负载均衡的作用。在内存中记录集群中所有存储组和存储服务器的状态信息，是客户端和数据服务器交互的枢纽。因为不记录文件索引信息，所以占用的内存量很少。Tracker是FastDFS的协调者，负责管理所有的storage server和group，每个storage在启动后会连接Tracker，告知自己所属的group等信息，并保持周期性的心跳，tracker根据storage的心跳信息，建立group==>[storage server list]的映射表。Tracker需要管理的元信息很少，会全部存储在内存中；另外tracker上的元信息都是由storage汇报的信息生成的，本身不需要持久化任何数据，这样使得tracker非常容易扩展，直接增加tracker机器即可扩展为tracker cluster来服务，cluster里每个tracker之间是完全对等的，所有的tracker都接受stroage的心跳信息，生成元数据信息来提供读写服务。client：客户端，作为业务请求的发起方，通过专有接口，使用TCP/IP协议与跟踪器服务器或存储节点进行数据交互。FastDFS向使用者提供基本文件访问接口，比如upload、download、append、delete等，以客户端库的方式提供给用户使用。设计理念轻量级FastDFS服务端只有两个角色：Tracker server和Storage server。Tracker server在内存中记录分组和Storage server的状态等信息，不记录文件索引信息，占用的内存量很少。另外，客户端（应用）和Storage server访问Tracker server时，Tracker server扫描内存中的分组和Storage server状态信息，然后给出应答。由此可以看出Tracker server非常轻量化，不会成为系统瓶颈。FastDFS中的Storage server直接利用OS的文件系统存储文件。FastDFS不会对文件进行分块存储，客户端上传的文件和Storage server上的文件一一对应。对于互联网应用，文件分块存储没有多大的必要。它既没有带来多大的好处，又增加了系统的复杂性。FastDFS不对文件进行分块存储，与支持文件分块存储的DFS相比，更加简洁高效，并且完全能满足绝大多数互联网应用的实际需要。在FastDFS中，客户端上传文件时，文件ID不是由客户端指定，而是由Storage server生成后返回给客户端的。文件ID中包含了组名、文件相对路径和文件名，Storage server可以根据文件ID直接定位到文件。因此FastDFS集群中根本不需要存储文件索引信息，这是FastDFS比较轻量级的一个例证。而其他文件系统则需要存储文件索引信息，这样的角色通常称作NameServer。其中mogileFS采用MySQL数据库来存储文件索引以及系统相关的信息，其局限性显而易见，MySQL将成为整个系统的瓶颈。FastDFS轻量级的另外一个体现是代码量较小。最新的V2.0包括了C客户端API、FastDHT客户端API和PHP extension等，代码行数不到5.2万行。分组存储FastDFS采用了分组存储方式。集群由一个或多个组构成，集群存储总容量为集群中所有组的存储容量之和。一个组由一台或多台存储服务器组成，同组内的多台Storage server之间是对等的互备关系。文件上传、下载、删除等操作可以在组内任意一台Storage server上进行。类似木桶短板效应，一个组的存储容量为该组内存储服务器容量最小的那个，由此可见组内存储服务器的软硬件配置最好是一致的。用分组存储方式的好处是灵活、可控性较强。比如上传文件时，可以由客户端直接指定上传到的组。一个分组的存储服务器访问压力较大时，可以在该组增加存储服务器来扩充服务能力（纵向扩容）。当系统容量不足时，可以增加组来扩充存储容量（横向扩容）。采用这样的分组存储方式，可以使用FastDFS对文件进行管理，使用主流的Web server如Apache、nginx等进行文件下载。对等结构FastDFS集群中的Tracker server也可以有多台，Tracker server和Storage server均不存在单点问题。Tracker server之间是对等关系，组内的Storage server之间也是对等关系。传统的Master-Slave结构中的Master是单点，写操作仅针对Master。如果Master失效，需要将Slave提升为Master，实现逻辑会比较复杂。和Master-Slave结构相比，对等结构中所有结点的地位是相同，每个结点都是Master，不存在单点问题。FastDFS 功能原理文件上传文件上传流程文件上传内部原理选择tracker server和group当集群中不止一个tracker server时，由于tracker之间是完全对等的关系，客户端在upload文件时可以任意选择一个trakcer。 当tracker接收到upload_file的请求时，会为该文件分配一个可以存储该文件的group，使用store_lookup选择group的规则：0、Round robin，所有的group间轮询1、Specified group，指定某一个确定的group2、Load balance，剩余存储空间多的group优先选择storage server当选定group后，tracker会在group内选择一个storage server给客户端，使用store_server选择storage的规则：0、Round robin，在group内的所有storage间轮询1、First server ordered by ip，按ip排序2、First server ordered by priority，按优先级排序（优先级在storage上配置）选择storage path当分配好storage server后，客户端将向storage发送写文件请求，storage将会为文件分配一个数据存储目录 storage server可以有多个存放文件的存储路径（可以理解为多个磁盘），store_path支持如下规则：0、Round robin，多个存储目录间轮询2、剩余存储空间最多的优先生成文件名选定存储目录之后，storage会为文件生一个文件名，由storage server ip、文件创建时间、文件大小、文件crc32和一个随机数拼接而成，然后将这个二进制串进行base64编码，转换为可打印的字符串。选择两级目录 当选定存储目录之后，storage会为文件分配一个文件名，每个存储目录下有两级256*256的子目录，storage会按文件fileid进行两次hash，路由到其中一个子目录，然后将文件以这个文件标示为文件名存储到该子目录下。返回文件id当文件存储到某个子目录后，即认为该文件存储成功，接下来会为该文件返回一个文件id，由group、存储目录、两级子目录、内部文件名、文件后缀名（由客户端指定，主要用于区分文件类型）拼接而成。group1/M00/00/00/xxx.png组名：文件上传后所在的存储组名称，在文件上传成功后有存储服务器返回，需要客户端自行保存。虚拟磁盘路径：存储服务器配置的虚拟路径，与磁盘选项store_path*对应。数据两级目录：存储服务器在每个虚拟磁盘路径下创建的两级目录，用于存储数据文件。文件名：与文件上传时不同。是由存储服务器根据特定信息生成，文件名包含：源存储服务器IP地址、文件创建时间戳、文件大小、随机数和文件拓展名等信息。文件下载客户端带上文件名信息请求Tracker服务获取到存储服务器的ip地址和端口，然后客户端根据返回的IP地址和端口号请求下载文件，存储服务器接收到请求后返回文件给客户端。跟upload_file一样，在download_file时客户端可以选择任意tracker server。客户端发送download请求给某个tracker，必须带上文件名信息，tracke从文件名中解析出文件的group、大小、创建时间等信息，然后为该请求选择一个storage用来服务读请求。选择哪个 storage server 作为下载服务器 使用download_server 规则如下:0: 轮询方式，可以下载当前文件的任意一个 storage server进行轮询1: 哪个为源storage server 就用哪个由于group内的文件同步时在后台异步进行的，所以有可能出现在读到时候，文件还没有同步到某些storage server,为了尽量避免访问到这样的storage，会有相应的文件同步规则。文件同步文件同步原理写文件时，客户端将文件写至group内一个storage server即认为写文件成功，storage server写完文件后，会由后台线程将文件同步至同group内其他的storage server。每个storage写文件后，同时会写一份binlog，binlog里不包含文件数据，只包含文件名等元信息，这份binlog用于后台同步，storage会记录向group内其他storage同步的进度，以便重启后能接上次的进度继续同步。进度以时间戳的方式进行记录，所以最好能保证集群内所有server的时钟保持同步。storage的同步进度会作为元数据的一部分汇报到tracker上，tracke在选择storage的时候会以同步进度作为参考。比如一个group内有A、B、C 三个storage server，A向C同步到进度为T1 ，B向C同步到时间戳为T2（T2 > T1)，tracker接收到这些同步进度信息时，就会进行整理，将最小的那个做为C的同步时间戳，本例中T1即为C的同步时间戳（即所有T1以前写的数据都已经同步到C上了）。同理，根据上述规则，tracker会为A、B生成一个同步时间戳。tracker选择group内可用的storage的规则该文件上传到的源头storage源头storage只要存活着，肯定包含这个文件，源头的地址被编码在文件名中。文件创建时间戳==storage被同步到的时间戳 且(当前时间-文件创建时间戳) > 文件同步最大时间（如5分钟)文件创建后，认为经过最大同步时间后，肯定已经同步到其他storage了。文件创建时间戳 < storage被同步到的时间戳。同步时间戳之前的文件确定已经同步了(当前时间 - 文件创建时间戳) > 同步延迟阀值。经过同步延迟阈值时间，认为文件肯定已经同步了。文件删除删除处理流程与文件下载类似：Client询问Tracker server可以删除指定文件的Storage server，参数为文件ID（包含组名和文件名）。Tracker server返回一台可用的Storage server。Client直接和该Storage server建立连接，完成文件删除。文件删除API：delete_file断点续传提供appender file的支持，通过upload_appender_file接口完成，appender file允许在创建后，对该文件进行append操作。实际上，appender file与普通文件的存储方式是相同的，不同的是，appender file不能被合并存储到trunk file。续传涉及到的文件大小MD5不会改变。续传流程与文件上传类似，先定位到源storage，完成完整或部分上传，再通过binlog进行同group内server文件同步。断点续传的API:upload_appender_file文件HTTP访问支持FastDFS的tracker和storage都内置了http协议的支持，客户端可以通过http协议来下载文件，tracker在接收到请求时，通过http的redirect机制将请求重定向至文件所在的storage上。除了内置的http协议外，FastDFS还提供了通过apache或nginx扩展模块下载文件的支持。
MongoDB逻辑结构MongoDB 与 MySQL 中的架构相差不多，底层都使用了可插拔的存储引擎以满足用户的不同需要。用户可以根据程序的数据特征选择不同的存储引擎,在最新版本的 MongoDB 中使用了 WiredTiger 作为默认的存储引擎，WiredTiger 提供了不同粒度的并发控制和压缩机制，能够为不同种类的应用提供了最好的性能和存储率。在存储引擎上层的就是 MongoDB 的数据模型和查询语言了，由于 MongoDB 对数据的存储与 RDBMS 有较大的差异，所以它创建了一套不同的数据模型和查询语言。MongoDB的数据模型描述数据模型内嵌内嵌的方式指的是把相关联的数据保存在同一个文档结构之中。MongoDB的文档结构允许一个字段或者一个数组内的值作为一个嵌套的文档。引用引用方式通过存储数据引用信息来实现两个不同文档之间的关联,应用程序可以通过解析这些数据引用来访问相关数据。如何选择数据模型选择内嵌:1）数据对象之间有包含关系 ,一般是数据对象之间有一对多或者一对一的关系 。2）需要经常一起读取的数据。3）有 map-reduce/aggregation 需求的数据放在一起，这些操作都只能操作单个 collection。选择引用:1）当内嵌数据会导致很多数据的重复，并且读性能的优势又不足于覆盖数据重复的弊端 。2）需要表达比较复杂的多对多关系的时候 。3）大型层次结果数据集 嵌套不要太深。MongoDB存储引擎存储引擎概述存储引擎是MongoDB的核心组件，负责管理数据如何存储在硬盘和内存上。MongoDB支持的存储引擎有MMAPv1 ,WiredTiger和InMemory。InMemory存储引擎用于将数据只存储在内存中，只将少量的元数据(meta-data)和诊断日志（Diagnostic）存储到硬盘文件中，由于不需要Disk的IO操作，就能获取所需的数据，InMemory存储引擎大幅度降低了数据查询的延迟（Latency）。从mongodb3.2开始默认的存储引擎是WiredTiger,3.2版本之前的默认存储引擎是MMAPv1，mongodb4.x版本不再支持MMAPv1存储引擎。storage: journal: enabled: true dbPath: /data/mongo/ ##是否一个库一个文件夹 directoryPerDB: true ##数据引擎 engine: wiredTiger ##WT引擎配置 WiredTiger: engineConfig: ##WT最大使用cache（根据服务器实际情况调节） cacheSizeGB: 2 ##是否将索引也按数据库名单独存储 directoryForIndexes: true journalCompressor:none （默认snappy） ##表压缩配置 collectionConfig: blockCompressor: zlib (默认snappy,还可选none、zlib) ##索引配置 indexConfig: prefixCompression: trueWiredTiger存储引擎优势文档空间分配方式WiredTiger使用的是BTree存储，MMAPV1是线性存储，需要Padding并发级别WiredTiger文档级别锁，MMAPV1引擎使用表级锁数据压缩snappy (默认) 和 zlib，相比MMAPV1(无压缩) 空间节省数倍。内存使用WiredTiger 可以指定内存的使用大小。Cache使用WT引擎使用了二阶缓存WiredTiger Cache，File System Cache来保证Disk上的数据的最终一致性。而 MMAPv1 只有 journal 日志。WiredTiger引擎包含的文件和作用WiredTiger.basecfg: 存储基本配置信息，与 ConfigServer有关系WiredTiger.lock: 定义锁操作table*.wt: 存储各张表的数据WiredTiger.wt: 存储table* 的元数据WiredTiger.turtle: 存储WiredTiger.wt的元数据journal: 存储WAL(Write Ahead Log)WiredTiger存储引擎实现原理写请求WiredTiger的写操作会默认写入Cache ,并持久化到WAL (Write Ahead Log)，每60s或Log文件达到2G做一次checkpoint (当然我们也可以通过在写入时传入 j: true 的参数强制 journal 文件的同步，writeConcern{ w: , j: , wtimeout: }) 产生快照文件。WiredTiger初始化时，恢复至最新的快照状态，然后再根据WAL恢复数据，保证数据的完整性。Cache是基于BTree的，节点是一个page，root page是根节点，internal page是中间索引节点，leaf page真正存储数据，数据以page为单位读写。WiredTiger采用Copy on write的方式管理写操作（insert、update、delete），写操作会先缓存在cache里，持久化时，写操作不会在原来的leaf page上进行，而是写入新分配的page，每次checkpoint都会产生一个新的root page。checkpoint流程对所有的table进行一次checkpoint，每个table的checkpoint的元数据更新至WiredTiger.wt对WiredTiger.wt进行checkpoint，将该table checkpoint的元数据更新至临时文件WiredTiger.turtle.set将WiredTiger.turtle.set重命名为WiredTiger.turtle。上述过程如果中间失败，WiredTiger在下次连接初始化时，首先将数据恢复至最新的快照状态，然后根据WAL恢复数据，以保证存储可靠性。Journaling在数据库宕机时 , 为保证 MongoDB 中数据的持久性，MongoDB 使用了 Write Ahead Logging 向磁盘上的 journal 文件预先进行写入。除了 journal 日志，MongoDB 还使用检查点（checkpoint）来保证数据的一致性，当数据库发生宕机时，我们就需要 checkpoint 和 journal 文件协作完成数据的恢复工作。在数据文件中查找上一个检查点的标识符在 journal 文件中查找标识符对应的记录重做对应记录之后的全部操作
互联网应用架构发展随着用户群体逐渐扩大，网站的流量成倍增长，常规的单体架构已无法满足请求压力和业务的快速迭代，下面我们以一个招聘系统为例，从最开始的单体架构分析，一步步演变到现在的微服务架构。单体应用架构最初，用户量、数据量规模都比较小，项目所有的功能模块都放在一个工程中编码、编译、打包并且部署在一个Tomcat容器中，这种架构模式就是单体应用架构，这样的架构既简单实用、便于维护，成本又低，成为了当时的主流架构方式。优点：项目前期开发节奏快，团队成员少的时候能够快速迭代。架构简单：MVC架构，只需要借助IDE开发、调试即可。易于测试：只需要通过单元测试或者浏览器完成。易于部署：打包成单一可执行的jar或者打成war包放到容器内启动。缺点：随着不断的功能迭代，单个项目过大，代码杂乱，耦合严重，开发团队逐渐壮大以后，沟通成本变高，如：代码从编译到启动耗时达到3-5分钟。新增业务困难：在冗杂的系统中增加新业务，维护旧功能，全是不可预测的问题。新人来了以后很难接手任务，学习成本高，需要大概一周时间才能上手开发。核心业务与边缘业务混合在一块，出现问题互相影响，如：一个临时活动流量猛涨，机器负载升高就会影响正常的业务服务。业务量上涨之后，单体应用架构进一步丰富变化，比如应用集群部署、使用Nginx进行负载均衡、增加缓存服务器、增加文件服务器、数据库集群并做读写分离等，通过以上措施增强应对高并发的能力、应对一定的复杂业务场景，但依然属于单体应用架构。垂直应用架构为了避免上面提到的那些问题，开始做模块的垂直划分，做垂直划分的原则是基于现有的业务特性来做，第一个核心目标是为了业务之间互不影响，第二个核心目标是在研发团队壮大后提高效率，减少之间的依赖。优点：系统拆分实现了流量分担，解决了并发问题。可以针对不同模块进行优化。方便水平扩展，负载均衡，容错率提高。系统间相互独立，互不影响，新的业务迭代时更加高效。缺点：服务之间相互调用，如果某个服务的端口或者ip地址发生改变，调用的系统得手动修改配置文件。搭建集群之后，实现负载均衡比较复杂，如：内网负载，在迁移机器时会影响调用方的路由，导致线上故障。服务之间调用方式不统一，基于httpclient、webservice，接口协议不统一。服务监控不到位：除了依靠端口、进程的监控，调用的成功率、失败率、总耗时等这些监控指标是没有的。SOA应用架构在做了垂直划分以后，模块随之增多，维护的成本在也变高，一些通用的业务和模块重复的越来越多，为了解决上面提到的接口协议不统一、服务无法监控、服务的负载均衡，引入了阿里巴巴开源的Dubbo ，一款高性能、轻量级的开源Java RPC框架，它提供了三大核心能力：面向接口的远程方法调用，智能容错和负载均衡，以及服务自动注册和发现。SOA (Service-Oriented Architecture)，即面向服务的架构。根据实际业务，把系统拆分成合适的、独立部署的模块，模块之间相互独立（通过Webservice/Dubbo等技术进行通信）。优点： 分布式、松耦合、扩展灵活、可重用。缺点： 服务抽取粒度较大、服务调用方和提供方耦合度较高（接口耦合度）。微服务应用架构微服务架构可以说是SOA架构的一种拓展，这种架构模式下它拆分粒度更小、服务更独立。把应用拆分成为一个个微小的服务，不同的服务可以使用不同的开发语言和存储，服务之间往往通过Restful等轻量级通信。微服务架构关键在于微小、独立、轻量级通信。微服务是在SOA上做的升华，粒度更加细致，微服务架构强调的一个重点是“业务需要彻底的组件化和服务化”。微服务架构和SOA架构相似又不同：微服务架构和SOA架构很明显的一个区别就是服务拆分粒度的不同，但是对于一些架构发展来说，我们所看到的SOA阶段服务拆分粒度相对来说已经比较细了，所以上述SOA到微服务，从服务拆分上来说变化并不大，只是引入了相对完整的新一代的Spring Cloud微服务技术。所以，上述我们看到的都是架构演变的阶段结果，每一个阶段其实都经历了很多变化，服务拆分其实也是经过了从粗到细，并非绝对的一步到位。例：我们在SOA架构的初期，“简历投递模块”和“人才搜索模块”都有简历内容展示的需求，只不过说可能略有区别，一开始在两个模块中各维护了一套简历查询和展示的代码；后期我们将服务更细粒度拆分，拆分出简历基础服务，那么不同模块调用这个基础服务即可。微服务架构体现的思想及优缺点微服务架构设计的核心思想就是“微”，拆分的粒度相对比较小，这样的话每个服务具有单一职责，开发的耦合度就会降低，微小的功能可以独立部署扩展、灵活性强，升级改造影响范围小（比如升级JDK）。微服务架构的优点:微服务很小，便于特定业务功能的聚焦.微服务很小，每个微服务都可以被一个小团队单独实施（开发、测试、部署上线、运维），团队合作在一定程度上解耦，便于实施敏捷开发。微服务很小，便于重用和模块之间的组装。微服务很独立，不同的微服务可以使用不同的语言开发，松耦合。微服务架构下，更容易引入新技术。微服务架构下，我们可以更好的实现DevOps开发运维一体化。微服务架构的缺点微服务架构下，分布式复杂难以管理，当服务数量增加，管理将越加复杂。微服务架构下，分布式链路跟踪难等。微服务架构中的一些概念服务注册与服务发现服务注册： 服务提供者将所提供服务的信息（服务器IP和端口、服务访问协议等）注册/登记到注册中心。服务发现： 服务消费者能够从注册中心获取到较为实时的服务列表，然后根据一定的策略选择一个服务访问。负载均衡负载均衡即将请求压力分配到多个服务器（应用服务器、数据库服务器等），以此来提高服务的性能、可靠性。熔断熔断即断路保护。微服务架构中，如果下游服务因访问压力过大而响应变慢或失败，上游服务为了保护系统整体可用性，可以暂时切断对下游服务的调用。这种牺牲局部，保全整体的措施（即防止雪崩）就叫做熔断。链路追踪微服务架构项目往往拆分成很多个服务，那么一次请求就需要涉及到很多个服务。不同的微服务可能是由不同的团队开发，可能使用不同的编程语言实现，整个项目也有可能部署在了很多服务器上（甚至百台、千台），横跨多个不同的数据中心。所谓链路追踪，就是对一次请求涉及的很多个服务链路进行日志记录、性能监控。每一个请求都有一个唯一的taceid来贯穿整个请求链路的始终，tracid会存储到整个调用链路的上下文中，而每个微服务都会有日志系统来记录日志，此时该请求的tracid也会同时被记录到日志中去，通过对每个微服务的日志进行分析，即可进行链路追踪分析问题。在一个请求贯穿整个链路的过程中，每经过一个微服务的过程称之为块（span），所有的块构成了一个链路，每个块都有一个spanid，用来分析当前微服务的性能指标等。API网关微服务架构下，不同的微服务往往会有不同的访问地址，客户端可能需要调用多个服务的接口才能完成一个业务需求，如果让客户端直接与各个微服务通信可能出现：客户端需要调用不同的url地址，增加了维护调用难度。在一定的场景下，也存在跨域请求的问题（前后端分离就会碰到跨域问题，原本我们在后端采用Cors就能解决，现在利用网关，那么就放在网关这层做好了）。每个微服务都需要进行单独的身份认证。API网关就可以较好的统一处理上述问题，API请求调用统一接入API网关层，由网关转发请求。API网关更专注在安全、路由、流量等问题的处理上（微服务团队专注于处理业务逻辑即可），它的功能比如：统一接入（路由）安全防护（统一鉴权，负责网关访问身份认证验证，与“访问认证中心”通信，实际认证业务逻辑交移“访问认证中心”处理）黑白名单（实现通过IP地址控制禁止访问网关功能，控制访问）协议适配（实现通信协议校验、适配转换的功能）流量管控（限流）⻓短链接支持容错能力（负载均衡）