Kafka3.1部署和Topic主题数据生产与消费
文章目录
- 前言
- 一、Kafka3.1X版本在Windows11主机部署
- 二、Kafk生产Topic主题数据
- 1.kafka生产数据
- 2.JAVA kafka客户端消费数据
- 总结
前言
本章节主要讲述Kafka3.1X版本在Windows11主机下部署以及JAVA对Kafka应用:
一、Kafka3.1X版本在Windows11主机部署
1.安装JDK配置环境变量
2.Zookeeper(zookeeper-3.7.1)
zk
部署后的目录位置:D:\setup\apache-zookeeper-3.7.1
3.安装Kafka3.1X
3.1 下载包(kafka_2.12-3.1.2.tgz)
Kafka
3.2、 解压并进入Kafka目录:
根目录:D:\setup\kafka3.1.2
3、 编辑config/server.properties文件
注意 log.dirs=D:\setup\kafka3.1.2\logs 为根目录下的\logs
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://localhost:9092
log.dirs=D:\\setup\\kafka3.1.2\\logs
4.运行Zookeeper
Zookeeper安装目录D:\setup\apache-zookeeper-3.7.1\bin,按下Shift+右键,选择“打开命令窗口”选项,打开命令行
.\zkServer.cmd;
5.运行Kafka
Kafka安装目录D:\setup\kafka3.1.2,按下Shift+右键,选择“打开命令窗口”选项,打开命令行
.\bin\windows\kafka-server-start.bat .\config\server.properties
二、Kafk生产Topic主题数据
1.kafka生产数据
创建Topic主题heima
.\bin\windows\kafka-topics.bat --bootstrap-server localhost:9092 --create --topic heima --partitions 2 --replication-factor 1
Created topic heima.
查看Topic主题heima
.\bin\windows\kafka-topics.bat --describe --bootstrap-server localhost:9092 --topic heima
Topic主题heima生产数据
.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic heima
在 > 符号后输入数据:
{"mobilePhone":"186xxxx1234","roleCode":"super_admin_xxx"}
2.JAVA kafka客户端消费数据
2.1 pom.xml文件配置kafka客户端-kafka-clients-2.0.1版本
<!-- kafka客户端 --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.0.1</version></dependency>
2.2 JAVA数据读取文件
package com.ems.mgr.web.controller.thirdparty;
import com.alibaba.fastjson.JSONObject;
import com.ems.mgr.common.utils.spring.SpringUtils;
import com.ems.mgr.system.service.ISysUserService;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;/*** Kafka服务器操作与数据读取*/
public class KafkaUtilDemo {public static final Logger log = LoggerFactory.getLogger(KafkaUtilDemo.class);public static final Properties props = new Properties();
// protected ISysUserService userService = SpringUtils.getBean(ISysUserService.class);public static void init(String kafakservers) {// 配置Kafka消费者属性props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafakservers);props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");}/*** 持续监听并处理kafa消息,当手机号mobilePhone非空时进入数据同步操作* @param kafaktopic* @return*/public static String poll(String kafaktopic) {String msg = "";try {KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Collections.singletonList(kafaktopic));log.info("Kafka消费者订阅指定主题,持续监听并处理消息");while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(60000));for (ConsumerRecord<String, String> record : records) {log.info("offset = " + record.offset() + ",key = " + record.key() + ",value = " + record.value());msg = record.value();if (!StringUtils.isBlank(record.value())) {JSONObject jsonObject = JSONObject.parseObject(record.value());String mobilePhone = jsonObject.getString("mobilePhone");if (StringUtils.isBlank(mobilePhone)) {log.error("Kafka消费者手机号mobilePhone为空");} else {KafkaUtilDemo kafkaUtil = new KafkaUtilDemo();kafkaUtil.syncSystemInfoTask(jsonObject);}}}}} catch (Exception e) {log.error("Kafka消费者订阅指定主题,持续监听并处理消息 error msg=" + e.getMessage());}return msg;}public boolean syncSystemInfoTask(JSONObject jsonObject) {boolean repsBln = true;try {String mobilePhone = jsonObject.getString("mobilePhone");String roleType = jsonObject.getString("roleType");String roleCode = jsonObject.getString("roleCode");log.info("业务数据同步操作................");} catch (Exception e) {repsBln = false;log.error("Kafka消费者同步入库异常,error msg=" + e.getMessage());}return repsBln;}public static void main(String[] args) {try {String kafakservers = "localhost:9092";String kafaktopic = "heima";init(kafakservers);poll(kafaktopic);} catch (Exception e) {log.error("error msg=" + e.getMessage());}}}
3 执行KafkaUtilDemo 文件,查看消费数据。
总结
pom.xml文件在引入spring-kafka 会由于版本问题出现
org.apache.kafka
kafka-clients
2.0.1
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.2.8.RELEASE</version></dependency>
相关文章:
Kafka3.1部署和Topic主题数据生产与消费
文章目录 前言一、Kafka3.1X版本在Windows11主机部署二、Kafk生产Topic主题数据1.kafka生产数据2.JAVA kafka客户端消费数据 总结 前言 本章节主要讲述Kafka3.1X版本在Windows11主机下部署以及JAVA对Kafka应用: 一、Kafka3.1X版本在Windows11主机部署 1.安装JDK配…...
ICIF2023化工展首亮相,宏工科技解决方案助力制造升级
ICIF China 2023中国国际化工展览会于9月4日-6日在上海新国际博览中心举办。宏工科技携化工物料处理一站式解决方案首次亮相,同化工行业全产业链共叙物料处理自动化未来。 宏工科技是一家提供物料处理自动化设备、系统与服务的国家级高新技术企业,业务覆…...
本地部署kubesphere集群
本地部署kubesphere集群 本文采用一主两从结构 1.前置硬件准备 准备最少3台机器,本人分配如下 IP:192.168.58.10 (主) 192.168.58.11 (节点1) 192.168.58.12 (节点2) 系统镜像…...
HNU小学期工训-STC15单片机模型大作业实验报告
STC15单片机模型大作业实验报告 全称:基于STC15单片机与OLED显示模块&PC端演示的多功能声光温振时钟智能手表模型 计科210X 甘晴void 202108010XXX 【请注意:本作业入选优秀范例,直接照抄源码有很大风险】 【建议理解原理之后作改动】 …...
【计算机网络】 TCP协议头相关知识点
文章目录 TCP协议头 TCP协议头 我们来看一下TCP协议头里都有什么东西,研究一下为什么TCP协议是可靠的呢 TCP协议可靠是因为在协议头里带着一些校验的数据 首先是源端口和目的端口,这两个是UDP中也有的,但是UDP中只有这两个,没有…...
深度学习相关VO梳理
相关论文 基于学习的VO 相关: DeepVO Towards End-to-End Visual Odometry with Deep Recurrent Convolutional Neural Networks(ICRA,2017) TartanVO: A Generalizable Learning-based VO(CoRL2021) SimVODIS: Simultaneous Vis…...
SpringMVC---CRUD实现
思路分析 搭建环境逆向生层对应的类(model、mapper.xml、mapper.java)编写业务逻辑层编写web层(控制器)前端页面 一、环境搭建 1.1、导入项目所需依赖(pom.xml) <project xmlns"http://maven.apache.org/POM/4.0.0"…...
vue+elementUI el-select 自定义搜索逻辑(filter-method)
下拉列表的默认搜索是搜索label显示label,我司要求输入id显示label名称 <el-form-item label"部门:"><el-select v-model"form.region1" placeholder"请选择部门" filterable clearable:filter-method"dataFilter&qu…...
数据库——事务
事务是指作为一个整体被执行的一系列操作。在数据库管理系统中,事务是指一组数据库操作(如插入、更新、删除等)的逻辑单元,也就是说事务的本质是把多个操作打包成一个操作,并且它要么完全执行,要么完全不执…...
echarts折线图每段显示不同的颜色
效果图 配置项: zqChartFour: {title: {text: "一天用电量分布",subtext: "纯属虚构",},tooltip: {trigger: "axis",axisPointer: {type: "cross",},},toolbox: {show: true,feature: {saveAsImage: {},},},xAxis: {type:…...
设计模式-单例模式(Singleton)
文章目录 前言一、单例模式的概念二、单例模式的实现三、单例模式的应用场景四、单例模式优缺点优点:缺点:总结 前言 单例模式(Singleton Pattern)是一种创建型设计模式,它确保一个类只有一个实例,并提供一…...
优漫动游 常见的AI视频生成网站的官方网站:
1、Lumen5 Lumen5是一款在线视频制作工具,利用人工智能技术能够迅速将文本、和音乐转换为视频。它可以帮助你把博客文章、社交媒体内容等转化为吸引人的视频,从而提高你的品牌曝光率和社交媒体的参与度。 2.Animoto Animoto是一个视频制作平台&…...
Vue中数据可视化关系图展示与关系图分析
Vue中数据可视化关系图展示与关系图分析 数据可视化是现代Web应用程序的重要组成部分之一,它可以帮助我们以图形的方式呈现和分析复杂的数据关系。Vue.js是一个流行的JavaScript框架,它提供了强大的工具来构建数据可视化应用。本文将介绍如何使用Vue.js…...
【启扬方案】基于启扬安卓屏一体机的医疗手推车解决方案
医疗手推车作为医院基础设施的一部分,被广泛应用于医院内部,包括急诊室、手术室、病房和其他临床部门。伴随着互联网技术的发展和行业的渗透,智慧医疗受到越来越多的青睐,这也使得很多医疗设施得到了改进,医疗手推车也…...
JavaScript实现MD5加密的6种方式
关于MD5: MD5.js是通过前台js加密的方式对用户信息,密码等私密信息进行加密处理的工具,也可称为插件。 在本案例中 可以看到MD5共有6种加密方法: 1, hex_md5(value) 2, b64_md5(value) 3, …...
腾讯云和阿里云2核2G服务器租用价格表对比
2核2G云服务器可以选择阿里云服务器或腾讯云服务器,腾讯云轻量2核2G3M带宽服务器95元一年,阿里云轻量2核2G3M带宽优惠价108元一年,不只是轻量应用服务器,阿里云还可以选择ECS云服务器u1,腾讯云也可以选择CVM标准型S5云…...
抖音无需API开发连接Stable Diffusion,实现自动根据评论区的指令生成图像并返回
抖音用户使用场景: 随着AI绘图的热度不断升高,许多抖音达人通过录制视频介绍不同的AI工具,包括产品背景、使用方法以及价格等,以吸引更多的用户。其中,Stable Diffusion这款产品受到了许多博主达人的青睐。在介绍这款产…...
MySQL(三)
DDL(数据定义语言) 库 /* 创建数据库testone */ create database testone; /* 查询数据库testone */ show databases; /* 选择数据库testone */ use testone; /* 删除数据库testone */ drop database testone; 表 创建表 create table table_name (…...
汽车级肖特基二极管DSS220-Q 200V 2A
DSS220-Q是什么二极管?贵司有生产吗? 肖特基二极管DSS220-Q符合汽车级AEC Q101标准吗? DSS220-Q贴片肖特基二极管参数是什么封装?正向电流和反向电压是多大? DSS220-Q肖特基二极管需要100KK,有现货吗&#…...
maven jetty post 上传长度设置
maven jetty post 上传长度设置 <plugin><groupId>org.eclipse.jetty</groupId><artifactId>jetty-maven-plugin</artifactId><version>9.4.8.v20171121</version><configuration><scanIntervalSeconds>1</scanInter…...
LeetCode 面试题 03.03. 堆盘子
文章目录 一、题目二、C# 题解 一、题目 堆盘子。设想有一堆盘子,堆太高可能会倒下来。因此,在现实生活中,盘子堆到一定高度时,我们就会另外堆一堆盘子。请实现数据结构 SetOfStacks,模拟这种行为。SetOfStacks 应该由…...
Python-函数进阶
函数的多返回值 按照返回值的顺序, 写对应顺序的多个变量接受即可, 变量之间用逗号隔开,支持不同类型的数据return def test_return():return 1, 2, 3x, y, z test_return()print(x) print(y) print(z)函数参数种类 使用方式上的不同&am…...
实操Hadoop大数据高可用集群搭建(hadoop3.1.3+zookeeper3.5.7+hbase3.1.3+kafka2.12)
前言 纯实操,无理论,本文是给公司搭建测试环境时记录的,已经按照这一套搭了四五遍大数据集群了,目前使用还未发现问题。 有问题麻烦指出,万分感谢! PS:Centos7.9、Rocky9.1可用 集群配置 iph…...
如何在 Ubuntu 上安装和使用 Nginx?
ginx(发音为“engine-x”)是一种流行的 Web 服务器软件,以其高性能和可靠性而闻名。它是许多流行网站使用的开源软件,包括 Netflix、GitHub 和 WordPress。Nginx 可以用作 Web 服务器、负载均衡器、反向代理和 HTTP 缓存等。 它以…...
seatunnel win idea 本地调试
调试FakeSource,LocalFile # Set the basic configuration of the task to be performed env {execution.parallelism 1job.mode "BATCH" }# Create a source to connect to Mongodb source {# This is a example source plugin **only for test and d…...
链路追踪Skywalking快速入门
目录 1 Skywalking概述1.1 微服务系统监控三要素1.2 什么是链路追踪1.2.1 链路追踪1.2.2 OpenTracing1、数据模型:2、核心接口语义 1.3 常见APM系统1.4 Skywalking介绍1、SkyWalking 核心功能:2、SkyWalking 特点:3、Skywalking架构图&#x…...
全开源影视APP源码带后台 苍穹影视APP源码 免受权带安装教程
苍穹影视 V20 全新后台七彩视界免受权开源源码此版本为天穹公益版开源无解密安装教程 全新后台很是都雅,源码全开源无加密。 PC 端对接教程: 建议在浮图下操作 正常安装前后端 然后安装米酷 cms 根据教程安装即可 米酷 cms 对接部门已被我改动,只要在安装…...
Qt+C++自建网页浏览器-Chrome blink最新内核基础上搭建-改进版本
程序示例精选 QtC自建网页浏览器-Chrome blink最新内核基础上搭建-改进版本 如需安装运行环境或远程调试,见文章底部个人QQ名片,由专业技术人员远程协助! 前言 这篇博客针对<<QtC自建网页浏览器-Chrome blink最新内核基础上搭建-改进版…...
这场科技巨变,有生之年有希望
见到一文,遂分享欲爆棚,总结如下。 具有人类水平的人工智能大约什么时候可以出现? 人类水平的人工智能,指的是,不需要借助人类,机器能够比人类更好地完成每项任务。 针对这个问题,有家机构在201…...
zemax优化功能
1、三种优化方法 zemax的三种优化方法中,局部优化会找到局部的极小值点,全局优化会找到整体的最小值点。 锤形优化适用于先用全局优化找到大概值后,进一步完善光学系统 对于评价函数单调或者局部最小值就是全局最小值的情况,使…...
计算机网站建设 是什么/seoaoo
在训练分类器之前有一步骤是样本的创建,而创建之前须对正样本与副样本缩放到统一大小,如何对整个文件夹里的图片进行批缩放呢?程序如下: 在VC6.0opencv2.1运行没问题, 在vs2010下面出现如下错误: error C24…...
wordpress调用单页面/seo研究中心道一老师
主要内容:函数指针一、函数指针定义int maxValue(int a,int b){return a > b ?a : b;}函数名和数组名一样是地址,存在在代码区int maxValue(int a。int b)int (*p)(int。int) NULL函数指针定义。p是变量,其它是类型…...
搜索别人的网站是带logo的请问怎么做的/南京响应式网站建设
有 时当 使用 一个 包含 多 个 参 数 的 方 法 时 , 由 于参 数 过 多 会 导 致 可 读 性 严 重 下 降 , 如 : 有 时当 使用 一个 包含 多 个 参 数 的 方 法 时 , 由 于参 数 过 多 会 导 致 可 读 性 严 重 下 降 , …...
zencart 一个产品网站下单/长沙网站se0推广优化公司
269页程序清单10.19 flc.c程序有错误。 如图中所示,方法1处应加上const限定,否则有些编译器会出现从不兼容指针传递参数的警告,方法2没有方法1安全,方法2是去除原型和定义中的const限定。 下面展示第1种修改方法: 像…...
网站建设资讯/推广平台都有哪些
一个Project 引用 另外 一个Project 显示黄色叹号,后来发现 后一本Project的build设定为.Net4.5, 前一个为4.0, 将版本改为一致后,问题解决。转载于:https://www.cnblogs.com/fdyang/p/4270041.html...
bt磁力搜索引擎/厦门关键词优化平台
Linux中一切皆文件 一、什么是进程 在linux中,每个程序都会有一个自己的进程,每个进程都有一个id号和父进程进程有两种存在方式:前台与后台,一般服务都是后台,而程序是前台运行 二、ps命令 ps :查看当前系…...