- Java锁的逻辑(结合对象头和ObjectMonitor)
- 还在用饼状图?来瞧瞧这些炫酷的百分比可视化新图形(附代码实现)⛵
- 自动注册实体类到EntityFrameworkCore上下文,并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
标签:Kafka3.Kafka-eagle3; 。
Kafka是一个开源的分布式事件流平台,常被用于高性能数据管道、流分析、数据集成和关键任务应用,基于Zookeeper协调的处理平台,也是一种消息系统,具有更好的吞吐量、内置分区、复制和容错,这使得它成为大规模消息处理应用程序的一个很好的解决方案; 。
1、下载安装包:kafka_2.13-3.5.0.tgz
2、配置环境变量
open -e ~/.bash_profile
export KAFKA_HOME=/本地路径/kafka3.5
export PATH=$PATH:$KAFKA_HOME/bin
source ~/.bash_profile
3、该目录【kafka3.5/bin】启动zookeeper
zookeeper-server-start.sh ../config/zookeeper.properties
4、该目录【kafka3.5/bin】启动kafka
kafka-server-start.sh ../config/server.properties
1、生产者
kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>id-1-message
>id-2-message
2、消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
id-1-message
id-2-message
3、查看topic列表
kafka-topics.sh --bootstrap-server localhost:9092 --list
test-topic
4、查看消息列表
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --partition 0
id-1-message
id-2-message
配置和部署 。
1、下载安装包:kafka-eagle-bin-3.0.2.tar.gz
2、配置环境变量
open -e ~/.bash_profile
export KE_HOME=/本地路径/efak-web-3.0.2
export PATH=$PATH:$KE_HOME/bin
source ~/.bash_profile
3、修改配置文件:system-config.properties
efak.zk.cluster.alias=cluster1
cluster1.zk.list=localhost:2181
efak.url=jdbc:mysql://127.0.0.1:3306/kafka-eagle
4、本地新建数据库:kafka-eagle,注意用户名和密码是否一致
5、启动命令
efak-web-3.0.2/bin/ke.sh start
命令语法: ./ke.sh {start|stop|restart|status|stats|find|gc|jdk|version|sdate|cluster}
6、本地访问【localhost:8048】 username:admin password:123456
KSQL语句测试 。
select * from `test-topic` where `partition` in (0) order by `date` desc limit 5
select * from `test-topic` where `partition` in (0) and msg like '%5%' order by `date` desc limit 3
这里关于依赖的管理就比较复杂了,首先 spring-kafka 组件选择与boot框架中spring相同的依赖,即 6.0.10 版本,在 spring-kafka 最近的版本中 3.0.8 符合; 。
但是该版本使用的是 kafka-clients 组件的 3.3.2 版本,在Spring文档的kafka模块中,明确说明 spring-boot:3.1 要使用 kafka-clients:3.4 ,所以从 spring-kafka 组件中排除掉,重新依赖 kafka-clients 组件; 。
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${spring-kafka.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
</dependency>
配置kafka连接地址,监听器的消息应答机制,消费者的基础模式; 。
spring:
# kafka配置
kafka:
bootstrap-servers: localhost:9092
listener:
missing-topics-fatal: false
ack-mode: manual_immediate
consumer:
group-id: boot-kafka-group
enable-auto-commit: false
max-poll-records: 10
properties:
max.poll.interval.ms: 3600000
模板类 KafkaTemplate 用于执行高级的操作,封装各种消息发送的方法,在该方法中,通过 topic 和 key 以及消息主体,实现消息的生产; 。
@RestController
public class ProducerWeb {
@Resource
private KafkaTemplate<String, String> kafkaTemplate;
@GetMapping("/send/msg")
public String sendMsg (){
try {
// 构建消息主体
JsonMapper jsonMapper = new JsonMapper();
String msgBody = jsonMapper.writeValueAsString(new MqMsg(7,"boot-kafka-msg"));
// 发送消息
kafkaTemplate.send("boot-kafka-topic","boot-kafka-key",msgBody);
} catch (JsonProcessingException e) {
e.printStackTrace();
}
return "OK" ;
}
}
编写消息监听类,通过 KafkaListener 注解控制监听的具体信息,在实现消息生产和消费的方法测试后,使用可视化工具 kafka-eagle 查看topic和消息列表; 。
@Component
public class ConsumerListener {
private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
@KafkaListener(topics = "boot-kafka-topic")
public void listenUser (ConsumerRecord<?,String> record, Acknowledgment acknowledgment) {
try {
String key = String.valueOf(record.key());
String body = record.value();
log.info("\n=====\ntopic:boot-kafka-topic,key{},body:{}\n=====\n",key,body);
} catch (Exception e){
e.printStackTrace();
} finally {
acknowledgment.acknowledge();
}
}
}
文档仓库:
https://gitee.com/cicadasmile/butte-java-note
源码仓库:
https://gitee.com/cicadasmile/butte-spring-parent
最后此篇关于SpringBoot3集成Kafka的文章就讲到这里了,如果你想了解更多关于SpringBoot3集成Kafka的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
Windows 集成 (NTLM) 身份验证和 Windows 集成 (Kerberos) 之间有什么区别? 如何在IIS6中实现这些 w.r.t. MSDN 最佳答案 Kerberos 和 NTLM
Keycloak是一个用 Java 编写的开源身份验证和身份管理解决方案。它提供了一个nodejs适配器,使用它我能够成功地与express集成。这是有效的路由文件: 'use strict'
这是我关于 Bamboo 的第二个问题 ( My First One )。阅读建议信息后我的理解是,我需要一个构建工具,例如 nAnt 或 MSbuild 来编写一个获取源代码并构建它的脚本(我正在开
可用于将第三方应用程序与 jira 4.3 集成的身份验证方案有哪些?显然,从客户那里获取用户名和密码听起来很荒谬。另外,我知道 oauth 身份验证仅适用于版本 5。请告诉我。谢谢。 附注。我不是在
我有一个使用 DDS 的旧版 C++ 应用程序用于异步通信/消息传递。我需要将此应用程序集成到使用 JMS 进行消息传递的 JavaEE 环境中。除了构建独立的 JMS/DDS 桥接模块之外,我还有其
我正在尝试使用 Whatsapp 发送测试消息,但收到此错误消息: "error":{"code":27,"description":"Recipient not available on chann
我想将 photologue 与我的 Django 应用程序集成,并使用它在车辆库存中显示照片......有点像 Boost Motor Group Inc. 提供的内容。我已经集成了该应用程序,所以
我目前正在尝试弄清楚如何与 fujitsu scansnap 扫描仪集成,但没有从 fujitsu 找到有关 fujitsu scansnap 管理器如何调用您的应用程序并将文件发送到您的应用程序的详
在我的项目中,我使用了 9 个(九个)int-ip:udp-inbound-channel-adapter 和一个 jms:inbound-channel-adapter。 Jms 适配器从服务器接收
在我们当前的原型(prototype)中,大多数标准 HTML 控件都被小程序取代,最重要的是表单提交由小程序触发。 有没有一种方法可以像 一样在服务器端调用关联的操作 ? 本文Applet and
是否可以使用 twilio 号码从 whatsapp 发送/接收短信?有人用whatsapp试过twilio吗?我问过客服,如果可能的话,他说,不确定,但很多人都问过这个问题。 最佳答案 万一其他人来
我们办公室中几乎不存在版本控制,这显然导致了很多麻烦。我们想使用SVN和Notepad++进行设置...任何人都对如何实现此目标有任何想法?我已经开始研究并浏览了这个网站: http://www.sw
曾经有提供这种集成的 spring-modules 项目;但是,该项目现已弃用。现在有没有人继续支持这种集成?谢谢。 最佳答案 工作正在进行中。 http://blog.athico.com/sear
我的理解是,根据 http://wiki.dbpedia.org/Datasets,DBpedia 从 YAGO 获取类层次结构,而不是实体。 .但是,类似 http://dbpedia.org/cl
任何人都可以帮助我如何将 OpenCMS 与 Java Spring Web 应用程序集成。已经用谷歌搜索并浏览了很多网站但没有用。所以,请帮助我。 最佳答案 我认为将 SpringMVC 与 Ope
我正在尝试使用新的 migs getaway (MPGS) 我遵循了下一个 url 中的代码 https://ap-gateway.mastercard.com/api/documentation/i
关闭。这个问题不满足Stack Overflow guidelines .它目前不接受答案。 想改善这个问题吗?更新问题,使其成为 on-topic对于堆栈溢出。 4年前关闭。 Improve thi
我有一个 cmake 项目。我想轻松完成以下操作 搜索光标下任何变量、函数等的声明、定义和引用,这些可能在外部头文件中声明,其路径是在CMakeLists.txt中使用INCLUDE_DIRECTOR
有人能给我指点一下 Objective-C(或 c/c++)库的方向,或者教通过 FTP 上传或下载的教程(Objective-C)吗?最好能展示如何将文件下载到临时目录,然后稍后上传?我不介意针对
集成()给出了非常错误的答案: integrate(function (x) dnorm(x, -5, 0.07), -Inf, Inf, subdivisions = 10000L) # 2.127
我是一名优秀的程序员,十分优秀!