1 Apache Kafka 基本操作

1.1 准备

本机ip：192.168.135.145

关闭防火墙：systemctl stop firewalld

1.2 启动Zookeeper

  1. zookeeper-server-start.sh config/zookeeper.properties

1.3 启动Kafka Broker

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

1.4 创建主题Topic

  1. kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 -- partitions 1 --topic test

输出

  1. Created topic test.

1.5 查看主题Topic

  1. kafka-topics.sh  --list --zookeeper localhost:2181

输出

test

1.6 修改主题Topic

1.6.1 修改分区数Partition

  1. kafka-topics.sh --zookeeper localhost:2181 --alter --topic topicName --partitions 8

1.6.2 增加配置

  1. kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --config flush.messages=1

1.6.3 删除配置

  1. kafka-topics.sh --zookeeper localhost:2181 --alter --topic test --delete-config flush.messages

1.6.4 删除主题Topic

  1. kafka-topics.sh --zookeeper localhost:2181 --delete --topic topicName

1.7 生产者

  1. kafka-console-producer.sh --broker-list 192.168.135.145:9092 --topic test

1.8 消费者

  1. kafka-console-consumer.sh --bootstrap-server 192.168.135.145:9092 --topic test --from-beginning

例子：

Kafka 基本操作 - 图1
Kafka 基本操作 - 图2
注意：ip不能写错，否则无法建立连接。

2 集群模式

克隆两台虚拟机
复制两个配置文件

  1. cd /opt/kafka_2.11-2.4.0/
  2. cp config/server.properties config/server-1.properties
  3. cp config/server.properties config/server-2.properties
  4. vim config/server-1.properties
  5.     broker.id=0
  6.     listeners=PLAINTEXT://192.168.135.145:9092
  7.     log.dirs=/tmp/kafka-logs-0
  8.     zookeeper.connect=192.168.135.145:2181,192.168.135.146:2181,192.168.135.147:2181
  9.     host.name=192.168.135.145
  10. vim config/server-1.properties
  11.     broker.id=1
  12.     listeners=PLAINTEXT://192.168.135.146:9092
  13.     log.dirs=/tmp/kafka-logs-0
  14.     zookeeper.connect=192.168.135.145:2181,192.168.135.146:2181,192.168.135.147:2181
  15.     host.name=192.168.135.146
  16. vim config/server-2.properties
  17.     broker.id=1
  18.     listeners=PLAINTEXT://192.168.135.147:9092
  19.     log.dirs=/tmp/kafka-logs-0
  20.     zookeeper.connect=192.168.135.145:2181,192.168.135.146:2181,192.168.135.147:2181
  21.     host.name=192.168.135.147

克隆三个窗口

  1. Broker1
  2. kafka-server-start.sh config/server.properties
  3. Broker2
  4. kafka-server-start.sh config/server-1.properties
  5. Broker3
  6. kafka-server-start.sh config/server-2.properties

2.1 创建主题

创建三个分区，两个副本

  1. kafka-topics.sh --create --zookeeper 192.168.135.145:2181 --replication-factor 2 -- partitions 3 --topic test

2.2 查看主题

  1. kafka-topics.sh  --list --zookeeper 192.168.135.145:2181,192.168.135.146:2181,192.168.135.147:2181

2.3 生产者

  1. kafka-console-producer.sh --broker-list 192.168.135.145:9092,192.168.135.146:9092,192.168.135.147:9092 --topic test

2.4 消费者

  1. kafka-console-consumer.sh --from-beginning --topic test  --zookeeper 192.168.135.145:2181,192.168.135.146:2181,192.168.135.147:2181

2.5 查看各节点运行情况

  1. kafka-topics.sh --describe --zookeeper 192.168.135.145:2181 --topic test

输出

  1.  Topic: test   PartitionCount: 3    ReplicationFactor: 2  Configs:<br />     Topic: test   Partition: 0   Leader: 0    Replicas: 0,2,1   Isr: 0,2,1

2.6 查看主题消费进度

  1. bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv
  2. Group           Topic              Pid Offset   logSize    Lag    Owner 
  3. pv              page_visits        0   21       21         0      none 
  4. pv              page_visits        1   19       19         0      none 
  5. pv              page_visits        2   20       20         0      none

