消息队列的流派

什么是 MQ

有 Broker 的 MQ

重 Topic

轻 Topic

AMQP 中有四种 exchange

• Direct exchange：key 就等于 queue
• Fanout exchange：无视 key，给所有的 queue 都来一份
• Topic exchange：key 可以用“宽字符”模糊匹配 queue
• Headers exchange：无视 key，通过查看消息的头部元数据来决定发给那个
• queue（AMQP 头部元数据非常丰富而且可以自定义）

这种结构的架构给通讯带来了很大的灵活性，我们能想到的通讯方式都可以用这四种exchange 表达出来。如果你需要一个企业数据总线（在乎灵活性）那么 RabbitMQ 绝对的值得一用

无 Broker 的 MQ

一、Kafka介绍

1.Kafka的使用场景

2.Kafka基本概念

首先，让我们来看一下基础的消息(Message)相关术语：

因此，从一个较高的层面上来看，producer通过网络发送消息到Kafka集群，然后consumer 来进行消费，如下图： 

服务端(brokers)和客户端(producer、consumer)之间通信通过 TCP协议 来完成。

二、kafka基本使用

1.安装前的环境准备

• 安装jdk

• 安装zk

• 官网下载kafka的压缩包:http://kafka.apache.org/downloads

• 解压缩至如下路径

  /usr/local/kafka/Copy to clipboardErrorCopied

• 修改配置文件：/usr/local/kafka/kafka2.11-2.4/config/server.properties

  #broker.id属性在kafka集群中必须要是唯一
  broker.id= 0
  #kafka部署的机器ip和提供服务的端口号
  listeners=PLAINTEXT://192.168.65.60:9092
  #kafka的消息存储文件
  log.dir=/usr/local/data/kafka-logs
  #kafka连接zookeeper的地址
  zookeeper.connect= 192.168.65.60:2181Copy to clipboardErrorCopied

  2.启动kafka服务器

进入到bin目录下。使用命令来启动

./kafka-server-start.sh -daemon../config/server.propertiesCopy to clipboardErrorCopied

验证是否启动成功：

进入到zk中的节点看id是 0 的broker有没有存在（上线）

ls /brokers/ids/Copy to clipboardErrorCopied

server.properties核心配置详解：

3.创建主题topic

执行以下命令创建名为“test”的topic，这个topic只有一个partition，并且备份因子也设置为1

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 1 --partitions 1 --topic testCopy to clipboardErrorCopied

查看当前kafka内有哪些topic

./kafka-topics.sh --list --zookeeper 172.16.253.35:2181Copy to clipboardErrorCopied

4.发送消息

./kafka-console-producer.sh --broker-list 172.16.253.38:9092 --topic testCopy to clipboardErrorCopied

5.消费消息

对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输 出， 默认是消费最新的消息 。使用kafka的消费者消息的客户端，从指定kafka服务器的指定 topic中消费消息

方式一：从最后一条消息的偏移量+1开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --topic testCopy to clipboardErrorCopied

方式二：从头开始消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092 --from-beginning --topic testCopy to clipboardErrorCopied

几个注意点：

• 消息会被存储
• 消息是顺序存储
• 消息是有偏移量的
• 消费时可以指明偏移量进行消费

三、Kafka中的关键细节

1.消息的顺序存储

2. 单播消息的实现

./kafka-console-consumer.sh --bootstrap-server 10.31.167.10:9092 --consumer-property group.id=testGroup --topic testCopy to clipboardErrorCopied

3.多播消息的实现

在一些业务场景中需要让一条消息被多个消费者消费，那么就可以使用多播模式。

kafka实现多播，只需要让不同的消费者处于不同的消费组即可。

./kafka-console-consumer.sh --bootstrap-server 10.31.167.10:9092 --consumer-property group.id=testGroup1 --topic test./kafka-console-consumer.sh --bootstrap-server 10.31.167.10:9092 --consumer-property group.id=testGroup2 --topic testCopy to clipboardErrorCopied

4.查看消费组及信息

# 查看当前主题下有哪些消费组
./kafka-consumer-groups.sh --bootstrap-server 10.31.167.10:9092 --list
# 查看消费组中的具体信息：比如当前偏移量、最后一条消息的偏移量、堆积的消息数量
./kafka-consumer-groups.sh --bootstrap-server 172.16.253.38:9092 --describe --group testGroupCopy to clipboardErrorCopied

