AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Kafka笔记

2023-02-22 100
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云原生网关 MSE Higress，422元/月
服务治理 MSE Sentinel/OpenSergo，Agent数量 不受限
注册配置 MSE Nacos/ZooKeeper，118元/月
简介： Kafka笔记

Kafka单机版安装


注意


1）先做一下快照


2）注意开发文章中涉及的端口


3）注意路径问题，我将软件安装在 /opt/module下，你也可以安装在 /usr/local/下


4）centos  kafka_2.11-0.11.0.2.tgz   zookeeper-3.4.5.tar.gz


5）iZm5ea99qngm2v98asii1aZ 是我的主机名


6）多开几个控制窗口


安装JDK


安装zookeeper


安装Kafka


解压Kafka


tar -zxvf kafka_2.11-0.11.0.2.tgz -C /opt/module/


修改 kafka_2.11-0.11.0.2/config/server.properties 文件


1.png


2.png


修改 kafka_2.11-0.11.0.2/config/producer.properties 文件


3.png


修改 kafka_2.11-0.11.0.2/config/consumer.properties文件


4.png


添加配置Kafka环境变量


注意你Kafka路径的修改


vim /etc/profile


#KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin


source /etc/profile


启动Zookeeper


bin/zkServer.sh start


启动Kafka


非后台启动


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


后台启动 


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


关闭Kafka


bin/kafka-server-stop.sh stop


Kafka Eagle 监控安装


开放9999和8048端口


修改kafka-server-start.sh,添加箭头的那一行


export JMX_PORT="9999"


5.png


下载tar.gz（我们要kafka-eagle-bin-x.x.x.tar.gz中的kafka-eagle-web-1.4.1-bin.tar.gz）


下载kafka-eagle-bin-x.x.x.tar.gz并且解压，获取里面我们真正需要的kafka-eagle-web-1.4.1-bin.tar.gz


下载地址：Download - EFAK


链接：https://pan.baidu.com/s/1UKwokZKiE3ihLH-FOnDTZw

提取码：u4t3


继续解压kafka-eagle-web-1.4.1-bin.tar.gz


配置环境变量


#KE_HOME
export KE_HOME=/opt/module/kafka-eagle-web-1.4.1
export PATH=$PATH:$KE_HOME/bin


赋予权限（在kafka-eagle-web目录下）


chmod 777 ke.sh


6.png


修改kafka-eagle-web-1.4.1/conf/system-config.properties配置文件


7.png


8.png


9.png


下面可以配，也可以不配


######################################
# kafka mysql jdbc driver address
######################################
kafka.eagle.driver=com.mysql.jdbc.Driver
kafka.eagle.url=jdbc:mysql://127.0.0.1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password=root


10.png


启动监控插件


确保zk和kafka都启动

启动监控插件


./bin/ke.sh start


11.png


关闭监控插件


./bin/ke.sh stop


下图表示安装并且开启成功


12.png


遇见的问题


问题：正常启动但是立马关闭

问题原因：内存不足


13.png


解决办法：暂时还没找到，好菜


Kafka 命令行操作


查看当前服务器中的所有 topic


bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --list


创建 topic


选项说明： --topic 定义 topic 名 --replication-factor 定义副本数 --partitions 定义分区数


bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --create --replication-factor 1 --partitions 3 --topic first


14.png


删除 topic


需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除或者直接重启。


bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --delete --topic first


启动控制台生产消息


bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first


15.png


启动控制台消费消息


--from-beginning：会把 first 主题中以往所有的数据都读取出来。根据业务场景选择是否 增加该配置。


bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first


16.png


开启多个消费者位于同一个消费者组中


1）启动zookeeper


bin/zkServer.sh start


2)  启动Kafka


非后台启动


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


后台启动 


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


3) 创建 名称为 first 的topic


bin/kafka-topics.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --create --replication-factor 1 --partitions 3 --topic first


4）像first中生产消息


bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first


5）两个命令行消费消息（重点）（重点）（重点）


bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first --consumer.config config/consumer.properties


6）运行结果：一个消费者组中同一时刻只能有一个消费者消费


