kafka集群搭建及Java客户端使用

悠悠 2023-06-25 10:11 12阅读 0赞

# kafka集群搭建及Java客户端使用 #

## kafka简介 ##

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

*  高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒；
 *  可扩展性：kafka集群支持热扩展；
 *  持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；
 *  容错性：允许集群中节点故障（若副本数量为n,则允许n-1个节点故障）；
 *  高并发：支持数千个客户端同时读写。

**术语**

1.  Record(消息)：Kafka处理的主要对象。
2.  Topic(主题)：主题是承载消息的逻辑容器，在实际使用中多用来区分具体的业务。Kafka中的Topics总是多订阅者模式，一个topic可以拥有一个或者多个消费者来订阅它的数据。
3.  Partition(分区)：一个有序不变的消息序列。每个Topic下可以有多个分区。
4.  Oﬀset(消息位移)：表示分区中每条消息的位置信息，是一个单调递增且不变的值。
5.  Replica(副本)：Kafka中同一条消息能够被拷贝到多个地方以提供数据冗余，这些地方就是所谓的副本。副本还分为领导者（leader）副本和追随者(follower)副本，各自有不同的角色划分。副本是在分区层级下的，即每个分区可配置多个副本实现高可用。
6.  Broker(代理)：Kafka以集群的方式运行，集群中的每一台服务器称之为一个代理(broker)Producer(生产者)：消息生产者，向Broker发送消息的客户端。
7.  Consumer(消费者)：消息消费者，从Broker读取消息的客户端。
8.  ConsumerOﬀset(消费者位移)：表征消费者消费进度，每个消费者都有自己的消费者位移
9.  ConsumerGroup(消费者组)：每个Consumer属于一个特定的ConsumerGroup，一条消息可以被多个不同的ConsumerGroup消费，但是一个ConsumerGroup中只能有一个Consumer能够消费该消息。每一个Topic，下面可以有多个分区(Partition)日志文件。Partition是一个有序的message序列，这些message按顺序添加到一个叫做commitlog的文件中。每个partition中的消息都有一个唯一的编号，称之为oﬀset，用来唯一标示某个分区中的message。每个consumer是基于自己在commitlog中的消费进度(oﬀset)来进行工作的。在kafka中，消费oﬀset由consumer自己来维护；一般情况下我们按照顺序逐条消费commitlog中的消息，当然我可以通过指定oﬀset来重复消费某些消息，或者跳过某些消息。

**应用场景**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMjM5OTEz_size_16_color_FFFFFF_t_70]

1.  日志收集：用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer；
2.  消息系统：解耦生产者和消费者、缓存消息等；
3.  用户活动跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后消费者通过订阅这些topic来做实时的监控分析，亦可保存到数据库；
4.  运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；
5.  流式处理：比如sparkstreaming和storm

## kafka使用与集群搭建 ##

**环境准备**  
Kafka是用Scala语言开发的，运行在JVM上，在安装Kafka之前需要先安装JDK  
kafka依赖zookeeper，需要先安装zookeeper,[下载地址][Link 1]

wgethttp://archive.apache.org/dist/zookeeper/zookeeper-3.4.9/zookeeper-
    3.4.9.tar.gz
    tar-zxvfzookeeper-3.4.9.tar.gz
    cdzookeeper-3.4.9
    cpconf/zoo_sample.cfgconf/zoo.cfg
    #启动zookeeper服务
    bin/zkServer.shstart
    #启动客户端
    bin/zkCli.sh

**下载kafka安装包**  
参考[官网安装步骤][Link 2]

wgethttps://archive.apache.org/dist/kafka/1.1.1/kafka_2.11-1.1.1.tgz
    tar-zxvf kafka_2.11-1.1.1.tgz
    cd kafka_2.11-1.1.1.tgz

**启动kafka服务**

bin/kafka-server-start.sh config/server.properties

server.properties是kafka核心配置文件([官网][Link 3])