• Currennt-offset: 当前消费组的已消费偏移量
• Log-end-offset: 主题对应分区消息的结束偏移量(HW)
• Lag: 当前消费组未消费的消息数

四、主题、分区的概念

1.主题Topic

?主题Topic可以理解成是一个类别的名称。

2.partition分区

为一个主题创建多个分区

./kafka-topics.sh --create --zookeeper localhost:2181 --partitions 2 --topic test1Copy to clipboardErrorCopied

可以通过这样的命令查看topic的分区信息

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1Copy to clipboardErrorCopied

分区的作用：

• 可以分布式存储
• 可以并行写

实际上是存在data/kafka-logs/test-0 和 test-1中的0000000.log文件中

小细节：

五、Kafka集群及副本的概念

1.搭建kafka集群， 3 个broker

准备 3 个server.properties文件

每个文件中的这些内容要调整

• server.properties

  broker.id= 0
  listeners=PLAINTEXT://192.168.65.60:
  log.dir=/usr/local/data/kafka-logsCopy to clipboardErrorCopied

• server1.properties

  broker.id= 1
  listeners=PLAINTEXT://192.168.65.60:
  log.dir=/usr/local/data/kafka-logs-Copy to clipboardErrorCopied

• server2.properties

  broker.id= 2
  listeners=PLAINTEXT://192.168.65.60:
  log.dir=/usr/local/data/kafka-logs-Copy to clipboardErrorCopied

  使用如下命令来启动 3 台服务器

./kafka-server-start.sh -daemon../config/server0.properties
./kafka-server-start.sh -daemon../config/server1.properties
./kafka-server-start.sh -daemon../config/server2.propertiesCopy to clipboardErrorCopied

搭建完后通过查看zk中的/brokers/ids 看是否启动成功

2.副本的概念

./kafka-topics.sh --create --zookeeper 172.16.253.35:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic

通过查看主题信息，其中的关键数据：

• replicas：当前副本存在的broker节点
• leader：副本里的概念
  • 每个partition都有一个broker作为leader。
  • 消息发送方要把消息发给哪个broker？就看副本的leader是在哪个broker上面。副本里的leader专⻔用来接收消息。
  • 接收到消息，其他follower通过poll的方式来同步数据。
• follower：leader处理所有针对这个partition的读写请求，而follower被动复制leader，不提供读写（主要是为了保证多副本数据与消费的一致性），如果leader所在的broker挂掉，那么就会进行新leader的选举，至于怎么选，在之后的controller的概念中介绍。

通过kill掉leader后再查看主题情况

# kill掉leader
ps -aux | grep server.properties
kill 17631
# 查看topic情况
./kafka-topics.sh --describe --zookeeper 172.16.253.35:2181 --topic my-replicated-topicCopy to clipboardErrorCopied

isr： 可以同步的broker节点和已同步的broker节点，存放在isr集合中。

3.broker、主题、分区、副本

• kafka集群中由多个broker组成
• 一个broker中存放一个topic的不同partition——副本

4.kafka集群消息的发送

./kafka-console-producer.sh --broker-list 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --topic my-replicated-topicCopy to clipboardErrorCopied

5.kafka集群消息的消费

./kafka-console-consumer.sh --bootstrap-server 172.16.253.38:9092,172.16.253.38:9093,172.16.253.38:9094 --from-beginning --topic my-replicated-topicCopy to clipboardErrorCopied

6.关于分区消费组消费者的细节

消费组中消费者的数量不能比一个topic中的partition数量多，否则多出来的消费者消费不到消息。

六、Kafka的Java客户端-生产者

1.引入依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.4.1</version>
</dependency>Copy to clipboardErrorCopied

2.生产者发送消息的基本实现

#### //消息的发送方
public class MyProducer {private final static String TOPIC_NAME = "my-replicated-topic";public static void main(String[] args) throws ExecutionException,InterruptedException {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.31.167.10:9092,10.31.167.10:9093,10.31.167.10:9094");
//把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());Producer<String, String> producer = new KafkaProducer<String,String>(props);Order order = new Order((long) i, i);
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));
RecordMetadata metadata = producer.send(producerRecord).get();
//=====阻塞=======
System.out.println("同步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" +metadata.offset());Copy to clipboardErrorCopied

