个人中心
gpt4 book ai didi

spring boot集成kafka开发，接收kafka消息，Java（1）

转载 作者：知者 更新时间：2024-03-13 06:19:35 27 4
gpt4 key购买 nike

在这篇文章（1）基础上：

kafka连接zookeeper配置搭建环境、安装和启动，Windows环境下_Zhang Phil-CSDN博客kafka，分布式集群架构下，高性能的流式事件数据（主要是消息）集成、发布（生产）和订阅（分发、消费）组件（中间件）。kafka依赖zooeeper（数据后端），这里有Windows下安装配置启动zookeeper的 文章（1）：Windows 10环境zookeeper单机伪集群部署和配置_Zhang Phil-CSDN博客Windows 10环境zookeeper单机伪集群部署和配置1，首先到zookeeper项目主页地址下载项目包，https://archive.apache.org/dist

https://blog.csdn.net/zhangphil/article/details/123086790

实现一个简单功能：通过kafka控制台发送消息，然后在spring boot编程实现的java代码程序里面接收消息。

按照文章（1）启动zookeeper和kafka，使用命令：

kafka-topics.bat --list --bootstrap-server 127.0.0.1:9092

查看Windows环境下，列出主机127.0.0.1上kafka的主题topic。本例选择其中一个topic主题，使用kafka命令在控制台发送消息，接收方则用spring框架里面的java程序实现。

下面开始实现spring框架的kafka消息接收java代码。

（1）在application.properties里面配置spring环境中的kafka参数：

server.port=7999
spring.kafka.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.bootstrap-servers=127.0.0.1:9092
spring.kafka.consumer.group-id=zhangphil-group
spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.bootstrap-servers=127.0.0.1:9092
spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.StringOrBytesSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

(2)添加注解@EnableKafka， 打开spring application的kafka配置：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.EnableKafka;

@EnableKafka
@SpringBootApplication
public class SpringKafkaApplication {
    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaApplication.class, args);
    }
}

（3）写一个kafka消息接收器：

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

@Component
public class MyKafka {
    @KafkaListener(topics = "zhangphil_demo", groupId = "${spring.kafka.consumer.group-id}")
    public void msg_listener(ConsumerRecord<?, ?> record) {
        System.out.println(record.topic() + "," + record.offset() + "," + record.value());
        System.out.println(record);
    }
}

以上java程序写完后，即可启动spring application。

然后在kafka控制台，用命令:

kafka-console-producer.bat --topic zhangphil_demo --bootstrap-server localhost:9092

在主题topic为zhangphil_demo里面发送测试消息 hello,world!2022,运行输出如图：

27 4 0
文章推荐： antd为Tree组件标题附加操作按钮功能
文章推荐： Springboot整合Websocket遇到的坑_websocket session不支持序列化，无法存储至redis_Websocket相关问题总结(Session共享，用户多端登录等)
文章推荐： Django（二）
文章推荐： 网络数据包分析软件Wireshark简介
知者
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com