Java中Kafka的基本使用

简介:

由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。

这里测试的环境是基于之前写的springcloud项目,博客地址:https://blog.csdn.net/qq_43222167/article/details/105577097

基本流程呢就是:
Producer(生产者),向Kafka发送消息,发送消息的时候呢,会 根据你设置的Topic进行分类
Consumer(消费者),通过配置文件建立起与Kafka的连接,不断的拉取并对这些消息进行处理
Topic(主题,也可以理解成分类):对消息指定主题进行分类,消费者可以只关注自己需要的Topic中的消息即可

每一个主题下可以拥有不同的目录（ partition ），目录的命令是根据主题（Topic）+ 有序序号组成的。

不同于activeMQ消息队列， Kafka 的设计是把所有的 Message 都要写入速度低、容量大的硬盘，以此来换取更强的存储能力 ， 同时因为 Kafka 在磁盘上只做 Sequence I/O （顺序写） ，所以并没有对硬盘带来过大的损失

并且在读取数据的时候改变读取方式，通过0拷贝的方式读取，依次减少读的时间

上篇博客已经讲到怎么在Windows中安装Kafka:https://blog.csdn.net/qq_43222167/article/details/106096668

我们先启动Zookeeper的服务,再启动Kafka的服务

启动服务后导入jar包

 <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

我这里jar包的位置是在order里面,也就是消费者和生产者项目的父级项目

导入jar包后编写在生产者里面编写一个新的类,类中的内容如下:

/* 发送kafka */
    @Autowired
    private KafkaTemplate<String, String> KafkaTemplate;

    @GetMapping("sendK")
    @ResponseBody
    public String sendMsgToKafka() {
    	//send(topic:主题,data:数据)
        KafkaTemplate.send("message"
                , "hello,Kafka!");

        return "发送消息到Kafka完毕";
    }

对配置文件的配置(yml文件,只贴出kafka的内容)

spring:
  kafka:
    producer:
      bootstrap-servers: 127.0.0.1:9092	#Kafka的位置

接下来是对消费者项目的编写:
新的类:

package com.jbit.util;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class MyKafkaListener {

    @KafkaListener(topics = {"message"})
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println("数据接收完毕："+message);
        }
    }
}

配置文件(只贴出kafka的内容):

spring:
  kafka:
    producer:
      bootstrap-servers: 127.0.0.1:9092
    consumer:
      group-id: consumer-group  #命令kafka的组id,可以随便起

测试:
访问的ip为:http://localhost:项目提供者的端口/sendK
我这里为:http://localhost:8071/sendK

结果可以看到在消费者控制台里已经拉取到提供者里面发来的消息了