3.发送消息到指定分区上

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0 , order.getOrderId().toString(), JSON.toJSONString(order));Copy to clipboardErrorCopied

4.未指定分区，则会通过业务key的hash运算，算出消息往哪个分区上发

//未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, order.getOrderId().toString(), JSON.toJSONString(order));Copy to clipboardErrorCopied

5.同步发送

生产者同步发消息，在收到kafka的ack告知发送成功之前一直处于阻塞状态

//等待消息发送成功的同步阻塞方法
RecordMetadata metadata = producer.send(producerRecord).get();
System.out.println("同步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" +metadata.offset());

6.异步发消息

生产者发消息，发送完后不用等待broker给回复，直接执行下面的业务逻辑。可以提供callback，让broker异步的调用callback，告知生产者，消息发送的结果

//要发送 5 条消息
Order order = new Order((long) i, i);
//指定发送分区
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0 , order.getOrderId().toString(),JSON.toJSONString(order));
//异步回调方式发送消息
producer.send(producerRecord, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {System.err.println("发送消息失败：" +exception.getStackTrace());
}
if (metadata != null) {
System.out.println("异步方式发送消息结果：" + "topic-" +metadata.topic() + "|partition-"+ metadata.partition() + "|offset-" + metadata.offset());}}
});Copy to clipboardErrorCopied

7.关于生产者的ack参数配置

在同步发消息的场景下：生产者发动broker上后，ack会有 3 种不同的选择：

• （ 1 ）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
• （ 2 ）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
• （ 3 ）acks=-1或all： 需要等待 min.insync.replicas(默认为 1 ，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

code:

props.put(ProducerConfig.ACKS_CONFIG, "1");Copy to clipboardErrorCopied

8.其他一些细节

• 发送会默认会重试 3 次，每次间隔100ms
• 发送的消息会先进入到本地缓冲区（32mb），kakfa会跑一个线程，该线程去缓冲区中取16k的数据，发送到kafka，如果到 10 毫秒数据没取满16k，也会发送一次。

七、消费者

1.消费者消费消息的基本实现

public class MyConsumer {
private final static String TOPIC_NAME = "my-replicated-topic";
private final static String CONSUMER_GROUP_NAME = "testGroup";public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"10.31.167.10:9092,10.31.167.10:9093,10.31.167.10:9094");
// 消费分组名
props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
//创建一个消费者的客户端
KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);
// 消费者订阅主题列表
consumer.subscribe(Arrays.asList(TOPIC_NAME));while (true) {
/*
* poll() API 是拉取消息的⻓轮询
*/
ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(),record.offset(), record.key(), record.value());}}}
}Copy to clipboardErrorCopied

2.自动提交offset

• 设置自动提交参数 - 默认

// 是否自动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");Copy to clipboardErrorCopied

消费者poll到消息后默认情况下，会自动向broker的_consumer_offsets主题提交当前主题-分区消费的偏移量。

自动提交会丢消息： 因为如果消费者还没消费完poll下来的消息就自动提交了偏移量，那么此 时消费者挂了，于是下一个消费者会从已提交的offset的下一个位置开始消费消息。之前未被消费的消息就丢失掉了。

3.手动提交offset

• 设置手动提交参数

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");Copy to clipboardErrorCopied

在消费完消息后进行手动提交

• 手动同步提交

  if (records.count() > 0 ) {
  // 手动同步提交offset，当前线程会阻塞直到offset提交成功
  // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
  consumer.commitSync();
  }Copy to clipboardErrorCopied

• 手动异步提交

  if (records.count() > 0 ) {
  // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
  consumer.commitAsync(new OffsetCommitCallback() {
  @Override
  public void onComplete(Map<TopicPartition, OffsetAndMetadata>offsets, Exception exception) {if (exception != null) {System.err.println("Commit failed for " + offsets);System.err.println("Commit failed exception: " +exception.getStackTrace());}}});
  }Copy to clipboardErrorCopied

4.消费者poll消息的过程

• 消费者建立了与broker之间的⻓连接，开始poll消息。
• 默认一次poll 500条消息

props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500 );Copy to clipboardErrorCopied

可以根据消费速度的快慢来设置，因为如果两次poll的时间如果超出了30s的时间间隔，kafka会认为其消费能力过弱，将其踢出消费组。将分区分配给其他消费者。

