消息中间件基本架构
消息中间件基本上可以分为三部分：
- Product 生产者，消息的产生方，生产者会将消息发送到消息队列。
- queue 消息队列，接受并存储生产者的消息。
- Consumer 消费者，消费消息队列中的消息。
消息中间件的作用
消息队列的主要作用削峰填谷、异步、解耦。
消费消息的两种模式
消息消费的两种模式，发布订阅模式和点对点模式：
发布订阅模式（一对多）
消费者消费完消息后，消息不会被删除，消息会存储一段时间，这种模式下的消息会被所有消费者消费。
这种模式下消费消息又有两种方式：
-
推模式，这种方式由queue直接将消息推给消费者，由于各消费者处理消息的能力不一样，所以这种方式某些消费者可能消费不过来推过来的消息。
image.png -
拉模式，这种方式是由消费者去拉取queue中的消息，这种方式下消费者可以根据自己的消费能力来拉取数据，但是每个消费者需要维护一个拉取消息的任务，及时队列里面没有消息，这个任务也不能停下来。
image.png
- 点对点模式（一对一）：点对点模式是消费者主动拉取数据，消息确认被消费后，消息队列会删除队列中的消息，一条消息只会被一个消费者消费。
Kafka基本概念
Kafka是一个分布式的消息队列，使用Zookeeper进行集群的管理。
不同消费者组之间是基于发布/订阅的拉模式来实现的；同一消费者组里面的消费者，消费消息使用的是点对点模式，只是这里消费完消息后不会删除消息。
Kafka的特性
- 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。
- 可扩展性：kafka集群支持热扩展。
- 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。
- 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）。
- 高并发：支持数千个客户端同时读写。
Producer
消息生产者，就是向 kafka broker 发消息的客户端;
Broker
一台 kafka 服务器就是一个 broker。一个kafka集群由多个 broker 组成，然后通过Zookeeper来进行集群的管理。
Consumer
消息消费者，向 kafka broker 取消息的客户端;
Consumer Group
消费者组，由多个 consumer 组成。同一个消费者组中消费者只能消费一个分区里面的数据。
Topic
Topic是一个逻辑上的消息队列，同一类型的消息可以放到一个Topic（消息队列）中。主要作用是用来屏蔽底层分区和副本的复杂逻辑。
Partition
Partition（分区），是Kafka下数据存储的基本单元，这个是物理上的概念。一个 topic 可以分为多个 partition，每个 partition 都是一个有序的队列，同一个分区（partition）可以被不同的消费者组同时消费，但是同一个消费者组内只能被一个消费者消费，通过提升分区数量可以提升同一个Topic的吞吐量。
Replica
Replica（副本），就是Partition的一个备份。一个分区（Partition）只能有一个leader，但是可以设置多个副本（follower），同一分区的副本不能在同一台机器上。也就是说如果有2 台 Broker，那么一个分区就最多会有1个副本。leader的主要作用是：完成与生产者、消费者的交互；follower的主要作用是：做数据备份，当 leader 发生故障时，某个 follower 会成为新的 leader，以此来保证kafka的可用性。
分区中的所有副本统称为 AR，而ISR 是指与leader 副本保持同步状态的副本集合，当然leader副本本身也是这个集合中的一员。
LEO标识每个分区中最后一条消息的下一个位置，分区的每个副本都有自己的LEO，ISR中最小的LEO即为HW，俗称高水位，消费者只能拉取到HW之前的消息。
如果我们需要提升kafka消费性能的时候，我们需要同时扩充Partition（分区）数量和同一消费者组中的消费者数量。
我这里使用数据库设计中的分库分表设计来做一个类比：一个订单库，DB分为db_order_1、db_order_2，然后每个数据库中的t_order表有分为t_order_1、t_order_2表。这里db_order_1、db_order_2就相当于kafka的Broker；t_order_1、t_order_2就相当于kafka的Partition（分区）；t_order是一个逻辑表，就相当于kafka中的Topic；如果还有从库，那么从库中的t_order_1、t_order_2就相当于kafka的Replica（副本）。
代码示例
发送消息
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
public <T> void plainNotify(String topic, T obj) {
try {
this.kafkaTemplate.send(topic, JSON.toJSONString(obj));
} catch (Exception e) {
// TODO
}
}
消费消息
@Service
@Slf4j
public class Consumer {
@KafkaListener(id = "ConsumerGroupId", topics = {"TOPIC1", "TOPIC2"})
public void consumer(List<String> msgs, Acknowledgment ack) {
for (String msg : msgs) {
// TODO
}
ack.acknowledge();
}
}
