一、kafka主要特点如下
-
同时为发布和订阅提供高吞吐量。 Kafka 的设计目标是以时间复杂度为 0(1 ）的方式提 供消息持久化能力的 ， 即使对 TB 级别以上数据也能保证常数时间的访 问 性能 ，即使 在非常廉价的商用机器上也能做到单机支持每秒 lOOK 条消息的传输。
-
消息持久化。 将消息持久化到磁盘 ， 因此可用于批量消费 ，例如 ETL 以及实时应用程 序。 通过将数据持久化到硬盘以及复制可 以防止数据丢失 。
-
分布式。支持服务器间的消息分区及分布式消费 ，同时保证每个 Partition 内的消息顺 序传输。 其 内 部 的 Producer、 Broker 和 Consumer 都是分布式架构，这更易于 向外扩展。
-
消费消息采用 Pull 模式 。 消息被处理的状态是在 Consumer 端维护的，而不是由服务 器端维护， Broker 无状态 ， Consumer 自己保存 offset。
-
支持 Online 和 Offiine 场景 ，同时支持离线数据处理和实时数据处理。
二、kafka基本参数概念
-
Broker:Kafka 集群中 的一 台或多 台服务器 。
-
Topic:发布到 Kafka 的每条消息都有一个类别，这个类别就被称为 Topic （物理上 ，不同 Topic 的消息分开存储；逻辑上，虽然一个 Topic 的消息被保存在一个或多个 Broker 上 ， 但用户只需指定消息的 Topic 即可生产或消费数据 ， 而不必关心数据存于何处）。
-
Partition ：物理上的 Topic 分区 ， 一个 Topic 可以分为 多个 Partition ， 每个 Partition 都是 一个有序的队列 。 Partition 中的每条消息都会被分配－个有序的 ID ( offset ）。
-
Producer：消 息和数据的生产者，可 以理解为向 Kafka 发消息的客户端。
-
Consumer：消息和数据的消费者 ，可以理解为从 Kafka 取消息的客户端。
-
Consumer Group （消费者组)：每个消费者都属于一个特定 的消费者组（可为每个消 费 者指定组名， 若不指定组名，则属于默认的组） 。这是 Kafka 用来实现一个 Topic 消 息 的广播（发送给所有 的消 费者〉 和单播（发送给任意一个消 费者） 的手段 。 一个 Topic 可以有多个消费者组 。 Topic 的消息会被复制（不是真的 复制， 是概念上的）到所有的 消费者组中，但每个消费者组只会把消息发送给该组中的一个消费者。如果要实现广 播 ， 只要每个消费者都有一个独立的消费者组就可以了；如果要实现单播，只要所有 的消费者都在同一个消费者组中就行。使用消费者组还可以对消费者进行自由分组， 而不需要多次发送消息到不同的 Topic 。
三、kafka可视化客户端工具（Kafka Tool）的基本使用
-
安装：选择对应版本，傻瓜式安装即可。
-
简单使用:
-
连接kafka：
image.png
-
配置以字符串的形式显示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 等。
网友评论