可以通过这个值进行设置：

props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000 );Copy to clipboardErrorCopied

如果每隔1s内没有poll到任何消息，则继续去poll消息，循环往复，直到poll到消息。如果超出了1s，则此次⻓轮询结束。

ConsumerRecords<String, String> records =consumer.poll(Duration.ofMillis( 1000 ));Copy to clipboardErrorCopied

消费者发送心跳的时间间隔

props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000 );Copy to clipboardErrorCopied

kafka如果超过 10 秒没有收到消费者的心跳，则会把消费者踢出消费组，进行rebalance，把分区分配给其他消费者。

props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000 );Copy to clipboardErrorCopied

5.指定分区消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));Copy to clipboardErrorCopied

6.消息回溯消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME,0 )));Copy to clipboardErrorCopied

7.指定offset消费

consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0 )));
consumer.seek(new TopicPartition(TOPIC_NAME, 0 ), 10 );Copy to clipboardErrorCopied

8.从指定时间点消费

List<PartitionInfo> topicPartitions =consumer.partitionsFor(TOPIC_NAME);
//从 1 小时前开始消费
long fetchDataTime = new Date().getTime() - 1000 * 60 * 60 ;
Map<TopicPartition, Long> map = new HashMap<>();
for (PartitionInfo par : topicPartitions) {map.put(new TopicPartition(TOPIC_NAME, par.partition()),fetchDataTime);
}
Map<TopicPartition, OffsetAndTimestamp> parMap =consumer.offsetsForTimes(map);
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry :parMap.entrySet()) {TopicPartition key = entry.getKey();OffsetAndTimestamp value = entry.getValue();if (key == null || value == null) continue;Long offset = value.offset();System.out.println("partition-" + key.partition() +"|offset-" + offset);System.out.println();//根据消费里的timestamp确定offsetif (value != null) {consumer.assign(Arrays.asList(key));consumer.seek(key, offset);}
}Copy to clipboardErrorCopied

9.新消费组的消费偏移量

• latest(默认) ：只消费自己启动之后发送到主题的消息
• earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)

  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");Copy to clipboardErrorCopied

  八 、Springboot中使用Kafka

1.引入依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>Copy to clipboardErrorCopied

2.配置文件

server:port: 8080
spring:kafka:bootstrap-servers: 172.16.253.21: 9093producer: # 生产者retries: 3 # 设置大于 0 的值，则客户端会将发送失败的记录重新发送batch-size: 16384buffer-memory: 33554432acks: 1# 指定消息key和消息体的编解码方式key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializerconsumer:group-id: default-groupenable-auto-commit: falseauto-offset-reset: earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializermax-poll-records: 500listener:# 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交# RECORD# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交# BATCH# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交# TIME# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交# COUNT# TIME | COUNT　有一个条件满足时提交# COUNT_TIME# 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交# MANUAL# 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种# MANUAL_IMMEDIATEack-mode: MANUAL_IMMEDIATEredis:host: 172.16.253.21Copy to clipboardErrorCopied

3.消息生产者

• 发送消息到指定topic

@RestController
public class KafkaController {private final static String TOPIC_NAME = "my-replicated-topic";@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;@RequestMapping("/send")public void send() {kafkaTemplate.send(TOPIC_NAME, 0 , "key", "this is a msg");}
}Copy to clipboardErrorCopied

4.消息消费者

• 设置消费组，消费指定topic

  @KafkaListener(topics = "my-replicated-topic",groupId = "MyGroup1")
  public void listenGroup(ConsumerRecord<String, String> record,Acknowledgment ack) {String value = record.value();System.out.println(value);System.out.println(record);//手动提交offsetack.acknowledge();
  }Copy to clipboardErrorCopied

• 设置消费组、多topic、指定分区、指定偏移量消费及设置消费者个数。

@KafkaListener(groupId = "testGroup", topicPartitions = {
@TopicPartition(topic = "topic1", partitions = {"0", "1"}),
@TopicPartition(topic = "topic2", partitions = "0",partitionOffsets = @PartitionOffset(partition = "1",initialOffset = "100"))}
,concurrency = "3")//concurrency就是同组下的消费者个数，就是并发消费数，建议小于等于分区总数
public void listenGroup(ConsumerRecord<String, String> record,Acknowledgment ack) {String value = record.value();System.out.println(value);System.out.println(record);//手动提交offsetack.acknowledge();
}Copy to clipboardErrorCopied

