zookeeper+kafka群集

什么是消息队列

消息队列（Message Queue）是一种通信方法，用于在分布式系统中异步地传递消息。消息队列通过消息中间件将消息从发送者（Producer）传递给接收者（Consumer），允许应用程序解耦并提高系统的可扩展性和可靠性。

消息队列的特征

  1. 异步通信：消息队列允许发送者和接收者独立操作，发送者可以在发送消息后立即返回，不必等待接收者处理。
  2. 解耦：通过消息队列，发送者和接收者不必直接通信，从而减少了系统之间的依赖。
  3. 可靠性：消息队列系统通常提供消息持久化和重试机制，确保消息不会丢失。
  4. 可扩展性：通过消息队列，系统可以容易地扩展，以处理大量的并发消息。
  5. 负载均衡：消息队列可以将消息分发给多个消费者，从而实现负载均衡。
  6. 顺序保证：某些消息队列系统可以保证消息的顺序。

为什么需要消息队列

  1. 异步处理：允许后端系统异步处理任务，提高系统响应速度。
  2. 流量削峰：通过消息队列可以缓冲高峰流量，防止系统过载。
  3. 解耦系统：使得系统组件可以独立开发、部署和维护，提高系统的灵活性。
  4. 提高可靠性：通过持久化消息，确保在系统故障时消息不会丢失。

Kafka 基本概念

Kafka 是一种分布式消息队列系统，旨在处理高吞吐量的实时数据流。它由 LinkedIn 开发，并已捐赠给 Apache 基金会，成为开源项目。

Kafka 角色术语

  1. Producer（生产者）：发布消息到一个或多个 Kafka 主题。
  2. Consumer（消费者）：订阅一个或多个主题并处理发布到这些主题的消息。
  3. Broker：Kafka 集群中的一台服务器，负责接收和存储消息。
  4. Topic：消息的分类，每个主题可以有多个分区。
  5. Partition：主题的子单位，每个分区是一个有序的、不可变的消息序列。
  6. Replica：分区的副本，用于容错和高可用性。
  7. Leader：每个分区都有一个 Leader，负责所有读写请求。
  8. 追随者：从 Leader 副本同步数据的副本。

Topic 和 Partition

  • Topic：Kafka 中消息的类别，可以理解为一个消息队列的名称。
  • Partition：Topic 的分区，一个 Topic 可以有多个分区，分区是 Kafka 实现高吞吐量和并行处理的关键。

Producer 生产机制

  1. 消息生产:P roducer 将消息发布到一个或多个 Topic。
  2. 分区选择:P roducer 可以根据键值或使用默认分区器将消息分配到特定的分区。
  3. 确认机制:P roducer 可以配置不同的确认机制（例如 leader ack，all replicas ack）来确保消息的可靠传递。

Consumer 消费机制

  1. 订阅 Topic：Consumer 订阅一个或多个 Topic，Kafka 将消息推送给 Consumer。
  2. 消费位移：每个 Consumer 维护一个位移（offset），表示当前消费到的消息位置。
  3. 消费组：多个 Consumer 可以组成一个消费组（Consumer Group），Kafka 将消息分配给组内的 Consumer，实现负载均衡。

Zookeeper 概念介绍

ZooKeeper 是一个分布式协调服务，主要用于分布式应用程序中的配置管理、命名服务、分布式同步和组服务。

Zookeeper 的工作原理

  1. 数据模型：ZooKeeper 使用树状的数据模型，数据存储在 znode 中，类似于文件系统。
  2. 一致性保证：ZooKeeper 保证顺序一致性、原子性、单一系统映像和可靠性。
  3. Leader 选举：ZooKeeper 集群通过 Leader 选举保证数据的一致性和容错性。Leader 负责处理所有写请求，Follower 处理读请求。
  4. 会话管理：ZooKeeper 通过会话管理保持客户端连接的状态，检测并处理客户端故障。
  5. 通知机制：客户端可以对特定的 znode 设置监视器，当 znode 状态变化时，ZooKeeper 会通知相应的客户端。

单节点部署kafka

主机：

kafka1:192.168.10.101

1：安装zookeeper

[root@kafka1 ~]# yum -y install java