<table> 
 <thead> 
  <tr> 
   <th>Property</th> 
   <th>Default</th> 
   <th>Description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>broker.id</td> 
   <td>0</td> 
   <td>每个broker都可以用一个唯一的非负整数id进行标识；这个id可以作为broker的“名字”，你可以选择任意你喜欢的数字作为id，只要id是唯一的即可</td> 
  </tr> 
  <tr> 
   <td>log.dirs</td> 
   <td>/tmp/kafka-logs</td> 
   <td>kafka存放数据的路径。这个路径并不是唯一的，可以是多个，路径之间只需要使用逗号分隔即可；每当创建新partition时，都会选择在包含最少partitions的路径下进行</td> 
  </tr> 
  <tr> 
   <td>listeners</td> 
   <td>9092</td> 
   <td>server接受客户端连接的端口</td> 
  </tr> 
  <tr> 
   <td>zookeeper.connect</td> 
   <td>localhost:2181</td> 
   <td>zooKeeper连接字符串的格式为hostname:port，此处hostname和port分别ZooKeeper集群中某个节点的host和port;zookeeper如果是集群，连接方式为hostname1:port1, hostname2:port2,hostname3:port3</td> 
  </tr> 
  <tr> 
   <td>log.retention.hours</td> 
   <td>168</td> 
   <td>每个日志文件删除之前保存的时间。默认数据保存时间对所有topic都一样</td> 
  </tr> 
  <tr> 
   <td>min.insync.replicas</td> 
   <td>1</td> 
   <td>当producer设置acks为-1时，min.insync.replicas指定replicas的最小数目（必须确认每一个repica的写数据都是成功的），如果这个数目没有达到，producer发送消息会产生异常</td> 
  </tr> 
  <tr> 
   <td>delete.topic.enable</td> 
   <td>false</td> 
   <td>是否运行删除主题</td> 
  </tr> 
 </tbody> 
</table>

**创建主题**

#创建分区数是1，副本数是1的主题为test的topic
    bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
    #查看topic列表
    bin/kafka-topics.sh --list --zookeeper localhost:2181
    #查看命令帮助
    bin/kafka-topics.sh

**启动Producer发送消息**

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

**启动Consumer消费消息**

# --group 指定消费组 --from-beginning 从头开始消费 
    bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test -- group consumer1 --from-beginning
    #	查看消费组
    bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

*  单播消息：kafka中，在同一个消费组里,一条消息只能被某一个消费者消费
 *  多播消息：针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组

**集群配置**  
配置3个broker

>	cp config/server.properties config/server-1.properties
    >	cp config/server.properties config/server-2.properties
    config/server-1.properties:
    	broker.id=1
    	listeners=PLAINTEXT://:9093
    	log.dir=/tmp/kafka-logs-1
    config/server-2.properties:
    	broker.id=2
    	listeners=PLAINTEXT://:9094
    	log.dir=/tmp/kafka-logs-2
    #启动broker
    bin/kafka-server-start.sh -daemon config/server-1.properties bin/kafka-server-start.sh -daemon config/server-2.properties
    #创建一个 副本为3，分区为3的topic
    bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 - -partitions 3 --topic my-replicated-topic
    #	查看topic的情况
    bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

*  leader节点负责给定partition的所有读写请求。
 *  replicas表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
 *  isr是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

**Java中kafka‐clients应用**  
Java中使用kafka,引入maven依赖

<dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>1.1.1</version>
    </dependency>

具体Java客户端学习源码地址：[项目源码][Link 4]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMjM5OTEz_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2019122800320612.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMjM5OTEz,size_16,color_FFFFFF,t_70
[Link 1]: http://archive.apache.org/dist/zookeeper/
[Link 2]: http://kafka.apache.org/11/documentation.html#quickstart
[Link 3]: http://kafka.apache.org/11/documentation.html#brokercon%EF%AC%81gs
[Link 4]: https://github.com/glj381413362/kafka-learn