九、Kafka集群Controller、Rebalance和HW

1.Controller

• Kafka集群中的broker在zk中创建临时序号节点，序号最小的节点（最先创建的节点）将作为集群的controller，负责管理整个集群中的所有分区和副本的状态：
  • 当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
  • 当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
  • 当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

2.Rebalance机制

前提是：消费者没有指明分区消费。当消费组里消费者和分区的关系发生变化，那么就会触发rebalance机制。

这个机制会重新调整消费者消费哪个分区。

在触发rebalance机制之前，消费者消费哪个分区有三种策略：

• range：通过公示来计算某个消费者消费哪个分区
• 轮询：大家轮着消费
• sticky：在触发了rebalance后，在消费者消费的原分区不变的基础上进行调整。

3.HW和LEO

十、Kafka线上问题优化

1.如何防止消息丢失

• 发送方： ack是 1 或者-1/all 可以防止消息丢失，如果要做到99.9999%，ack设成all，把min.insync.replicas配置成分区备份数
• 消费方：把自动提交改为手动提交。

2.如何防止消息的重复消费

幂等性如何保证：

• mysql 插入业务id作为主键，主键是唯一的，所以一次只能插入一条
• 使用redis或zk的分布式锁（主流的方案）

3.如何做到顺序消费RocketMQ

• 发送方：在发送时将ack不能设置 0 ，关闭重试，使用同步发送，等到发送成功再发送下一条。确保消息是顺序发送的。
• 接收方：消息是发送到一个分区中，只能有一个消费组的消费者来接收消息。因此，kafka的顺序消费会牺牲掉性能。

4.解决消息积压问题

• 方案一：在一个消费者中启动多个线程，让多个线程同时消费。——提升一个消费者的消费能力（增加分区增加消费者）。
• 方案二：如果方案一还不够的话，这个时候可以启动多个消费者，多个消费者部署在不同的服务器上。其实多个消费者部署在同一服务器上也可以提高消费能力——充分利用服务器的cpu资源。
• 方案三：让一个消费者去把收到的消息往另外一个topic上发，另一个topic设置多个分区和多个消费者 ，进行具体的业务消费。

5.延迟队列

延迟队列的应用场景：在订单创建成功后如果超过 30 分钟没有付款，则需要取消订单，此时可用延时队列来实现

• 创建多个topic，每个topic表示延时的间隔

  • topic_5s: 延时5s执行的队列
  • topic_1m: 延时 1 分钟执行的队列
  • topic_30m: 延时 30 分钟执行的队列

• 消息发送者发送消息到相应的topic，并带上消息的发送时间

• 消费者订阅相应的topic，消费时轮询消费整个topic中的消息

  • 如果消息的发送时间，和消费的当前时间超过预设的值，比如 30 分钟
  • 如果消息的发送时间，和消费的当前时间没有超过预设的值，则不消费当前的offset及之后的offset的所有消息都消费
  • 下次继续消费该offset处的消息，判断时间是否已满足预设值

十一、Kafka-eagle监控平台

安装Kafka-eagle

官网下载压缩包

EFAK

• 安装jdk
• 解压缩后修改配置文件 system-config.properties

# 配置zk  去掉cluster2
efak.zk.cluster.alias=cluster1
cluster1.zk.list=172.16.253.35:2181
# cluster2.zk.list=xdn10:2181,xdn11:2181,xdn12:2181# 配置mysql
kafka.eagle.driver=com.mysql.cj.jdbc.Driver
kafka.eagle.url=jdbc:mysql://172.16.253.22:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password= 123456Copy to clipboardErrorCopied

• 修改/etc/profile

  export  JAVA_HOME=/usr/local/jdk/jdk1.8.0_191
  CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar
  export KE_HOME=/home/aisys/efak-web-2.0.9
  export PATH=$PATH:$KE_HOME/bin:$JAVA_HOME/binCopy to clipboardErrorCopied

• 刷新配置

  source /etc/profileCopy to clipboardErrorCopied

• 进入到bin目录，为ke.sh增加可执行的权限

chmod +x ke.shCopy to clipboardErrorCopied

• 启动kafka-eagle

  ./ke.sh start

本文链接：https://my.lmcjl.com/post/1302.html