17.png


KafkaAPI


Demoo: SSM,SpringBoot or other demo - Gitee.com


pom.xml


依赖的版本要和


<dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>0.11.0.2</version>
    </dependency>


生产者


不带回调函数的生产者


package com.kafka.producer;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
/**
 * 创建生成者：生产消息
 * 1）启动zookeeper
 * ./bin/zkServer.sh start
 * 2)启动Kafka server
 * bin/kafka-server-start.sh config/server.properties
 * 3)启动控制台消息消费者
 * bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first
 * 4)运行Main方法
 *
 */
public class CustomProducer {
  public static void main(String[] args) {
    Properties props = new Properties();
    // Kafka服务端的主机名和端口号 
    //props.put("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092");
    props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092");
    // 等待所有副本节点的应答
    props.put("acks", "all");
    // 消息发送最大尝试次数
    props.put("retries", 0);
    // 一批消息处理大小
    props.put("batch.size", 16384);
    // 请求延时
    props.put("linger.ms", 1);
    // 发送缓存区内存大小
    props.put("buffer.memory", 33554432);
    // key序列化
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // value序列化
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //实例化KafkaProducer
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    for (int i = 0; i < 50; i++) {
      //public ProducerRecord(String topic, K key, V value) 
      //发送消息
      producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), "hello world-" + i));
    }
    //释放资源
    producer.close();
    System.out.println("-------------OVER-------------------");
  }
}


带有回调函数的生产者 


package com.kafka.producer;
import java.util.Properties;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
 * 创建生成者：生产消息
 * 1）启动zookeeper
 * ./bin/zkServer.sh start
 * 2)启动Kafka server
 * bin/kafka-server-start.sh config/server.properties
 * 3)启动控制台消息消费者
 * bin/kafka-console-consumer.sh --zookeeper iZm5ea99qngm2v98asii1aZ:2181 --from-beginning --topic first
 * 4)运行Main方法
 *
 */
public class CustomProducerWithCallBack {
  public static void main(String[] args) throws InterruptedException {
    Properties props = new Properties();
    // Kafka服务端的主机名和端口号
    props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092");
    // 等待所有副本节点的应答
    props.put("acks", "all");
    // 消息发送最大尝试次数
    props.put("retries", 0);
    // 一批消息处理大小
    props.put("batch.size", 16384);
    // 增加服务端请求延时
    props.put("linger.ms", 1);
    // 发送缓存区内存大小
    props.put("buffer.memory", 33554432);
    // key序列化
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    // value序列化
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    //实例化KafkaProducer
    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
    for (int i = 0; i < 50; i++) {
      //发送消息
      kafkaProducer.send(new ProducerRecord<String, String>("test1", "hh" + i), new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception exception) {
          if (metadata != null) {
            System.out.println(metadata.partition() + "---" + metadata.offset());
          }
        }
      });
    }
    kafkaProducer.close();
  }
}


自定义分区


package com.kafka.producer;
import java.util.Map;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
// 自定义分区
//props.put("partitioner.class", "com.kafka.CustomPartitioner");
public class CustomPartitioner implements Partitioner {
  @Override
  public void configure(Map<String, ?> configs) {
  }
  @Override
  public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 控制分区
      //巴里巴拉
    return 1;
  }
  @Override
  public void close() {
  }
}


消费者


package com.kafka.consumer;
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
/**
 * 创建消费者：消费消息
 * 1）启动zookeeper
 * ./bin/zkServer.sh start
 * 2)启动Kafka server
 * bin/kafka-server-start.sh config/server.properties
 * 3)启动控制台生产消息
 * bin/kafka-console-producer.sh --broker-list iZm5ea99qngm2v98asii1aZ:9092 --topic first
 * 4)运行Main方法
 *
 */
