Kafka简介+Kafka Tool使用简介+使用实例

一、kafka主要特点如下

同时为发布和订阅提供高吞吐量。 Kafka 的设计目标是以时间复杂度为 0(1 ）的方式提供消息持久化能力的，即使对 TB 级别以上数据也能保证常数时间的访问性能，即使在非常廉价的商用机器上也能做到单机支持每秒 lOOK 条消息的传输。
消息持久化。将消息持久化到磁盘，因此可用于批量消费，例如 ETL 以及实时应用程序。通过将数据持久化到硬盘以及复制可以防止数据丢失。
分布式。支持服务器间的消息分区及分布式消费，同时保证每个 Partition 内的消息顺序传输。其内部的 Producer、 Broker 和 Consumer 都是分布式架构，这更易于向外扩展。
消费消息采用 Pull 模式 。消息被处理的状态是在 Consumer 端维护的，而不是由服务器端维护， Broker 无状态， Consumer 自己保存 offset。
支持 Online 和 Offiine 场景，同时支持离线数据处理和实时数据处理。

二、kafka基本参数概念
1. Broker:Kafka 集群中的一台或多台服务器。
2. Topic:发布到 Kafka 的每条消息都有一个类别，这个类别就被称为 Topic （物理上，不同 Topic 的消息分开存储；逻辑上，虽然一个 Topic 的消息被保存在一个或多个 Broker 上，但用户只需指定消息的 Topic 即可生产或消费数据，而不必关心数据存于何处）。
3. Partition ：物理上的 Topic 分区，一个 Topic 可以分为多个 Partition ，每个 Partition 都是一个有序的队列。 Partition 中的每条消息都会被分配－个有序的 ID ( offset ）。
4. Producer：消息和数据的生产者，可以理解为向 Kafka 发消息的客户端。
5. Consumer：消息和数据的消费者，可以理解为从 Kafka 取消息的客户端。
6. Consumer Group （消费者组)：每个消费者都属于一个特定的消费者组（可为每个消费者指定组名，若不指定组名，则属于默认的组）。这是 Kafka 用来实现一个 Topic 消息的广播（发送给所有的消费者〉和单播（发送给任意一个消费者）的手段。一个 Topic 可以有多个消费者组。 Topic 的消息会被复制（不是真的复制，是概念上的）到所有的消费者组中，但每个消费者组只会把消息发送给该组中的一个消费者。如果要实现广播，只要每个消费者都有一个独立的消费者组就可以了；如果要实现单播，只要所有的消费者都在同一个消费者组中就行。使用消费者组还可以对消费者进行自由分组，而不需要多次发送消息到不同的 Topic 。
三、kafka可视化客户端工具（Kafka Tool）的基本使用
1. 下载地址：http://www.kafkatool.com/download.html。
2. 安装：选择对应版本，傻瓜式安装即可。
3. 简单使用:
  1. 连接kafka：
    
    image.png
  2. 配置以字符串的形式显示kafka消息体:
  注意：更改完Content Types，要点击Update和Refresh按钮
  
  image-20200708162806709.png

image.png

Kafka Tool工具详细使用访问地址:https://www.cnblogs.com/frankdeng/p/9452982.html

四、Kafka的安装##

详细安装访问：https://www.jianshu.com/p/c74e0ec577b0
macOS 可以用homebrew快速安装,访问地址:https://www.jianshu.com/p/cddd25da8061

五、Kafka—topic的查询和创建##

原文链接：https://www.jianshu.com/p/06884c5bf3f1
查看topic列表：

[root@shtest01 ~]# /usr/lib/kafka/bin/./kafka-topics.sh --list --zookeeper localhost:2181/kafka

创建topic：

[root@shtest01 ~]# /usr/lib/kafka/bin/./kafka-topics.sh --create --topic topictest03 --replication-factor 1 --partitions 1 --zookeeper host01:2181,host02:2181,host03:2181/kafka
Created topic "topictest03".

--create：创建命令；
--topic：后面指定topic名称；
--replication-factor：后面指定副本数；
--partitions：指定分区数，根据broker的数量决定；
--zookeeper：后面指定zookeeper.connect的zk链接
查看某个topic：

[root@shtest01 ~]# /usr/lib/kafka/bin/./kafka-topics.sh --zookeeper localhost:2181/kafka --topic topictest03 --describe
Topic:topictest03       PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: topictest03      Partition: 0    Leader: 1001    Replicas: 1001  Isr: 1001

五、Java 访问 Kafka 实例

Kafka 作为消息系统的一种，当然可以像其他消息中间件一样作为消息数据中转的平台。下面以 Java 语言为例，看一下如何使用 Kafka 来发送和接收消息。
1、引入依赖

<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.11.0.3</version>
        </dependency>

2、消息生产者

public class ProducerSample {

    public static void main(String[] args) {
        Map<String,Object> props = new HashMap<String,Object>();
        props.put("bootstrap.servers","localhost:9092");
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("zk.connect","127.0.0.1:2181");

        String topic= "test-topic";

        Producer<String,String> producer = new KafkaProducer<String, String>(props);
        producer.send(new ProducerRecord<String, String>(topic,"idea-key2","java-message 1"));
        producer.send(new ProducerRecord<String, String>(topic,"idea-key2","java-message 2"));
        producer.send(new ProducerRecord<String, String>(topic,"idea-key2","java-message 3"));
        producer.close();
    }

}

示例中用 KafkaProducer 类来创建一个消息生产者,该类的构造函数入参是一系列属性值。下面看一下这些属性具体都是什么含义。

bootstrap.servers 表示 Kafka 集群。如果集群中有多台物理服务器，则服务器地址之间用逗号分隔，比如” 192.168.1.1 :9092,192.168.1.2:9092” 。 localhost 是笔者电脑的地址，9092 是 Kafka 服务器默认监听的端口号。

key.serializer 和 value.serializer 表示消息的序列化类型。 Kafka 的消息是以键值对的形式发送到 Kafka 服务器的，在消息被发送到服务器之前，消息生产者需要把不同类型的消息序列化为二进制类型，示例中是发送文本消息到服务器，所以使用的是StringSerializer。

key.deserializer 和 value.deserializer 表示消息的反序列化类型。把来自 Kafka 集群的二进制消息反序列化为指定的类型,因为序列化用的是String类型，所以用StringDeserializer 来反序列化。
zk.connect 用于指定 Kafka 连接 ZooKeeper 的 URL ，提供了基于 ZooKeeper 的集群服务器自动感知功能，可以动态从 ZooKeeper 中读取 Kafka 集群配置信息。

有了消息生产者之后，就可以调用 send 方法发送消息了。该方法的入参是 ProducerRecord类型对象， ProducerRecord 类提供了多种构造函数形参，常见的有如下三种：

ProducerRecord(topic,partition,key,value)；

ProducerRecord(topic,key,value)；

ProducerRecord(topic, value);

其中 topic 和 value 是必填的， partition 和 key 是可选的。如果指定了 pa时tion，那么消息会被发送至指定的 partition ；如果没指定 partition 但指定了 Key，那么消息会按照 hash(key）发送至对应的 partition：如果既没指定 partition 也没指定 key，那么消息会按照 round-robin 模式发送（即以轮询的方式依次发送〉到每一个 partition。示例中将向 test-topic 主题发送三条消息。

3、消息消费者

public class ConsumerSample {

    public static void main(String[] args) {
        String topic = "test-topic";
        Properties props = new Properties();
        props.put("bootstrap.servers","localhost:9092");
        props.put("group.id","testGroup1");
        props.put("enable.auto.commit","true");
        props.put("auto.commit.interval.ms","1000");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String,String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(topic));
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(100);
            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());
            }
        }

    }

}

和消息生产者类似，这里用 KafkaConsumer 类来创建一个消息消费者，该类的构造函数入参也是一系列属性值。

bootstrap. servers 和生产者一样，表示 Kafka 集群。

group.id 表示消费者的分组 ID。

enable.auto.commit 表示 Consumer 的 offset 是否自动提交。

auto.commit.interval .ms 用于设置自动提交 offset 到 ZooKeeper 的时间间隔，时间单位是毫秒。

key. deserializer 和 value.deserializer 表示用字符串来反序列化消息数据。

消息消费者使用 subscribe 方法订阅了 Topic 为 test-topic 的消息。 Consumer 调用poll 方法来轮询 Kafka 集群的消息，一直等到 Kafka 集群中没有消息或达到超时时间（示例中设置超时时间为 100 毫秒）为止。如果读取到消息，则打印出消息记录的 pa此ition, offset、key 等。

Kafka简介+Kafka Tool使用简介+使用实例

一、kafka主要特点如下

二、kafka基本参数概念

三、kafka可视化客户端工具（Kafka Tool）的基本使用

四、Kafka的安装##

五、Kafka—topic的查询和创建##

五、Java 访问 Kafka 实例