Kafka:安装和配置

producer:发布消息的对象,称为消息产生者 (Kafka topic producer)
topic:Kafka将消息分门别类,每一个消息称为一个主题(topic)
consumer:订阅消息并处理发布消息的对象称为消费者(consumer)
broker:已发布的消息保存在一组服务器中,称为kafka集群,集群中的每一个服务器都是一个代理(broker),消费者(consumer)可以订阅一个或者多个主题(topic),并从broker中拉取数据,从而消费这些已发布的信息。
1、Kafka对zookeeper是一个强依赖,保存Kafka相关的节点数据,所以安装kafka之前要先安装zookeeper
下载镜像
docker pull zookeeper:3.4.14
创建容器
docker run -d --name zookeeper -p 2181:2181 zookeeper:3.4.14
下载镜像
docker pull wurstmeister/kafka:2.12-2.3.1
创建容器
docker run -d --name kafka \
--env KAFKA_ADVERTISED_HOST_NAME=192.168.200.130 \
--env KAFKA_ZOOKEEPER_CONNECT=192.168.200.130:2181 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.200.130:9092 \
--env KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
--env KAFKA_HEAP_OPTS="-Xmx256M -Xms256M" \
--net=host wurstmeister/kafka:2.12-2.3.1
2、入门案例
①创建kafka-demo工程并引入依赖
<!--kafka--><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency>
②创建ProducerQuickStart生产者类并实现
package com.heima.kafkademo.sample;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.Properties;/*生产者*/
public class ProducerQuickStart {public static void main(String[] args) {/*1、kafka配置信息*/Properties properties = new Properties();//kafka连接地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");//发送失败,失败的重试次数properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,5);//key和value的序列化器properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");/*2、生产对象*/KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);//封装发送消息的对象ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");/*3、发送消息*/producer.send(record);/*4、关闭通道,负责消息发送不成功*/producer.close();}
}
③创建ConsumerQuickStart消费者类并实现
package com.heima.kafkademo.sample;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.clients.producer.ProducerConfig;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;/*消费者*/
public class ConsumerQuickStart {public static void main(String[] args) {/*1、kafka配置信息*/Properties properties = new Properties();//kafka连接地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");//发送失败,失败的重试次数properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,5);//key和value的序列化器properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");/*2、消费者对象*/KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);/*3、订阅主题*/consumer.subscribe(Collections.singletonList("itheima-topic"));//当前线程处于一直监听状态while (true){//4、获取消息ConsumerRecords<String,String> consumerRecords =consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : consumerRecords) {System.out.println(record.key());System.out.println(record.value());}}}
}
④运行测试
成功接收到消息

总结
-
生产者发送消息,多个消费者订阅同一个主题,只能有一个消费者收到消息(一对一)
-
生产者发送消息,多个消费者订阅同一个主题,所有消费者都能收到消息(一对多)
下一篇: springboot集成kafka收发消息
相关文章:
Kafka:安装和配置
producer:发布消息的对象,称为消息产生者 (Kafka topic producer) topic:Kafka将消息分门别类,每一个消息称为一个主题(topic) consumer:订阅消息并处理发布消息的对象…...
786. 第k个数
文章目录 QuestionIdeasCode Question 给定一个长度为 n 的整数数列,以及一个整数 k ,请用快速选择算法求出数列从小到大排序后的第 k 个数。 输入格式 第一行包含两个整数 n 和 k 。 第二行包含 n 个整数(所有整数均在 1∼109 范围内&…...
用友-NC-Cloud远程代码执行漏洞[2023-HW]
用友-NC-Cloud远程代码执行漏洞[2023-HW] 一、漏洞介绍二、资产搜索三、漏洞复现PoC小龙POC检测脚本: 四、修复建议 免责声明:请勿利用文章内的相关技术从事非法测试,由于传播、利用此文所提供的信息或者工具而造成的任何直接或者间接的后果及损失&#…...
Transformer(二)(VIT,TNT)(基于视觉CV)
目录 1.视觉中的Attention 2.VIT框架(图像分类,不需要decoder) 2.1整体框架 2.2.CNN和Transformer遇到的问题 2.3.1CNN 2.3.2Transformer 2.3.3二者对比 2.4.公式理解 3TNT 参考文献 1.视觉中的Attention 对于人类而言看到一幅图可以立…...
Scratch 详解 之 线性→代数之——求两线段交点坐标
可能有人要问:求交点坐标有什么用呢?而且为啥要用线代来求?直线方程不行吗??? 这个问题,我只能说,直线方程计算的次数过多了,而且动不动就要考虑线的方向,90的…...
Python-组合数据类型
今天要介绍的是Python的组合数据类型 整理不易,希望得到大家的支持,欢迎各位读者评论点赞收藏 感谢! 目录 知识点知识导图1、组合数据类型的基本概念1.1 组合数据类型1.2 集合类型概述1.3 序列类型概述1.4 映射类型概述 2、列表类型2.1 列表的…...
vue3+vue-simple-uploader实现大文件上传
vue-simple-uploader本身是基于vue2实现,如果要使用vue3会报错。如何在vue3中使用,可参考我的另一篇文章:解决vue3中不能使用vue-simple-uploader__Jyann_的博客-CSDN博客 一.实现思路 使用vue-simple-uploader组件的uploader组件,设置自动上传为false,即可开启手动上传。…...
自适应变异麻雀搜索算法及其Matlab实现
麻雀搜索算法( sparrow search algorithm,SSA) 是2020 年新提出的一种元启发式算法[1],它是受麻雀种群的觅食和反捕食行为启发,将搜索群体分为发现者、加入者和侦察者 3 部分,其相互分工寻找最优值,通过 19 个标准测试…...
ETL技术入门之ETLCloud初认识
首先ETL是什么? ETL代表“Extract, Transform, Load”,是一种用于数据集成和转换的过程。它在数据管理和分析中扮演着重要的角色。下面我们将分解每个步骤: Extract(抽取): 这一步骤涉及从多个不同的数据源…...
uniapp项目如何运行在微信小程序模拟器上
在HbuilderX中的小程序写完后自己一定要保存,否则会出不来效果 那么怎么让uniapp项目运行在微信小程序开发工具中呢 1 在hbuilderx中点击运行到小程序模拟器 2 然后在项目目录中会生成一个文件夹 在微信小程序开发软件中的工具>安全设置>打开端口 或者在微…...
数据挖掘全流程解析
数据挖掘全流程解析 数据指标选择 在这一阶段,使用直方图和柱状图的方式对数据进行分析,观察什么数据属性对于因变量会产生更加明显的结果。 如何绘制直方图和条形统计图 数据清洗 观察数据是否存在数据缺失或者离群点的情况。 数据异常的两种情况…...
详细介绍如何对音乐信息进行检索和音频节拍跟踪
在本文中,我们将了解节拍的概念,以及我们在尝试跟踪节拍时面临的挑战。然后我们将介绍解决问题的方法以及业界最先进的解决方案。 介绍 音乐就在我们身边。每当我们听到任何与我们的心灵和思想相关的音乐时,我们就会迷失其中。我们下意识地随着听到的节拍而敲击。您一定已…...
Java课题笔记~ HTTP协议(请求和响应)
Servlet最主要的作用就是处理客户端请求,并向客户端做出响应。为此,针对Servlet的每次请求,Web服务器在调用service()方法之前,都会创建两个对象 分别是HttpServletRequest和HttpServletResponse。 其中HttpServletRequest用于封…...
在x86下运行的Ubuntu系统上部署QEMU用于模拟RISC-V硬件环境
1.配置工作环境 sudo apt install gcc bison flex libncurses-dev ninja-build \pkg-config build-essential zlib1g-dev pkg-config libglib2.0-dev \binutils-dev libboost-all-dev autoconf libtool libssl-dev \libpixman-1-dev python-capstone virtualenv software-prop…...
网络爬虫选择代理IP的标准
Hey,小伙伴们!作为一家http代理产品供应商,我知道网络爬虫在选择代理IP时可能会遇到些问题,毕竟市面上有很多选择。别担心!今天我要给大家分享一些实用的建议,帮助你们选择适合网络爬虫的代理IP。一起来看看…...
RxJava 复刻简版之三,map 多次中转数据
案例代码:https://gitee.com/bobidali/lite-rx-java/commit/292e9227a5491f7ec6a07f395292ef8e6ff69290 RxJava 的调用第一步是封装了观察者接受了数据的处理,进一步就是使用 map 将数据操作传递给上下游 1、类似Observer.create 创建一个简单的观察者…...
06 Word2Vec模型(第一个专门做词向量的模型,CBOW和Skip-gram)
博客配套视频链接: https://space.bilibili.com/383551518?spm_id_from=333.1007.0.0 b 站直接看 配套 github 链接:https://github.com/nickchen121/Pre-training-language-model 配套博客链接:https://www.cnblogs.com/nickchen121/p/15105048.html 神经网络语言模型(NNL…...
Axure RP9小白安装教程
第一步: 打开:Axure中文学习网 第二步: 鼠标移动软件下载,点击Axure RP 9下载既可 第三步: 注意:Axure RP 9 MAC正式版为苹果版本,Axure RP 9 WIN正式版为Windows版本 中文汉化包ÿ…...
腾讯云CVM服务器2核2g1m带宽支持多少人访问?
腾讯云2核2g1m的服务器支持多少人同时访问?2核2g1m云服务器短板是在1M公网带宽上,腾讯云服务器网以网站应用为例,当大规模用户同时访问网站时,很大概率会卡在公网带宽上,所以压根就谈不上2核2G的CPU内存计算性能是否够…...
8.12学习笔记
在PyTorch中,Dataset和DataLoader是用于处理数据的两个重要类。Dataset类是一个抽象类,用于表示数据集。它的主要作用是将数据加载到内存中,并提供一种统一的方式来访问数据。为了使用Dataset类,你需要继承它并实现两个方法&#…...
vscode里如何用git
打开vs终端执行如下: 1 初始化 Git 仓库(如果尚未初始化) git init 2 添加文件到 Git 仓库 git add . 3 使用 git commit 命令来提交你的更改。确保在提交时加上一个有用的消息。 git commit -m "备注信息" 4 …...
【网络安全产品大调研系列】2. 体验漏洞扫描
前言 2023 年漏洞扫描服务市场规模预计为 3.06(十亿美元)。漏洞扫描服务市场行业预计将从 2024 年的 3.48(十亿美元)增长到 2032 年的 9.54(十亿美元)。预测期内漏洞扫描服务市场 CAGR(增长率&…...
2024年赣州旅游投资集团社会招聘笔试真
2024年赣州旅游投资集团社会招聘笔试真 题 ( 满 分 1 0 0 分 时 间 1 2 0 分 钟 ) 一、单选题(每题只有一个正确答案,答错、不答或多答均不得分) 1.纪要的特点不包括()。 A.概括重点 B.指导传达 C. 客观纪实 D.有言必录 【答案】: D 2.1864年,()预言了电磁波的存在,并指出…...
Linux C语言网络编程详细入门教程:如何一步步实现TCP服务端与客户端通信
文章目录 Linux C语言网络编程详细入门教程:如何一步步实现TCP服务端与客户端通信前言一、网络通信基础概念二、服务端与客户端的完整流程图解三、每一步的详细讲解和代码示例1. 创建Socket(服务端和客户端都要)2. 绑定本地地址和端口&#x…...
算法:模拟
1.替换所有的问号 1576. 替换所有的问号 - 力扣(LeetCode) 遍历字符串:通过外层循环逐一检查每个字符。遇到 ? 时处理: 内层循环遍历小写字母(a 到 z)。对每个字母检查是否满足: 与…...
Java数值运算常见陷阱与规避方法
整数除法中的舍入问题 问题现象 当开发者预期进行浮点除法却误用整数除法时,会出现小数部分被截断的情况。典型错误模式如下: void process(int value) {double half = value / 2; // 整数除法导致截断// 使用half变量 }此时...
华为OD机考-机房布局
import java.util.*;public class DemoTest5 {public static void main(String[] args) {Scanner in new Scanner(System.in);// 注意 hasNext 和 hasNextLine 的区别while (in.hasNextLine()) { // 注意 while 处理多个 caseSystem.out.println(solve(in.nextLine()));}}priv…...
算法250609 高精度
加法 #include<stdio.h> #include<iostream> #include<string.h> #include<math.h> #include<algorithm> using namespace std; char input1[205]; char input2[205]; int main(){while(scanf("%s%s",input1,input2)!EOF){int a[205]…...
【Java多线程从青铜到王者】单例设计模式(八)
wait和sleep的区别 我们的wait也是提供了一个还有超时时间的版本,sleep也是可以指定时间的,也就是说时间一到就会解除阻塞,继续执行 wait和sleep都能被提前唤醒(虽然时间还没有到也可以提前唤醒),wait能被notify提前唤醒…...
20250609在荣品的PRO-RK3566开发板的Android13下解决串口可以执行命令但是脚本执行命令异常的问题
20250609在荣品的PRO-RK3566开发板的Android13下解决串口可以执行命令但是脚本执行命令异常的问题 2025/6/9 20:54 缘起,为了跨网段推流,千辛万苦配置好了网络参数。 但是命令iptables -t filter -F tetherctrl_FORWARD可以在调试串口/DEBUG口正确执行。…...
