分布式专题｜想进入大厂，你得会点kafka

AI码师

发布于 2020-12-11 14:28:45

5950

发布于 2020-12-11 14:28:45

文章被收录于专栏：分布式｜微服务｜性能优化｜并发编程｜源码分析分布式｜微服务｜性能优化｜并发编程｜源码分析

点击上方蓝字关注我们文末有惊喜

介绍

什么是kafka？

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目

kafka能用在哪里？

日志收集：一个公司可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
消息系统：解耦和生产者和消费者、缓存消息等。
用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

kafka基本组件

Broker 消息中间件处理节点，一个Kafka节点就是一个broker，一个或者多个Broker可以组成一个Kafka集群
Topic Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic
Producer 消息生产者，向Broker发送消息的客户端
Consumer 消息消费者，从Broker读取消息的客户端
ConsumerGroup 每个Consumer属于一个特定的Consumer Group，一条消息可以被多个不同的Consumer Group消费，但是一个Consumer Group中只能有一个Consumer能够消费该消息
Partition 物理上的概念，一个topic可以分为多个partition，每个partition内部消息是有序的，每个partition又能支持分配多个副本，在多个副本所在的broker中，会选举出一个leader，leader分区负责处理读写请求，并且把数据同步给其他的follower中保存的副本。

kafka 难点理解

消息消费之后不会删除消息被消费者消费之后，还会保存在分区里面，什么时候会被删除呢？是通过配置参数 「log.retention.hours」决定的，如果这里设置成10，那么10小时之后，这个消息就会被删除
topic、partition和Broker之间的关系一个topic代表一个业务数据集，例如如果需要处理订单数据，可以为订单消息创建一个topic，所有的订单消息都会发到这个topic中；如果订单消息越来越多，那么就会造成这个topic变得越来越大，有可能会达到TB，肯定不能放在单机上面保存，我们可以对这个topic进行分区(partition)，这些分区会分散在不同的机器上面，划分多个分区，也是为了提高消息的并发消费，因为前面说过，一个分区只能被每个消费组中的一个消费者进行消费，如果拆分成多个分区，就可以同时被多个消费者进行消费； broker最容易理解了:运行kafka进程的机器就是一个broker；
kafka如何支持传统消息的两种模式：队列和订阅这两种模式都是基于kafka的消费机制决定的：生产者发送的消息会发到所有订阅了该topic的消费组(consumer grop)中，但是每个消费组中只有一个消费者能够消费到这条消息。

kafka如何保证消息顺序消费 kafka通过保证一个分区的消息只能被消费组中的一个消费者进行消费，所以生产者发送消息必须将消息发送到同一个分区中，才能保证消息顺序消费；

如何在docker上安装kafka

安装kafka的前提是你要安装zookeeper

安装zookeeper

# 创建文件夹
mkdir -p ~/docker/zookeeper/conf
mkdir -p ~/docker/zookeeper/data
mkdir -p ~/docker/zookeeper/datalog
docker run -d --name zookeeper \
--restart always \                        # docker服务启动时, 默认启动zookeeper容器
-p 2181:2181 -p 2888:2888 -p 3888:3888 \
-v ~/docker/zookeeper/conf:/conf \
-v ~/docker/zookeeper/data:/data \
-v ~/docker/zookeeper/datalog:/datalog \
zookeeper:3.4.14

安装kafka

 docker run  -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -v ~/docker/kafka/logs:/kafka -e KAFKA_ZOOKEEPER_CONNECT=zookeeper --link zookeeper:zookeeper -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${IP}:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

使用kafka自带的控制台生产者和消费者进行测试


# 开启生产者
docker exec -it kafka bash
# 创建主题
kafka-topics.sh --create -zookeeper zookeeper --topic lezai --partitions 3 -replication-factor 1
# 生产者连接kafka
kafka-console-producer.sh --topic lezai -bootstrap-server 127.0.0.1:9092

# 开启消费者
docker exec -it kafka bash
# 消费者连接kafka
kafka-console-consumer.sh --topic lezai -bootstrap-server 127.0.0.1:9092 --from-beginning

# 现在在生产者窗口输入内容，看看消费者窗口是否能收到

kafka命令行常规操作

1.查看topic的详细信息 
./kafka-topics.sh -bootstrap-server 127.0.0.1:2181 -describe -topic testKJ1 
  
2、为topic增加副本 
./kafka-reassign-partitions.sh -zookeeper 127.0.0.1:2181 -reassignment-json-file json/partitions-to-move.json -execute 
  
3、创建topic 
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testKJ1 
  
4、为topic增加partition 
./bin/kafka-topics.sh –zookeeper 127.0.0.1:2181 –alter –partitions 20 –topic testKJ1 
  
5、kafka生产者客户端命令 
./kafka-console-producer.sh --broker-list localhost:9092 --topic testKJ1 
  
6、kafka消费者客户端命令 
./kafka-console-consumer.sh -zookeeper localhost:2181 --from-beginning --topic testKJ1 
  
7、kafka服务启动 
./kafka-server-start.sh -daemon ../config/server.properties  
  
8、下线broker 
./kafka-run-class.sh kafka.admin.ShutdownBroker --zookeeper 127.0.0.1:2181 --broker #brokerId# --num.retries 3 --retry.interval.ms 60 
shutdown broker 
  
9、删除topic 
./kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic testKJ1 --zookeeper 127.0.0.1:2181 
./kafka-topics.sh --zookeeper localhost:2181 --delete --topic testKJ1 
  
10、查看consumer组内消费的offset 
./kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper localhost:2181 --group test --topic testKJ1

springboot 集成kafka

依赖导入

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

配置文件

server:
  port: 8080
spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092
    producer: # 生产者
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384
      buffer-memory: 33554432
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: mygroup
      enable-auto-commit: true

添加消息生产者

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;
    @Test
    public  void send(){
        kafkaTemplate.send("lezai",0,"key","kafkasss 发送数据");
        try {
            Thread.sleep(1000000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

添加两个消费者，存在不同消费组

    @KafkaListener(topics = "lezai",groupId = "testGroup")
    public void listen(ConsumerRecord<String, String> record) {
        String value = record.value();
        System.out.println("testGroup"+value);
        System.out.println(record);
    }


    @KafkaListener(topics = "lezai",groupId = "testGroup2")
    public void listen2(ConsumerRecord<String, String> record) {
        String value = record.value();
        System.out.println("testGroup2"+value);
        System.out.println(record);
    }
    
    // 可以切换为相同的groupId，来验证消息是否会被同一个消费组中的消费者消费