写点什么

Kafka 开飙了!5 分钟,带你体验一把“速度与激情”

作者:java易二三
  • 2023-08-21
    湖南
  • 本文字数:4823 字

    阅读完需:约 16 分钟

前置条件:你的电脑已经安装 Docker


主要内容:


使用 Docker 安装使用命令行测试消息队列的功能 zookeeper 和 kafka 可视化管理工具 Java 程序中简单使用 Kafka 使用 Docker 安装搭建 Kafka 环境单机版


下面使用的单机版的 Kafka 来作为演示,推荐先搭建单机版的 Kafka 来学习。


以下使用 Docker 搭建 Kafka 基本环境来自开源项目:github.com/simplesteph… 。当然,你也可以按照官方提供的来:github.com/wurstmeiste… 。


新建一个名为 zk-single-kafka-single.yml 的文件,文件内容如下:


version: '2.1'


services:zoo1:image: zookeeper:3.4.9hostname: zoo1ports:- "2181:2181"environment:ZOO_MY_ID: 1ZOO_PORT: 2181ZOO_SERVERS: server.1=zoo1:2888:3888volumes:- ./zk-single-kafka-single/zoo1/data:/data- ./zk-single-kafka-single/zoo1/datalog:/datalog


kafka1:image: confluentinc/cp-kafka:5.3.1hostname: kafka1ports:- "9092:9092"environment:KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNALKAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"KAFKA_BROKER_ID: 1KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1


volumes:  - ./zk-single-kafka-single/kafka1/data:/var/lib/kafka/datadepends_on:  - zoo1
复制代码


运行以下命令即可完成环境搭建(会自动下载并运行一个 zookeeper 和 kafka )


docker-compose -f zk-single-kafka-single.yml up 如果需要停止 Kafka 相关容器的话,运行以下命令即可:


docker-compose -f zk-single-kafka-single.yml down 集群版以下使用 Docker 搭建 Kafka 基本环境来自开源项目:github.com/simplesteph… 。


新建一个名为 zk-single-kafka-multiple.yml 的文件,文件内容如下:


version: '2.1'


services:zoo1:image: zookeeper:3.4.9hostname: zoo1ports:- "2181:2181"environment:ZOO_MY_ID: 1ZOO_PORT: 2181ZOO_SERVERS: server.1=zoo1:2888:3888volumes:- ./zk-single-kafka-multiple/zoo1/data:/data- ./zk-single-kafka-multiple/zoo1/datalog:/datalog


kafka1:image: confluentinc/cp-kafka:5.4.0hostname: kafka1ports:- "9092:9092"environment:KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNALKAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"KAFKA_BROKER_ID: 1KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"volumes:- ./zk-single-kafka-multiple/kafka1/data:/var/lib/kafka/datadepends_on:- zoo1


kafka2:image: confluentinc/cp-kafka:5.4.0hostname: kafka2ports:- "9093:9093"environment:KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka2:19093,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9093KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNALKAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"KAFKA_BROKER_ID: 2KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"volumes:- ./zk-single-kafka-multiple/kafka2/data:/var/lib/kafka/datadepends_on:- zoo1


kafka3:image: confluentinc/cp-kafka:5.4.0hostname: kafka3ports:- "9094:9094"environment:KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka3:19094,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9094KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXTKAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNALKAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"KAFKA_BROKER_ID: 3KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"volumes:- ./zk-single-kafka-multiple/kafka3/data:/var/lib/kafka/datadepends_on:- zoo1 运行以下命令即可完成 1 个节点 Zookeeper+3 个节点的 Kafka 的环境搭建。


docker-compose -f zk-single-kafka-multiple.yml up 如果需要停止 Kafka 相关容器的话,运行以下命令即可:


docker-compose -f zk-single-kafka-multiple.yml down 使用命令行测试消息的生产和消费一般情况下我们很少会用到 Kafka 的命令行操作。


1.进入 Kafka container 内部执行 Kafka 官方自带了一些命令


docker exec -ti docker_kafka1_1 bash2.列出所有 Topic


root@kafka1:/# kafka-topics --describe --zookeeper zoo1:21813.创建一个 Topic


root@kafka1:/# kafka-topics --create --topic test --partitions 3 --zookeeper zoo1:2181 --replication-factor 1Created topic test.我们创建了一个名为 test 的 Topic, partition 数为 3, replica 数为 1。


4.消费者订阅主题


root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic testsend hello from console -producer 我们订阅了 名为 test 的 Topic。


5.生产者向 Topic 发送消息


root@kafka1:/# kafka-console-producer --broker-list localhost:9092 --topic test


send hello from console -producer


我们使用 kafka-console-producer 命令向名为 test 的 Topic 发送了一条消息,消息内容为:“send hello from console -producer”


