目录
介绍
Kafka核心相关名称
kafka集群安装
kafka使用
kafka文件存储
Springboot整合kafka
首页 Java java教程 Java分布式之Kafka消息队列实例分析

Java分布式之Kafka消息队列实例分析

Apr 19, 2023 pm 04:10 PM
java kafka

介绍

Apache Kafka 是分布式发布-订阅消息系统,在 kafka官网上对 kafka 的定义:一个分布式发布-订阅消息传递系统。 它最初由LinkedIn公司开发,Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。Kafka是一种快速、可扩展的、设计内在就是分布式的,分区的和可复制的提交日志服务。

注意:Kafka并没有遵循JMS规范(),它只提供了发布和订阅通讯方式。

Kafka核心相关名称

  1. Broker:Kafka节点,一个Kafka节点就是一个broker,多个broker可以组成一个Kafka集群

  2. Topic:一类消息,消息存放的目录即主题,例如page view日志、click日志等都可以以topic的形式存在,Kafka集群能够同时负责多个topic的分发

  3. massage: Kafka中最基本的传递对象。

  4. Partition:topic物理上的分组,一个topic可以分为多个partition,每个partition是一个有序的队列。Kafka里面实现分区,一个broker就是表示一个区域。

  5. Segment:partition物理上由多个segment组成,每个Segment存着message信息

  6. Producer : 生产者,生产message发送到topic

  7. Consumer : 消费者,订阅topic并消费message, consumer作为一个线程来消费

  8. Consumer Group:消费者组,一个Consumer Group包含多个consumer

  9. Offset:偏移量,理解为消息 partition 中消息的索引位置

主题和队列的区别:

队列是一个数据结构,遵循先进先出原则

kafka集群安装

  • 每台服务器上安装jdk1.8环境

  • 安装Zookeeper集群环境

  • 安装kafka集群环境

  • 运行环境测试

Java分布式之Kafka消息队列实例分析

安装jdk环境和zookeeper这里不详述了。

kafka为什么依赖于zookeeper:kafka会将mq信息存放到zookeeper上,为了使整个集群能够方便扩展,采用zookeeper的事件通知相互感知。

kafka集群安装步骤:

1、下载kafka的压缩包

2、解压安装包

tar -zxvf kafka_2.11-1.0.0.tgz

3、修改kafka的配置文件 config/server.properties

配置文件修改内容:

  • zookeeper连接地址:zookeeper.connect=192.168.1.19:2181

  • 监听的ip,修改为本机的iplisteners=PLAINTEXT://192.168.1.19:9092

  • kafka的brokerid,每台broker的id都不一样broker.id=0

4、依次启动kafka

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

kafka使用

kafka文件存储

topic是逻辑上的概念,而partition是物理上的概念,每个partition对应于一个log文件,该log文件中存储的就是Producer生成的数据。Producer生成的数据会被不断追加到该log文件末端,为防止log文件过大导致数据定位效率低下,Kafka采取了分片和索引机制,将每个partition分为多个segment,每个segment包括:“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下,该文件夹的命名规则为:topic名称+分区序号。

例如:执行命令新建一个主题,分三个区存放放在三个broker中:

./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic kaico

Java分布式之Kafka消息队列实例分析

Java分布式之Kafka消息队列实例分析

  • 一个partition分为多个segment

  • .log 日志文件

  • .index 偏移量索引文件

  • .timeindex 时间戳索引文件

  • 其他文件(partition.metadata,leader-epoch-checkpoint)

Springboot整合kafka

maven依赖

 <dependencies>
        <!-- springBoot集成kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- SpringBoot整合Web组件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>
登录后复制

yml配置

# kafka
spring:
  kafka:
    # kafka服务器地址(可以多个)
#    bootstrap-servers: 192.168.212.164:9092,192.168.212.167:9092,192.168.212.168:9092
    bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094
    consumer:
      # 指定一个默认的组名
      group-id: kafkaGroup1
      # earliest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
      # latest:当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 缓存容量
      buffer-memory: 524288
      # 服务器地址
      bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094
登录后复制

生产者