public class CustomConsumer {
  public static void main(String[] args) {
    Properties props = new Properties();
    // 定义kakfa 服务的地址
    props.put("bootstrap.servers", "iZm5ea99qngm2v98asii1aZ:9092");
    // 制定consumer group 
    props.put("group.id", "g1");
    // 是否自动确认offset 
    props.put("enable.auto.commit", "true");
    // 自动确认offset的时间间隔 
    props.put("auto.commit.interval.ms", "1000");
    // key的序列化类
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    // value的序列化类 
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    // 定义consumer 
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    // 消费者订阅的topic, 可同时订阅多个 
    consumer.subscribe(Arrays.asList("first"));
        //拉取消息
    while (true) {
      // 读取数据，读取超时时间为100ms 
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records)
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
  }
}


拦截器：实现ProducerInterceptor接口


package com.kafka.interceptor;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
 * 通过拦截器添加时间戳
 *
 */
//      在main 方法中添加过滤器
//    List<String> interceptors = new ArrayList<>();
//    interceptors.add("com.kafka.interceptor.TimeInterceptor"); 
//    props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
//    String topic = "first";
//    Producer<String, String> producer = new KafkaProducer<>(props);
public class TimeInterceptor implements ProducerInterceptor<String, String> {
  @Override
  public void configure(Map<String, ?> configs) {
    // TODO Auto-generated method stub
  }
  @Override
  public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
    // 创建一个新的record，把时间戳写入消息体的最前部
    return new ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(),
        System.currentTimeMillis() + "," + record.value().toString());
  }
  @Override
  public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    // TODO Auto-generated method stub
  }
  @Override
  public void close() {
    // TODO Auto-generated method stub
  }
}


总结


一个分区的数据，只能交给一个消费者

一个消费者，可以消费多个分区的数据，可以跨主题

不会在同一时刻，同一组内，多个消费者同时消费数据

文章标签：
云消息队列 Kafka 版
微服务引擎
Java
消息中间件
监控
Linux
Kafka
关键词：
云消息队列 Kafka 版笔记
CBeann
目录
相关文章
愿天堂没有BUG（公众号同名）
|
7月前
|
消息中间件 监控 Kafka
不愧是Alibaba技术官，Kafka的精髓全写这本“限量笔记”里，服了
分布式，是程序员必备技能之一，在面试过程中属于必备类的，在工作中更是会经常用到。而Kafka是一个分布式的基于发布订阅的消息队列，目前它的魅力是无穷的，对于Kafka的奥秘，还需要我们细细去探寻。
愿天堂没有BUG（公众号同名）
79 0
怒放de生命
|
7月前
|
消息中间件 网络协议 Kafka
docker安装zk和kafka实战笔记
docker安装zk和kafka实战笔记
怒放de生命
135 0
docker安装zk和kafka实战笔记
愿天堂没有BUG（公众号同名）
|
7月前
|
消息中间件 分布式计算 监控
腾讯技术官手撸笔记，全新演绎“Kafka部署实战”，已开源
我们知道，当下流行的MQ非常多，不过很多公司在技术选型上还是选择使用Kafka。与其他主流MQ进行对比，我们会发现Kafka最大的优点就是吞吐量高。实际上Kafka是高吞吐低延迟的高并发、高性能的消息中间件，配置良好的Kafka集群甚至可以做到每秒几十万、上百万的超高并发写入。
愿天堂没有BUG（公众号同名）
101 0
Maynor
|
7月前
|
消息中间件 存储 Kafka
Kafka第二天笔记
Kafka第二天笔记
Maynor
67 0
Maynor
|
7月前
|
消息中间件 存储 Kafka
Kafka第一天笔记
Kafka第一天笔记
Maynor
53 0
朱季谦
|
消息中间件 Kafka Apache
kafka2.x常用命令笔记（一）创建topic，查看topic列表、分区、副本详情，删除topic，测试topic发送与消费
kafka2.x常用命令笔记（一）创建topic，查看topic列表、分区、副本详情，删除topic，测试topic发送与消费
朱季谦
547 0
码农小凡
|
消息中间件 中间件 Kafka
限时开源！阿里内部消息中间件合集：MQ+Kafka+体系图+笔记
近好多小伙伴说在准备金三银四的面试突击了，但是遇到消息中间件不知道该怎么学了，问我有没有成体系的消息中间件的学习方式。 额，有点不知所措，于是乎小编就想着做一次消息中间件的专题，归类整理了一些纯手绘知识体系图、面试以及相关的学习笔记。
码农小凡
239 1
白泽来了
|
消息中间件 数据采集 存储
Kafka架构介绍｜青训营笔记
本文主要讲解了kafka作为分布式消息队列的整体架构，主要从：1.zookeeper；2.broker；3.controller选举；4.coorinator进行介绍
白泽来了
188 0
Kafka架构介绍｜青训营笔记
怒放de生命
|
消息中间件 Kafka Shell
docker安装zk和kafka实战笔记
docker安装zk和kafka实战笔记
怒放de生命
254 0
docker安装zk和kafka实战笔记
愿天堂没有BUG（公众号同名）
|
消息中间件 存储 缓存
不愧是Alibaba技术官，Kafka的精髓全写这本“限量笔记”里，服了
分布式，是程序员必备技能之一，在面试过程中属于必备类的，在工作中更是会经常用到。而Kafka是一个分布式的基于发布订阅的消息队列，目前它的魅力是无穷的，对于Kafka的奥秘，还需要我们细细去探寻。 要谈对Kafka有多熟悉，我相信还是阿里的大佬们最有发言权，所以今天分享的内容，就是Alibaba内部供应的“限量笔记”，关于Kafka的精髓全部写在这里面了，不得不感叹：不愧是Alibaba的技术官啊，真的服了！
愿天堂没有BUG（公众号同名）
125 0

