一、Kafka相关概念
1、关于Kafka的描述
Kafka是由Apache开源，具有分布式、分区的、多副本的、多订阅者，基于Zookeeper协调的分布式处理平台，由Scala和Java语言编写。通常用来搜集用户在应用服务中产生的动作日志数据，并高速的处理。日志类的数据需要高吞吐量的性能要求，对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。
2、关于Kafka的功能特点
- 通过磁盘数据结构提供消息的持久化，消息存储也能够保持长时间稳定性;
- 高吞吐量，即使是非常普通的硬件Kafka也可以支持每秒超高的并发量;
- 支持通过Kafka服务器和消费机集群来分区消息;
- 支持Hadoop并行数据加载;
- API包封装的非常好，简单易用，上手快 ;
- 分布式消息队列。Kafka对消息保存时根据Topic（主题）进行归类，发送消息者称为Producer（生产者），消息接受者称为Consumer（消费者）;
3、Kafka消息功能
如下图所示，Kafka作为一个中间服务，代表一个broker（经纪人）角色，负责接收APP的消费与推送消息给其他相关APP。这里APP可分为Producer，Consumer。
消息的消费模式
点对点模式：点对点模式通常是一个基于拉取或者轮询的消息传递模型，消费者主动拉取数据，消息收到后从队列移除消息，这种模型不是将消息推送到客户端，而是从队列中请求消息。特点是发送到队列的消息被一个且只有一个消费者接收处理，即使有多个消费者监听队列也是如此。
发布订阅模式：订阅模式是一个基于推送的消费传送模型，消息产生后，Kafka会推送给所有订阅相关Topic的订阅者。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。
4、Kafka消息队列的作用
- 应用程序之间解耦，生产者与消费者相互独立，各自异步执行。
- 消息数据持久化存储，直到所有消息都被消费，规避消息数据丢失的风险。
- 流量削峰，使用Kafka消息队列可以帮助server承接访问压力，尽可能避免应用程序崩溃。
- 降低进程间的耦合度，系统部分应用组件发生崩溃时，不会影响到整体系统的运行。
- 保证消息顺序执行，解决特定场景业务需求。
5、Kafka相关术语介绍
- Broker
一台kafka服务器就是一个broker（经纪人）。一个集群由多个broker组成。一个broker可以容纳多个topic（消息主题）。
- Producer
消息生产者，就是向kafka broker发消息的APP客户端。
- Consumer
消息消费者，向kafka broker取消息的APP客户端。
- Topic
每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic，可以理解为一个队列。
- Consumer Group
每个Consumer属于一个特定的Consumer Group，可为每个Consumer指定group name，若不指定group name则属于默认的分组。
- Partition
一个庞大大的topic可以分布到多个broker上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体的顺序。Partition是物理上的概念，方便在集群中扩展，提高并发。
二、liunx系统下搭建Kafka环境
--新建kafka应用目录。并下载到当前目录下
cd /usr/local
mkdir kafka
cd kafka
--下载
wget https://downloads.apache.org/kafka/3.7.0/kafka-3.7.0-src.tgz
--解压
tar -zxvf kafka-3.7.0-src.tgz
--启动服务
cd kafka-3.7.0
./bin/kafka-server-start.sh config/server.properties
--查看服务
ps -aux |grep kafka
--开放kafka地址端口
vim server.properties
--添加下面注释
advertised.listeners=PLAINTEXT://10.98.3.22:9092
三、Springboot2整合Kafka 服务
1、导入基础依赖
<!-- SpringBoot依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- kafka 依赖 -->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.2.4.RELEASE</version>
</dependency>2、项目目录结构
3、生产者与消费者yml文件配置
#消费者配置
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
consumer:
group-id: test-consumer-group
#生产者配置
spring:
kafka:
bootstrap-servers: 127.0.0.1:90924、生成消息
@RestController
@RequestMapping("/kafka")
public class ProducerController {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
@RequestMapping("/send")
public HttpResult sendMsg () {
MsgLog msgLog = new MsgLog(1,"消息生成",
1,"消息日志",new Date()) ;
String msg = JSON.toJSONString(msgLog) ;
// 这里Topic如果不存在，会自动创建
kafkaTemplate.send("cicada-topic", msg);
return HttpResult.create(HttpStatus.SUCCESS,msg);
}
}@Component
public class ConsumerMsg {
private static Logger LOGGER = LoggerFactory.getLogger(ConsumerMsg.class);
//此注解是监听主题为cicada-topic的消息队列
@KafkaListener(topics = "cicada-topic")
public void listenMsg (ConsumerRecord<?,String> record) {
String value = record.value();
LOGGER.info("ConsumerMsg====>>"+value);
}
}