topic：创建时topic名称

pid：分区编号

offset：表示该parition已经消费了多少条message

logSize：表示该partition已经写了多少条message

Lag：表示有多少条message没有被消费。

Owner：表示消费者

3 Java API

3.1 pom

  1.    <dependencies>
  2.         <!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
  3.         <dependency>
  4.             <groupId>org.apache.kafka</groupId>
  5.             <artifactId>kafka-clients</artifactId>
  6.             <version>2.4.0</version>
  7.         </dependency>
  9.         <dependency>
  10.             <groupId>org.apache.kafka</groupId>
  11.             <artifactId>kafka-streams</artifactId>
  12.             <version>2.4.0</version>
  13.         </dependency>
  15.     </dependencies>
  17.     <build>
  18.         <plugins>
  19.             <!-- java编译插件 -->
  20.             <plugin>
  21.                 <groupId>org.apache.maven.plugins</groupId>
  22.                 <artifactId>maven-compiler-plugin</artifactId>
  23.                 <version>3.2</version>
  24.                 <configuration>
  25.                     <source>1.8</source>
  26.                     <target>1.8</target>
  27.                     <encoding>UTF-8</encoding>
  28.                 </configuration>
  29.             </plugin>
  30.         </plugins>
  31.     </build>

3.2 生产者

  1. public class SimpleProducer {
  3.     public static void main(String[] args) {
  5.         //Assign topicName to string variable
  6.         String topicName = "my-topic";
  8.         // create instance for properties to access producer configs
  9.         Properties props = new Properties();
  11.         //Assign localhost id
  12.         props.put("bootstrap.servers", "192.168.135.145:9092");
  14.         //Set acknowledgements for producer requests.
  15.         props.put("acks", "all");
  17.         //If the request fails, the producer can automatically retry,
  18.         props.put("retries", 0);
  20.         //Specify buffer size in config
  21.         props.put("batch.size", 16384);
  23.         //Reduce the no of requests less than 0
  24.         props.put("linger.ms", 1);
  26.         //The buffer.memory controls the total amount of memory available to the producer for buffering.
  27.         props.put("buffer.memory", 33554432);
  29.         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  31.         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
  33.         Producer<String, String> producer = new KafkaProducer<> (props);
  35.         for (int i = 0; i < 10; i++) {
  36.             // 发送消息
  37.             producer.send(new ProducerRecord<>(topicName,"hello world->"+ Integer.toString(i)));
  38.         }
  40.         System.out.println("Message sent successfully");
  41.         producer.close();
  42.     }
  43. }

3.3 消费者

  1. public class SimpleConsumer {
  3.     public static void main(String[] args) {
  4.         String topic = "my-topic";
  5.         String group = "test";
  6.         Properties props = new Properties();
  7.         props.put("bootstrap.servers", "192.168.135.145:9092");
  8.         props.put("group.id", group);
  9.         props.put("enable.auto.commit", "true");
  10.         props.put("auto.commit.interval.ms", "1000");
  11.         props.put("session.timeout.ms", "30000");
  12.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  13.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  14.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
  16.         consumer.subscribe(Arrays.asList(topic));
  17.         System.out.println("Subscribed to topic " + topic);
  19.         while (true) {
  20.             ConsumerRecords<String, String> records = consumer.poll(100);
  21.             for (ConsumerRecord<String, String> record : records) {
  22.                 // 接收消息
  23.                 System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
  24.             }
  25.         }
  26.     }
  27. }

注意：测试之前一定要关闭防火墙，否则建立不了连接。

输出

offset = 16, key = null, value = hello world->0
offset = 17, key = null, value = hello world->1
offset = 18, key = null, value = hello world->2
offset = 19, key = null, value = hello world->3
offset = 20, key = null, value = hello world->4
offset = 21, key = null, value = hello world->5
offset = 22, key = null, value = hello world->6
offset = 23, key = null, value = hello world->7
offset = 24, key = null, value = hello world->8
offset = 25, key = null, value = hello world->9