热门文章

最新文章

  • 1
    什么是Apache Kafka？如何将其与Spring Boot集成？
  • 2
    Spring Boot 与 Apache Kafka 集成详解：构建高效消息驱动应用
  • 3
    2024消息队列“四大天王”：Rabbit、Rocket、Kafka、Pulsar巅峰对决
  • 4
    go高并发之路——消息中间件kafka
  • 5
    使用kafka consumer加载数据加载异常并且报source table and destination table are not same错误解决办法
  • 6
    【手把手教你Linux环境下快速搭建Kafka集群】内含脚本分发教程，实现一键部署多个Kafka节点
  • 7
    2024最全Kafka集群方案汇总
  • 8
    解决kafka集群由于默认的__consumer_offsets这个topic的默认的副本数为1而存在的单点故障问题
  • 9
    SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS
  • 10
    springboot配置kafka生产者和消费者详解
  • 1
    Kafka 线程模型痛点攻克： 提升分区写入 2 倍性能
    145
  • 2
    spring kafka的问题集锦
    269
  • 3
    【消息中心】kafka消费失败重试10次的问题
    890
  • 4
    Kafka Exactly Once 语义实现原理：幂等性与事务消息
    192
  • 5
    Kafka 迁移工具 MirrorMaker2 原理起底
    287
  • 6
    一文搞懂 Kafka consumer 与 broker 交互机制与原理
    655
  • 7
    Kafka消息队列原理及应用详解
    283
  • 8
    扒开kafka内部组件，咱瞅一瞅都有啥？
    81
  • 9
    DataWorks操作报错合集之通过dataworks把表数据发送到kafka，表字段全是string类型如何解决
    102
  • 10
    人工智能平台PAI 操作报错合集之pyalink 1.6.1StreamOperator.fromDataframe(out_df, out_schema_str)之后直接连kafka sink会报下面的错误如何解决
    254

    • 相关课程

    更多
  • 消息队列Kafka入门课程
  • 分布式消息系统 Kafka 快速入门
  • RocketMQ 5.0学习资料
  • 消息队列RabbitMQ入门课程
  • 分布式协调系统 Zookeeper 快速入门
  • 消息队列 RocketMQ 消息集成

    • 相关电子书

    更多
  • Java Spring Boot开发实战系列课程【第16讲】：Spring Boot 2.0 实战Apache Kafka百万级高并发消息中间件与原理解析
  • MaxCompute技术公开课第四季 之 如何将Kafka数据同步至MaxCompute
  • 消息队列kafka介绍

    • 相关实验场景

    更多
  • RocketMQ中使用Java客户端发送消息和消费的应用
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)