Java分布式学习之Kafka消息队列

目录

• 介绍
• Kafka核心相关名称
• kafka集群安装
• kafka使用
• kafka文件存储
• Springboot整合kafka

介绍

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

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

kafka中文官网：http://kafka.apachecn.org/quickstart.html

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集群安装

参考官方文档：https://kafka.apachecn.org/quickstart.html

• 每台服务器上安装jdk1.8环境
• 安装Zookeeper集群环境
• 安装kafka集群环境
• 运行环境测试

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

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

kafka集群安装步骤：

1、下载kafka的压缩包，下载地址：https://kafka.apachecn.org/downloads.html

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

• 一个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消息队列的文章就介绍到这了,更多相关Java Kafka内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们！