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收发消息

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/36091.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

模拟 枚举

分享牛客算法基础精选题单题目打卡!!! 目录 字符串的展开 多项式输出 机器翻译 : 铺地毯 : [NOIP2016]回文日期 字符串的展开 原题链接 : 字符串的展开 思路 : 模拟 代码 : #include<iostream> #include<cstring> #include<algorithm> using na…

Java课题笔记~ ServletContext

单个Servlet的配置对象 web.xml <servlet><servlet-name>FirstServlet</servlet-name><servlet-class>com.ambow.test.FirstServlet</servlet-class><init-param><param-name>charset</param-name><param-value>utf-8&…

centos自动同步北京时间

1、安装ntpdate服务 yum -y install ntpdate 2、加入自动任务计划 查找ntpdate的路径&#xff1a; which ntpdate 复制这个路径。 编辑自动任务计划并加入ntpdate&#xff1a; crontab -e # 每小时第30分钟同步AD域控时间 30 * * * * /usr/sbin/ntpdate -u 192.168.2.8 > …

DP——动态规划

DP——动态规划 动态规划算法动态规划的一般步骤特殊DP——背包0-1背包问题完全背包问题 总结 动态规划算法 当涉及到解决具有重叠子问题的优化问题时&#xff0c;动态规划是一种常用的算法技术。它通过将问题分解为一系列重叠子问题&#xff0c;并使用递归或迭代的方式来解决…

Spring Cloud Gateway系例—GatewayFilter 工厂

目录 6.1.AddRequestHeader 6.2.AddRequestHeadersIfNotPresent 6.3.AddRequestParameter 6.4.AddResponseHeader 6.5.CircuitBreaker 6.5.1. 熔断指定的状态码 6.6.CacheRequestBody 6.7.DedupeResponseHeader 6.8.FallbackHeaders 6.9.JsonToGrpc 6.10.LocalRespo…

TypeScript 非空断言

TypeScript 非空断言 发布于 2020-04-08 15:20:15 17.5K0 举报 一、非空断言有啥用 介绍非空断言前&#xff0c;先来看个示例&#xff1a; function sayHello(name: string | undefined) {let sname: string name; // Error } 对于以上代码&#xff0c;TypeScript 编译器…

用户端Web自动化测试-L1

目录&#xff1a; Web自动化测试价值与体系环境安装与使用自动化用例录制自动化测试用例结构分析web浏览器控制常见控件定位方法强制等待与隐式等待常见控件交互方法自动化测试定位策略搜索功能自动化测试用户端Web自动化测试 1.Web自动化测试价值与体系 功能测试场景: UI 自…

IntelliJ Idea 编译时控制台上中文输出乱码

猜测原因是IDEA启动时未指定编码信息&#xff0c;故与系统编码保持一致&#xff08;windows中文系统默认为GBK编码&#xff09;,当以UTF-8编码进行编译在控制台会以GBK编码输出,从而导致乱码 解决方案 指定Idea启动时JVM的默认编码为UTF-8 Help -> Edit Custom Options P…

本地图片的image加密解密-Python 3.10-win10

本地图片的image加密解密- Python 3.10 pyt3int22 -根据1zip下图片批量生成加密的-物体识别.py import ioimport os import base64 import json # 指定图片文件夹 image_dir = "./1zip/" base64code_dir = "./base64code/" base64_to_dir = "./bas…

AUTOSAR规范与ECU软件开发(基础篇)2.5 AUTOSAR方法论

前言 AUTOSAR方法论(AUTOSAR Methodology) 中车用控制器软件的开发涉及系统级、 ECU级和软件组件级。 系统级主要考虑系统功能需求、 硬件资源、 系统约束, 然后建立系统架构; ECU级根据抽象后的信息对ECU进行配置; 系统级和ECU级设计的同时, 伴随着软件组件级的开发。 上…

Sql server还原失败(数据库正在使用,无法获得对数据库的独占访问权)

一.Sql server还原失败(数据库正在使用,无法获得对数据库的独占访问权) 本次测试使用数据库实例SqlServer2008r2版 错误详细&#xff1a; 标题: Microsoft SQL Server Management Studio ------------------------------ 还原数据库“Mvc_HNHZ”时失败。 (Microsoft.SqlServer.…

《甲午》观后感——GPT-3.5所写

《甲午》是一部令人深思的纪录片&#xff0c;通过生动的画面和真实的故事&#xff0c;向观众展示了中国历史上的一段重要时期。观看这部纪录片&#xff0c;我深受触动&#xff0c;对历史的认识也得到了深化。 首先&#xff0c;这部纪录片通过精心搜集的历史资料和珍贵的影像资料…

低成本搭建NAS,利用HFS进行内网穿透,实现公网访问

通过HFS低成本搭建NAS&#xff0c;并内网穿透实现公网访问 文章目录 通过HFS低成本搭建NAS&#xff0c;并内网穿透实现公网访问前言1.下载安装cpolar1.1 设置HFS访客1.2 虚拟文件系统 2. 使用cpolar建立一条内网穿透数据隧道2.1 保留隧道2.2 隧道名称2.3 成功使用cpolar创建二级…

JMS 消息队列接口基本使用指南

概述 介绍 JMS&#xff08;Java Message Service&#xff09;即 Java 消息服务应用程序接口&#xff0c;是一个 Java 平台中关于面向消息中间件&#xff08;MOM&#xff09;的 API&#xff0c;用于在两个应用程序之间&#xff0c;或分布式系统中发送消息&#xff0c;进行异步…

[保研/考研机试] KY103 2的幂次方 上海交通大学复试上机题 C++实现

题目链接&#xff1a; KY103 2的幂次方 https://www.nowcoder.com/share/jump/437195121691999575955 描述 Every positive number can be presented by the exponential form.For example, 137 2^7 2^3 2^0。 Lets present a^b by the form a(b).Then 137 is present…

k8s containerd 配置 http访问harbor image【最新--官方文档】

不看官方文档的代价&#xff1a;在搜索了很多中文资料发现配置了都不起作用&#xff0c;浪费了很多时间。 https://github.com/containerd/containerd/blob/main/docs/cri/config.md#registry-configuration The old CRI config pattern for specifying registry.mirrors and…

MySQL8安装和删除教程 保姆级(Windows)

下载 官网: mysql官网点击Downloads->MySQL Community(GPL) Downloads->MySQL Community Server(或者点击MySQL installer for Windows) Windows下有两种安装方式 在线安装 一般带有 web字样 这个需要联网离线安装 一般没有web字样 安装 下载好之后,版本号可以不一样&…

Postman中,既想传递文件,还想传递多个参数(后端)

需求:既想传文件又想传多个参数可以用以下方式实现

Django rest_framework Serializer中的create、Views中的create/perform_create的区别

Django rest_framework Serializer中的create、Views中的create/perform_create的区别 对于后端来说&#xff0c;前后端分离的方式能让前后端的开发都爽。和所有的爽一样&#xff0c;每爽一次都要付出一定的代价。而前后端分离的代价&#xff0c;就是后端要面对巨量的模块化的功…

C语言实现插入排序

什么是插入排序&#xff1f; 插入排序&#xff08;Insertion Sort&#xff09; 是一种简单且逐步构建有序序列的排序算法。它的思想是将数组分为两部分&#xff1a;已排序的部分和未排序的部分。初始时&#xff0c;已排序部分只包含数组的第一个元素&#xff0c;然后逐步将未排…