这个时候,你会发现消费者成功接收到了消息:


root@kafka1:/# kafka-console-consumer --bootstrap-server localhost:9092 --topic testsend hello from console -producerIDEA 相关插件推荐 Zoolytic-Zookeeper tool 这是一款 IDEA 提供的 Zookeeper 可视化工具插件,非常好用! 我们可以通过它:


可视化 ZkNodes 节点信息 ZkNodes 节点管理-添加/删除编辑 zkNodes 数据......实际使用效果如下:


使用方法:


打开工具:View->Tool windows->Zoolytic;点击 “+” 号后在弹出框数据:“127.0.0.1:2181” 连接 zookeeper;连接之后点击新创建的连接然后点击“+”号旁边的刷新按钮即可!KafkalyticIDEA 提供的 Kafka 可视化管理插件。这个插件为我们提供了下面这写功能:


多个集群支持主题管理:创建/删除/更改分区使用正则表达式搜索主题发布字符串/字节序列化的消息使用不同的策略消费消息实际使用效果如下:


使用方法:


打开工具:View->Tool windows->kafkalytic;点击 “+” 号后在弹出框数据:“127.0.0.1:9092” 连接;Java 程序中简单使用 Kafka 代码地址:github.com/Snailclimb/…


Step 1:新建一个 Maven 项目


Step2: pom.xml 中添加相关依赖


    <dependency>        <groupId>org.apache.kafka</groupId>        <artifactId>kafka-clients</artifactId>        <version>2.2.0</version>    </dependency>
复制代码


Step 3:初始化消费者和生产者


KafkaConstants 常量类中定义了 Kafka 一些常用配置常量。


public class KafkaConstants {public static final String BROKER_LIST = "localhost:9092";public static final String CLIENT_ID = "client1";public static String GROUP_ID_CONFIG="consumerGroup1";private KafkaConstants() {


}
复制代码


}ProducerCreator 中有一个 createProducer() 方法方法用于返回一个 KafkaProducer 对象


import org.apache.kafka.clients.producer.KafkaProducer;import org.apache.kafka.clients.producer.Producer;import org.apache.kafka.clients.producer.ProducerConfig;import org.apache.kafka.common.serialization.StringSerializer;


import java.util.Properties;


/**


  • @author shuang.kou*/public class ProducerCreator {

  • public static Producer<String, String> createProducer() {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);properties.put(ProducerConfig.CLIENT_ID_CONFIG, KafkaConstants.CLIENT_ID);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());return new KafkaProducer<>(properties);}}ConsumerCreator 中有一个 createConsumer() 方法方法用于返回一个 KafkaConsumer 对象


import org.apache.kafka.clients.consumer.Consumer;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.serialization.StringDeserializer;


import java.util.Properties;


public class ConsumerCreator {


public static Consumer<String, String> createConsumer() {    Properties properties = new Properties();    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstants.BROKER_LIST);    properties.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConstants.GROUP_ID_CONFIG);    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());    return new KafkaConsumer<>(properties);}
复制代码


}Step 4:发送和消费消息


生产者发送消息:


private static final String TOPIC = "test-topic";Producer<String, String> producer = ProducerCreator.createProducer();ProducerRecord<String, String> record =new ProducerRecord<>(TOPIC, "hello, Kafka!");try {//send messageRecordMetadata metadata = producer.send(record).get();System.out.println("Record sent to partition " + metadata.partition()+ " with offset " + metadata.offset());} catch (ExecutionException | InterruptedException e) {System.out.println("Error in sending record");e.printStackTrace();}producer.close();消费者消费消息:


Consumer<String, String> consumer = ConsumerCreator.createConsumer();// 循环消费消息 while (true) {//subscribe topic and consume messageconsumer.subscribe(Collections.singletonList(TOPIC));


ConsumerRecords<String, String> consumerRecords =consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println("Consumer consume message:" + consumerRecord.value());}}Step 5:测试


运行程序控制台打印出:


Record sent to partition 0 with offset 20Consumer consume message:hello, Kafka!开源项目推荐作者的其他开源项目推荐:


JavaGuide:【Java 学习+面试指南】 一份涵盖大部分 Java 程序员所需要掌握的核心知识。springboot-guide : 适合新手入门以及有经验的开发人员查阅的 Spring Boot 教程(业余时间维护中,欢迎一起维护)。programmer-advancement : 我觉得技术人员应该有的一些好习惯!spring-security-jwt-guide :从零入门 !Spring Security With JWT(含权限验证)后端部分代码。

用户头像

java易二三

关注

还未添加个人签名 2021-11-23 加入

还未添加个人简介

评论

发布
暂无评论
Kafka 开飙了!5分钟,带你体验一把“速度与激情”_Java_java易二三_InfoQ写作社区