[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz

[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper

[root@kafka1 ~]# cd /etc/zookeeper/conf

[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg

[root@kafka1 ~]# vim zoo.cfg

dataDir=/etc/zookeeper/zookeeper-data

[root@kafka1 ~]# cd /etc/zookeeper/

[root@kafka1 kafka]# mkdir /etc/zookeeper/zookeeper-data/

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

[root@kafka1 zookeeper]# ./bin/zkServer.sh status

2：安装kafka

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz

[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka

[root@kafka1 ~]# cd /etc/kafka/

[root@kafka1 kafka]# vim config/server.properties

log.dirs=/etc/kafka/kafka-logs      #60行

[root@kafka1 kafka]# mkdir /etc/kafka/kafka-logs

[root@kafka1 kafka]# bin/kafka-server-start.sh  config/server.properties &

检查两个端口的开启状态

[root@kafka1 kafka]# netstat -anpt | grep 2181

[root@kafka1 kafka]# netstat -anpt | grep 9092

注意：启动时先启动zookeeper，关闭时先关闭kafka

如果要关闭zookeeper

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

如果要关闭kafka

[root@192 kafka]# bin/kafka-server-stop.sh

如果关不了，就kill杀死该进程

3：测试

创建topic

bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

列出topic

bin/kafka-topics.sh --list --zookeeper kafka1:2181

查看topic

bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic test

生产消息

bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

消费消息

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

删除topic

bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test

群集部署kafka

主机

kafka1：192.168.10.101

kafka2：192.168.10.102

kafka3：192.168.10.103

1：zookeeper的部署

（1）安装zookeeper（三个节点的配置相同）

[root@kafka1 ~]# yum -y install java

[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz

[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper

（2）创建数据保存目录（三个节点的配置相同）

[root@kafka1 ~]# cd /etc/zookeeper/

[root@kafka1 zookeeper]# mkdir zookeeper-data

（3）修改配置文件（三个节点的配置相同）

[root@kafka1 zookeeper]# cd /etc/zookeeper/conf

[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg

[root@kafka1 ~]# vim zoo.cfg

dataDir=/etc/zookeeper/zookeeper-data

clientPort=2181

server.1=192.168.10.114:2888:3888

server.2=192.168.10.115:2888:3888

server.3=192.168.10.116:2888:3888

注释：zookeeper只用的端口

2181：对cline端提供服务

3888：选举leader使用

2888：集群内机器通讯使用（Leader监听此端口）

（4）创建节点id文件（按server编号设置这个id，三个机器不同）

节点1：

[root@kafka1 conf]# echo '1' > /etc/zookeeper/zookeeper-data/myid

节点2：

[root@kafka2 conf]# echo '2' > /etc/zookeeper/zookeeper-data/myid

节点3：

[root@kafka3 conf]# echo '3' > /etc/zookeeper/zookeeper-data/myid

（5）三个节点启动zookeeper进程

[root@kafka1 conf]# cd /etc/zookeeper/

[root@kafka1 zookeeper]# ./bin/zkServer.sh start

[root@kafka1 zookeeper]# ./bin/zkServer.sh status

2：kafka的部署

（1）kafka的安装（三个节点的配置相同）

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz

[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka

（2）修改配置文件

[root@kafka1 ~]# cd /etc/kafka/

[root@kafka2 kafka]# vim config/server.properties

broker.id=1       ##21行  修改，注意其他两个的id分别是2和3

listeners=PLAINTEXT://192.168.10.114:9092      #31行  修改，其他节点改成各自的IP地址

log.dirs=/etc/kafka/kafka-logs      ## 60行  修改

num.partitions=1      ##65行 分片数量，不能超过节点数

zookeeper.connect=192.168.10.114:2181,192.168.10.115:2181,192.168.10.116:2181

注释：

9092是kafka的监听端口

（3）创建日志目录（三个节点的配置相同）

[root@kafka1 kafka]# mkdir /etc/kafka/kafka-logs

（3）在所有kafka节点上执行开启命令，生成kafka群集（三个节点的配置相同）

[root@kafka1 kafka]# ./bin/kafka-server-start.sh config/server.properties &

如果启动不了，可以将/etc/kafka/kafka-logs中的数据清除再试试

3：测试

创建topic（任意一个节点）

bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

列出topic（任意一个节点）

bin/kafka-topics.sh --list --zookeeper kafka1:2181

bin/kafka-topics.sh --list --zookeeper kafka2:2181

bin/kafka-topics.sh --list --zookeeper kafka3:2181

生产消息

bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

消费消息

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

删除topic

bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test

扩展：将远程的logstash输出到kafka中

[root@logstash1 logstash]# vim conf.d/beats.conf

input {

  beats {

    port => "5044"

    codec => "json"

 }

}

output {

   kafka {

    bootstrap_servers => "192.168.10.114:9092,192.168.10.115:9092,192.168.10.116:9092"

    topic_id => "httpd-filebeat"

    batch_size => "5"

    codec => "json"

   }

}

在logstash中运行配置

[root@logstash1 logstash]# logstash -f /etc/logstash/conf.d/beats.conf

生产消息（任意一个节点）

bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

消费消息（任意一个节点）

bin/kafka-console-consumer.sh --bootstrap-server kafka2:9092 --topic test

删除topic（任意一个节点）

bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test

重启kafka

/etc/kafka/bin/kafka-server-stop.sh

/etc/zookeeper/bin/zkServer.sh restart

/etc/kafka/bin/kafka-server-start.sh config/server.properties &

问题解决

错误提示：

 Failed to acquire lock on file .lock in /tmp/kafka-logs. A Kafka instance in another process or thread is using this directory.

解决方法：

[root@kafka1 kafka]# rm -rf /tmp/kafka-logs/*

[root@kafka1 kafka]# netstat -lnp|grep 9092

pkill杀死kafka的进程号

[root@kafka2 kafka]# ./bin/kafka-server-start.sh config/server.properties &

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

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

相关文章

废品回收小程序：高效便捷回收，推动市场发展

废品回收小程序：高效便捷回收，推动市场发展

随着互联网的发展和人们日益提升的环保意识，对废品回收市场的关注度不断提高，废品回收成为了当下发展前景巨大的行业之一。 传统的废品回收体系不完善，存在较大的不便利性，回收流程繁琐。为了方便大众回收，连接回收企…
阅读更多...
git环境编译升级

git环境编译升级

1. 下载 wget https://www.kernel.org/pub/software/scm/git/git-2.43.0.tar.gz 编译从github下载的压缩包可能会报错，建议从git官网下载压缩包。 https://www.kernel.org/ 2. 编译 tar -zxvf git-2.34.5.tar.gz cd git-2.34.5 make prefix/usr/local all sudo ma…
阅读更多...
【数据结构】树和二叉树——Lesson1

【数据结构】树和二叉树——Lesson1

Hi~！这里是奋斗的小羊，很荣幸您能阅读我的文章，诚请评论指点，欢迎欢迎 ~~ 💥💥个人主页：奋斗的小羊 💥💥所属专栏：C语言 🚀本系列文章为个人学习…
阅读更多...
科普文：微服务技术栈梳理

科普文：微服务技术栈梳理

概叙 如上两图所示，微服务架构下，需要的组件很多，上面中也并未列全。下面将梳理一下国内微服务架构下，用到的技术栈，仅供参考。 科普文：12种常见的软件架构-CSDN博客 没有最好的架构，只有最适…
阅读更多...
Kubernetes RBAC 之 UserAccount

Kubernetes RBAC 之 UserAccount

Kubernetes RBAC 之 UserAccount 定义 Kubernetes 用户账户可以直接在终端使用，用来读写其有权限的资源或者非资源数据，而 Service Account 一般是用来绑定程序调用的。 创建用户 创建 Kubernetes 用户 kate # 进入 Kubernetes Public Key Infrastr…
阅读更多...
常用网络接口自动化测试框架应用

常用网络接口自动化测试框架应用

一、RESTful（resource representational state transfer)类型接口测试 (一）GUI界面测试工具：jmeter 1、添加线程组 2、添加http请求 3、为线程组添加察看结果树 4、写入接口参数并运行 5、在查看结果树窗口查看结果 6、多组数据可增加CSVDat…
阅读更多...
Android 桌面小组件使用

Android 桌面小组件使用

基本步骤 1.创建小组件布局 这里需要注意的事,小组件布局里不能使用自定义View,只能使用原生的组件,比如说LinearLayout,TextView,连约束布局都不能使用 <?xml version"1.0" encoding"utf-8"?> <LinearLayout xmlns:android"http://sch…
阅读更多...
python数据可视化（7）——绘制箱形图

python数据可视化（7）——绘制箱形图

课程学习来源&#xff1a;b站up&#xff1a;【蚂蚁学python】 【课程链接&#xff1a;【【数据可视化】Python数据图表可视化入门到实战】】 【课程资料链接&#xff1a;【链接】】 Python绘制箱形图分析北京天气数据 箱形图 箱形图&#xff08;Box-plot&#xff09;又称为盒…
阅读更多...
一站式短视频矩阵开发，高效托管！

一站式短视频矩阵开发，高效托管！

短视频矩阵系统源码SaaS解决方案提供全面的开发服务&#xff0c;包括可视化视频编辑、矩阵式内容分发托管以及集成的多功能开发支持。 短视频矩阵&#xff1a;引爆您的数字营销革命 短视频矩阵系统是一套多功能集成解决方案&#xff0c;专为提升在短视频平台上的内容创作、管理…
阅读更多...
【C++】多态-最全解析（多态是什么？如何使用多态？多态的底层原理是什么？）

【C++】多态-最全解析（多态是什么？如何使用多态？多态的底层原理是什么？）

目录 一、前言 二、多态是什么&#xff1f; 三、多态的定义及实现 &#x1f525; 多态的构成条件&#x1f525; &#x1f525; 虚函数的重写&#x1f525; &#x1f525;虚函数重写的两个例外 &#x1f525; &#x1f34d; 协变返回类型 &#x1f95d; 析构函数的重写…
阅读更多...
使用APEXSQL LOG解析sql server事务日志，进行审计与数据恢复

使用APEXSQL LOG解析sql server事务日志，进行审计与数据恢复

一 下载 https://download.csdn.net/download/sunke861/11449739 二 使用 解压安装包后&#xff0c;点击&#xff1a;ApexSQLLog.exe 2.1 连接数据库 连接要审计的数据库&#xff1a; 假如报错&#xff1a; 则点击ok关闭该窗口&#xff0c;然后点击左上方的New按钮&#xf…
阅读更多...
Git-Automatic merge failed； fix conflicts and then commit the result. 解决

Git-Automatic merge failed； fix conflicts and then commit the result. 解决

Git-Automatic merge failed; fix conflicts and then commit the result. 解决 文章目录 1. 杂话2. 问题2.1 先搞定版本A2.2 再搞定版本AC2.3 搞定AB版本2.4 冲突 3. 解决3.1 分析3.2 解决 1. 杂话 大伙儿应该都用过Git吧&#xff0c;具体是个啥东西我就不说了哈。之前我在用g…
阅读更多...
Codeforces Round 958 (Div. 2)(A~C)题

Codeforces Round 958 (Div. 2)(A~C)题

A. Split the Multiset 思路: 最优的策略是每次操作分出 k−1&#x1d458;−1 个 1&#xff0c;然后考虑最后是否会剩下一个单独的 1。 代码: #include<bits/stdc.h> using namespace std; #define N 1000005 typedef long long ll; typedef unsigned long long ull;…
阅读更多...
【找不到视图问题解决】@RestController 与 @Controller注解的使用区别

【找不到视图问题解决】@RestController 与 @Controller注解的使用区别

一、问题描述 苍穹外卖在菜品分页查询功能实现的过程中&#xff0c;出现了找不到视图的情况 2024-07-12 21:54:20.860 ERROR 22488 --- [nio-8080-exec-4] o.a.c.c.C.[.[.[/].[dispatcherServlet] : Servlet.service() for servlet [dispatcherServlet] in context with p…
阅读更多...
【Vim】为什么程序员喜欢用 Vim

【Vim】为什么程序员喜欢用 Vim

1. Vim介绍 Vim是一款高度可配置的文本编辑器&#xff0c;它被设计成作为一个工具&#xff0c;可以非常高效地进行文本编辑工作。以下是关于Vim的一些基本介绍&#xff1a; 历史&#xff1a;Vim 是 Vi 文本编辑器的改进版&#xff0c;最初由布莱姆米勒&#xff08;Bram Moole…
阅读更多...
Transformer注意力机制

Transformer注意力机制

Transformer注意力机制 &#x1f42c; 目录: 一、Transformer简介二、理解注意力机制自注意力机制多头注意力机制 一、Transformer简介 Transformer是一种用于自然语言处理(NLP)和其他序列到序列(Seq2Seq)任务的深度学习模型框架&#xff0c;它在2017年由Vaswani等人首次提出…
阅读更多...
手机m4a怎么转换成mp3，手机端即可完成格式转换

手机m4a怎么转换成mp3，手机端即可完成格式转换

M4A&#xff08;MPEG-4 Audio&#xff09;是一种无损压缩的音频格式&#xff0c;通常用于苹果设备和 iTunes 上&#xff0c;因为它能提供较高的音质同时占用较小的存储空间。 然而&#xff0c;MP3 作为最普及的音频格式之一&#xff0c;兼容性更强&#xff0c;几乎所有的播放设…
阅读更多...
【单元测试】SpringBoot

【单元测试】SpringBoot

【单元测试】SpringBoot 1. 为什么单元测试很重要&#xff1f;‼️ 从前&#xff0c;有一个名叫小明的程序员&#xff0c;他非常聪明&#xff0c;但有一个致命的缺点&#xff1a;懒惰。小明的代码写得又快又好&#xff0c;但他总觉得单元测试是一件麻烦事&#xff0c;觉得代码…
阅读更多...
ENSP中NAT的相关实验（两个私网，一个公网）

ENSP中NAT的相关实验（两个私网，一个公网）

题目 实验需求 1.按照图示配置IP地址&#xff0c;公网地址100.1.1.1/24 2.私网A通过NAPT&#xff0c;使R1接入到互联网&#xff0c;私网B通过EASY IP&#xff0c;使R3接入到互联网 3.私网A配置NAT SERVER把Telnet的Telnet服务发布到公网&#xff0c;使PC2可以访问 三、实验…
阅读更多...
el-table和 el-image图片预览使用插槽后层叠样式错乱问题

el-table和 el-image图片预览使用插槽后层叠样式错乱问题

问题&#xff1a; 解决办法&#xff1a;在el-image组件中添加preview-teleported 属性 最终效果
阅读更多...
最新文章

Copyright @ 2022~2023