@RestController
public class KafkaController {
	/**
	 * 注入kafkaTemplate
	 */
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	/**
	 * 发送消息的方法
	 *
	 * @param key
	 *            推送数据的key
	 * @param data
	 *            推送数据的data
	 */
	private void send(String key, String data) {
		// topic 名称 key   data 消息数据
		kafkaTemplate.send("kaico", key, data);
	}
	// test 主题 1 my_test 3
	@RequestMapping("/kafka")
	public String testKafka() {
		int iMax = 6;
		for (int i = 1; i < iMax; i++) {
			send("key" + i, "data" + i);
		}
		return "success";
	}
}
登录后复制

消费者

@Component
public class TopicKaicoConsumer {
    /**
     * 消费者使用日志打印消息
     */
    @KafkaListener(topics = "kaico") //监听的主题
    public void receive(ConsumerRecord<?, ?> consumer) {
        System.out.println("topic名称:" + consumer.topic() + ",key:" +
                consumer.key() + "," +
                "分区位置:" + consumer.partition()
                + ", 下标" + consumer.offset());
        //输出key对应的value的值
        System.out.println(consumer.value());
    }
}
登录后复制

以上是Java分布式之Kafka消息队列实例分析的详细内容。更多信息请关注PHP中文网其他相关文章!

本站声明
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系admin@php.cn

热AI工具

Undresser.AI Undress

Undresser.AI Undress

人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover

AI Clothes Remover

用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool

Undress AI Tool

免费脱衣服图片

Clothoff.io

Clothoff.io

AI脱衣机

AI Hentai Generator

AI Hentai Generator

免费生成ai无尽的。

热门文章

R.E.P.O.能量晶体解释及其做什么(黄色晶体)
2 周前 By 尊渡假赌尊渡假赌尊渡假赌
仓库:如何复兴队友
1 个月前 By 尊渡假赌尊渡假赌尊渡假赌
Hello Kitty Island冒险:如何获得巨型种子
4 周前 By 尊渡假赌尊渡假赌尊渡假赌

热工具

记事本++7.3.1

记事本++7.3.1

好用且免费的代码编辑器

SublimeText3汉化版

SublimeText3汉化版

中文版,非常好用

禅工作室 13.0.1

禅工作室 13.0.1

功能强大的PHP集成开发环境

Dreamweaver CS6

Dreamweaver CS6

视觉化网页开发工具

SublimeText3 Mac版

SublimeText3 Mac版

神级代码编辑软件(SublimeText3)

Java 中的平方根 Java 中的平方根 Aug 30, 2024 pm 04:26 PM

Java 中的平方根指南。下面我们分别通过例子和代码实现来讨论平方根在Java中的工作原理。

Java 中的完美数 Java 中的完美数 Aug 30, 2024 pm 04:28 PM

Java 完美数指南。这里我们讨论定义,如何在 Java 中检查完美数?,示例和代码实现。

Java 中的随机数生成器 Java 中的随机数生成器 Aug 30, 2024 pm 04:27 PM

Java 随机数生成器指南。在这里,我们通过示例讨论 Java 中的函数,并通过示例讨论两个不同的生成器。

Java中的Weka Java中的Weka Aug 30, 2024 pm 04:28 PM

Java 版 Weka 指南。这里我们通过示例讨论简介、如何使用weka java、平台类型和优点。

Java 中的阿姆斯特朗数 Java 中的阿姆斯特朗数 Aug 30, 2024 pm 04:26 PM

Java 中的阿姆斯特朗数指南。这里我们讨论一下java中阿姆斯特朗数的介绍以及一些代码。

Java 中的史密斯数 Java 中的史密斯数 Aug 30, 2024 pm 04:28 PM

Java 史密斯数指南。这里我们讨论定义,如何在Java中检查史密斯号?带有代码实现的示例。

Java Spring 面试题 Java Spring 面试题 Aug 30, 2024 pm 04:29 PM

在本文中,我们保留了最常被问到的 Java Spring 面试问题及其详细答案。这样你就可以顺利通过面试。

突破或从Java 8流返回? 突破或从Java 8流返回? Feb 07, 2025 pm 12:09 PM

Java 8引入了Stream API,提供了一种强大且表达力丰富的处理数据集合的方式。然而,使用Stream时,一个常见问题是:如何从forEach操作中中断或返回? 传统循环允许提前中断或返回,但Stream的forEach方法并不直接支持这种方式。本文将解释原因,并探讨在Stream处理系统中实现提前终止的替代方法。 延伸阅读: Java Stream API改进 理解Stream forEach forEach方法是一个终端操作,它对Stream中的每个元素执行一个操作。它的设计意图是处

See all articles