0


消息中间件 --Kafka

一、 Kafka

1.kafka介绍
    Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。

生产者发送消息,多个消费者只能有一个消费者接收到消息

生产者发送消息,多个消费者都可以接收到消息

  • producer:发布消息的对象称之为主题生产者(Kafka topic producer)
  • topic:Kafka将消息分门别类,每一类的消息称之为一个主题(Topic)
  • consumer:订阅消息并处理发布的消息的对象称之为主题消费者(consumers)
  • broker:已发布的消息保存在一组服务器中,称之为Kafka集群。集群中的每一个服务器都是一个 代理(Broker)。 消费者可以订阅一个或多个主题(topic),并从Broker拉数据,从而消费这些 已发布的消息。
2.kafka 下载(windows下)

ksfka下载链接如下,点击链接进入官网即可下载

kafka官网:http://kafka.apach e.org/

温馨提示:JDK版本至少需要1.8,高版本也可兼容;

第一步:将kafka文件夹解压缩,

注意:一定一定要解压到自己磁盘的根目录下,不然会出现闪退情况!!!

第二步:创建一个 data 空文件夹

后续需要用来存放日志文件,只要创建完成就可以了,kafka启动后会自动生成日志文件;

第三步: 修改 zookeeper.properties 配置文件

我们点击进入config文件夹,找到 zookeeper.properties 配置文件,双击进行修改

然后,我们找到 dataDir ,将它的值修改为我们刚才创建的 data 文件的路径,还要注意一点,在后面还要多加一个 "/zk",因为一会还要配置 server.properties ,所以要用将她们两个区分开

第四步:修改 server.properties 配置文件

和刚才一样,我们双击修改 "server.properties" 配置文件

我们修改 log.dirs 的值为刚才创建的 data 文件夹的路径,在路径末尾再添加上 "/kafka" ,用来和刚才的zk做区分,kafka 文件夹用来存放kafka的日志文件,zk 文件夹用来存放zoopeeper的日志文件;

第五步:创建 "zk.cmd" windows脚本文件

以记事本的方式打开,然后加入下面这句话,

这句话的含义就是启动 Zookeeper ,并且启动文件为 "zookeeper.properties" ;

call bin/windows/zookeeper-server-start.bat config/zookeeper.properties

第六步:创建 "kfk.cmd" windows脚本文件

仍然以记事本的方式打开,然后加入下面这句话,

这句话的含义就是启动 kafka ,并且启动文件为 "server.properties" ;

call bin/windows/kafka-server-start.bat config/server.properties

注意:先启动双击 zk.cmd 启动 zookeeper双击 kafka.cmd 启动 kafka,关闭的时候,需要先关闭 kafka,再关闭 zookeeper ;

3.kafka使用:

(1)创建kafka-demo项目,导入依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

(2)生产者发送消息

/**
 * @author 生产者
 * @version 1.0
 * @since 2024/9/2
 */
public class ProducerQuickStart {
    public static void main(String[] args) {

        // 1.kafka链接配置信息
        Properties prop = new Properties();

        // kafka链接地址
        // 指定 Kafka 集群的地址。在这个例子中,Kafka 服务运行在本地主机(localhost)的 9092 端口上。
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

        // key和value的序列化
        // 配置消息的 key 和 value 序列化类。
        // Kafka 需要将 key 和 value 转换为字节数组进行传输,这里使用的是 StringSerializer,表示 key 和 value 都是字符串类型。
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        // 2.创建kafka生产者对象
        // 生产者对象依赖于前面配置的 Properties 对象来连接 Kafka 集群并发送消息。
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);

        //3.发送消息
        /**
         * 第一个参数 :topic
         * 第二个参数:消息的key
         * 第三个参数:消息的value
         */
        // topic-first:消息要发送到的 Kafka 主题(topic)。
        // key-001:消息的 key,用于分区策略或消息的唯一标识。
        // hello kafka:消息的内容(value)。
        ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("topic-first","key-001","hello kafka");
        // 同步发送消息
        // 使用 Kafka 生产者的 send 方法将消息发送到 Kafka 集群中。
        // 这是一个异步操作,意味着消息会在后台发送,不会阻塞主线程。
        producer.send(kvProducerRecord);

        //4.关闭消息通道  必须要关闭,否则消息发送不成功
        producer.close();

    }
}

(3)消费者接收消息

/**
 * @author 甜甜
 * @version 1.0
 * @since 2024/9/2
 */
public class ConsumerQuickStart1 {

    public static void main(String[] args) {

        // 1.kafka的配置信息
        Properties prop = new Properties();

        // 链接地址
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        //key和value的反序列化器
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 设置消费者组
        // 指定消费者所属的消费者组(group1)。
        //
        prop.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");

        // 2.创建消费者对象
        // Kafka 使用消费者组来协调同一组中多个消费者的消息消费情况,确保每条消息只会被组内的一个消费者处理。
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(prop);

        //3.订阅主题
        // 消费者订阅 topic-first 主题。
        // Collections.singletonList("topic-first") 创建了一个包含 topic-first 的列表,表示消费者只订阅一个主题。
        // 对应的生产者那边的主题
        consumer.subscribe(Collections.singletonList("topic-first"));

        //4.拉取消息
        // 无限循环,用于不断地从 Kafka 中拉取消息。
        while (true) {
            // 读取数据,读取超时时间为100ms ,即每个1000ms拉取一次
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
            // ConsumerRecord 是 Kafka 中表示一条消息的对象,它包含消息的 key、value 以及一些元数据信息。
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord.key());
                System.out.println(consumerRecord.value());
            }
        }

    }

}

使用情景:

  • 生产者发送消息,多个消费者订阅同一个主题(多个消费者都是一个组)只能有一个消费者收到消息 (一对一)
  • 生产者发送消息,多个消费者订阅同一个主题(多个消费者不是一个组)所有消费者都能收到消息 (一对多)
4.springboot集成kafka
4.1导入spring-kafka依赖信息
<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!--  kafkfa  -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
        <dependency>
            <groupId>org.junit.jupiter</groupId>
            <artifactId>junit-jupiter-api</artifactId>
            <version>5.11.0-M2</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
4.2在resources下创建文件application.yml
server:
  port: 9991
spring:
  application:
    name: kafka-demo
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      retries: 10 #重试的次数
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: ${spring.application.name}-test
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
4.3消息生产者
4.4消息消费者

传递消息为对象

目前springboot整合后的kafka,因为序列化器是StringSerializer,这个时候如果需要传递对象可以有 两种方式

方式一:可以自定义序列化器,对象类型众多,这种方式通用性不强,本章节不介绍

方式二:可以把要传递的对象进行转json字符串,接收消息后再转为对象即可,本项目采用这种方式

  • 发送消息
  • 接收消息

二、消息中间件对比

消息中间件对比-选择建议


本文转载自: https://blog.csdn.net/m0_68041576/article/details/141898853
版权归原作者 甜甜不甜- 所有, 如有侵权,请联系我们删除。

“消息中间件 --Kafka”的评论:

还没有评论