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类,你需要继承它并实现两个方法&#…...
PHP和Node.js哪个更爽?
先说结论,rust完胜。 php:laravel,swoole,webman,最开始在苏宁的时候写了几年php,当时觉得php真的是世界上最好的语言,因为当初活在舒适圈里,不愿意跳出来,就好比当初活在…...
线程同步:确保多线程程序的安全与高效!
全文目录: 开篇语前序前言第一部分:线程同步的概念与问题1.1 线程同步的概念1.2 线程同步的问题1.3 线程同步的解决方案 第二部分:synchronized关键字的使用2.1 使用 synchronized修饰方法2.2 使用 synchronized修饰代码块 第三部分ÿ…...
UDP(Echoserver)
网络命令 Ping 命令 检测网络是否连通 使用方法: ping -c 次数 网址ping -c 3 www.baidu.comnetstat 命令 netstat 是一个用来查看网络状态的重要工具. 语法:netstat [选项] 功能:查看网络状态 常用选项: n 拒绝显示别名&#…...
【算法训练营Day07】字符串part1
文章目录 反转字符串反转字符串II替换数字 反转字符串 题目链接:344. 反转字符串 双指针法,两个指针的元素直接调转即可 class Solution {public void reverseString(char[] s) {int head 0;int end s.length - 1;while(head < end) {char temp …...
【决胜公务员考试】求职OMG——见面课测验1
2025最新版!!!6.8截至答题,大家注意呀! 博主码字不易点个关注吧,祝期末顺利~~ 1.单选题(2分) 下列说法错误的是:( B ) A.选调生属于公务员系统 B.公务员属于事业编 C.选调生有基层锻炼的要求 D…...
大语言模型(LLM)中的KV缓存压缩与动态稀疏注意力机制设计
随着大语言模型(LLM)参数规模的增长,推理阶段的内存占用和计算复杂度成为核心挑战。传统注意力机制的计算复杂度随序列长度呈二次方增长,而KV缓存的内存消耗可能高达数十GB(例如Llama2-7B处理100K token时需50GB内存&a…...
QT3D学习笔记——圆台、圆锥
类名作用Qt3DWindow3D渲染窗口容器QEntity场景中的实体(对象或容器)QCamera控制观察视角QPointLight点光源QConeMesh圆锥几何网格QTransform控制实体的位置/旋转/缩放QPhongMaterialPhong光照材质(定义颜色、反光等)QFirstPersonC…...
解决:Android studio 编译后报错\app\src\main\cpp\CMakeLists.txt‘ to exist
现象: android studio报错: [CXX1409] D:\GitLab\xxxxx\app.cxx\Debug\3f3w4y1i\arm64-v8a\android_gradle_build.json : expected buildFiles file ‘D:\GitLab\xxxxx\app\src\main\cpp\CMakeLists.txt’ to exist 解决: 不要动CMakeLists.…...
通过MicroSip配置自己的freeswitch服务器进行调试记录
之前用docker安装的freeswitch的,启动是正常的, 但用下面的Microsip连接不上 主要原因有可能一下几个 1、通过下面命令可以看 [rootlocalhost default]# docker exec -it freeswitch fs_cli -x "sofia status profile internal"Name …...
jdbc查询mysql数据库时,出现id顺序错误的情况
我在repository中的查询语句如下所示,即传入一个List<intager>的数据,返回这些id的问题列表。但是由于数据库查询时ID列表的顺序与预期不一致,会导致返回的id是从小到大排列的,但我不希望这样。 Query("SELECT NEW com…...
