zookeeper集群 +kafka集群

1.zookeeper

kafka3.0之前依赖于zookeeper

zookeeper是一个开源,分布式的架构,提供协调服务(Apache项目)

基于观察者模式涉及的分布式服务管理架构

存储和管理数据,分布式节点上的服务接受观察者的注册,一旦分布式节点上的数据如果发生变化,由zookeeper来负责通知分布式节点上的服务

zookeeper分为领导者,追随者 leader follower组成的集群

只要有一半以上的集群存活,zookeeper集群就可以正常工作,适用于安装奇数台的服务集群

全局数据一致,每个zookeeper每个节点都保存相同的数据,维护监控服务的数据一致。(主要作用)

数据更新的原子性,要么都成功,要么都失败

实时性,只要有变化,立刻同步。

zookeeper的应用场景

1.统一命名服务,在分布式的环境下,对所有的应用和服务进行统一命名

2.统一配置管理,配置文件同步,kafka的配置文件被修改,可以快速同步到其他节点

3.统一集群管理,实时掌握所有节点的状态

4.服务器动态上下限

5.负载均衡,把访问的服务器的数据,发送到访问最少的服务器处理客户端的请求

领导者和追随者:zookeeper的选举机制

三台服务器:A B C

A 先启动 发起第一选举,投票投给自己,只有一票,不满半数,A的状态是looking

B 启动 在发起一次选举,A和B分别投自己一票,交换选票信息,myid A发现B的myid比A的大,A的这票会转而投给B,A 0 B 2 没有半数以上结果,A B 会进入looking 、B有可能leader

C 启动 MYID C的myid最大 A和B都会把票投给C A 0 B 0 C 3

C的状态变为leader A B变为follower

只有leader确定,后续的服务器都是追随者

只有两种情况会开启选举机制:

1.初始化的情况会产生选举

2.服务器之间和eader丢失了连接状态

特殊情况下:

leader已经存在,建立连接即可

leader不存在,leader不存在

1.服务器ID大的胜出

2.EPOCH大,直接胜出

3.EPOCH相同,事务ID大的胜出

EPOCH每个leader任期的代号,没有leader,大家的逻辑地址相同。每投完一次之后,数据时递增

事务id表示服务器的每一次更新,每变更一次id变化一次‘

服务器ID

zookeeper 当中所有机器。每台机器不重复,和mysql保存一直

zookeeper+kafka(2.7.0)

kafka (3.4.1)

实现

zookeeper集群

192.168.233.10 zookeeper+kafka

192.168.233.20 zookeeper+kafka

192.168.233.30 zookeeper+kafka

所有

systemctl stop firewalld

setenforce 0

cd /opt

拖进去 apache-zookeeper-3.5.7-bin.tar.gz kafka_2.13-2.7.0

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel

java -version

cd /opt

tar -xf apache-zookeeper-3.5.7-bin.tar.gz

mv apache-zookeeper-3.5.7-bin zookeeper

cd zookeeper

cd /opt/zookeeper/conf

cp zoo_sample.cfg zoo.cfg

vim zoo.cfg

tickTime=2000

#服务器与客户端之间心跳时间,2秒检测一次服务器和客户端之间的通信

initLimit=10

#领导者和追随者之间,最对几次心跳数超时 10*2S 20S

syncLimit=5

#同步超时时间,领导者和追随者之间,同步通信超时的时间。5*2s leader会认为follower丢失,移除集群

16行

dataDir=/opt/zookeeper/data

dataLogDir=/opt/zookeeper/logs (需要改)

添加

server.1=192.168.233.10:3188:3288

server.2=192.168.233.20:3188:3288

server.3=192.168.233.30:3188:3288

##

server.1=192.168.233.10:3188:3288

1 定义每个zookeeper集群的初始myid

192.168.233.10 :服务器的ip地址

3188:领导者和追随者之间交换信息的端口(内部通信的端口)

3288:一旦leader丢失响应,开启选举,3288就是用来执行选举时的通信端口

##

wq!

mkdir /opt/zookeeper/data

mkdir /opt/zookeeper/logs

三台分别创建

10:

cd ..

cd data/

echo 1 > myid

20:

cd ..

cd data/

echo 2 > myid

30:

cd ..

cd data/

echo 3 > myid

所有

vim /etc/init.d/zookeeper

#!/bin/bash

#chkconfig:2345 20 90

#description:Zookeeper Service Control Script

ZK_HOME='/opt/zookeeper'

case $1 in

start)

echo "---------- zookeeper 启动 ------------"

$ZK_HOME/bin/zkServer.sh start

;;

stop)

echo "---------- zookeeper 停止 ------------"

$ZK_HOME/bin/zkServer.sh stop

;;

restart)

echo "---------- zookeeper 重启 ------------"

$ZK_HOME/bin/zkServer.sh restart

;;

status)

echo "---------- zookeeper 状态 ------------"

$ZK_HOME/bin/zkServer.sh status

;;

*)

echo "Usage: $0 {start|stop|restart|status}"

esac

wq!

三台服务器分别启动(一个一个起)

chmod +x /etc/init.d/zookeeper

chkconfig --add zookeeper

service zookeeper start

起完以后看状态

service zookeeper status

2.kafka

为什么要引入消息队列(MQ),首先他也是一个中间件。在高并发环境下,同步请求太多来不及处理,来不及处理的请求会阻塞,比方数据库就会形成行锁或者表锁,请求线程满了,超标了,too many connection。整个系统雪崩

消息队列的作用:异步处理请求 流量削峰,应用解耦 可恢复性 缓冲机制

解耦:

耦合:在软件系统当中,修改一个组件需要修改所有其他组件,高度耦合

低度耦合:修改其中一个组件,对其他最贱影响不大,无需修改所有

A B C

只有通信保证,其他的修改不影响整个集群,每个组件可以独立的扩展,修改,降低组件之间的依赖性

依赖点就是接口约束,通过不同的端口,保证集群通信

可恢复性:系统当中的有一部分组件消失,不影响整个系统,也就是在消息队列当中,即使有一个处理消息的进程失败,一旦恢复还可以重新加入到队列当中,继续处理消息。

缓冲机制:可以控制和优化数据经过系统的时间和速度,解决生产消息和消费消息处理速度不一致的问题

峰值的处理能力:消息队列在峰值情况之下,能够顶住突发的访问压力。避免专门为了突然情况而对系统进行修改

异步通信:允许用户把一个消息放入队列,但是不立即处理,等我想处理的时候在处理

消息队列的模式:

点对点 一对一:消息的生产者发送消息到队列中,消费者从队列中提取消息,消费者提取完之后,队列中被提取的消息将会被移除,后续消费者不能再消费队列当中的消息,消息队列可以有多个消费者,但是一个消息,只能由一个消费者提取

RABBITMQ

发布/订阅模式:一对多,又叫做观察者模式。消费者提取数据之后,队列当中的消息不会被清除

生产者发布一个消息到对象(主题)所有消费者都是通过主题获取消费之后,队列当中的消息不会消除

主题:topic topic类似于一个数据流的管道,生产者把消息发布到主题,消息从主题当中订阅数据,主题可以分区,每个分区都要自己的偏移量

分区:partition 每个主题都可以分成多个分区,每个分区是数据的有序子集,分区可以运行kafka进行水平扩展,以处理大量数据

消息在分钟按照偏移量存储,消费者可以独立每个分区的数据

偏移量:是每个消息在分区中唯一的标识,消费者可以通过偏移量来跟踪获取已读或者未读消息的位置,也可以提交偏移量来记录已处理的信息

生产者:producer 生产者把数据发送kafka的主题当中 负责写入消息

消费者:consumer从主题当中读取数据,消费者可以是一个也可以是多个,每个消费者有一个唯一的消费者组ID,kafka实现负载均衡和容错性

经纪人:Broker每个kafka节点都有一个Broker,每个Broker负责一台kafka服务器,id唯一,存储主题分区当中的数据,处理生产和消费者的请求,维护元数据(zookeeper)

zookeeper:zookeeper负责保存元数据,元数据就是topic的相关信息(发布在哪台主机上,指定了多少分区,以及副本数,偏移量)

zookeeper自建一个主题:_consumer_offsets,

3.0之后不依赖zookeeper的核心 元数据由kafka节点自己管理

消费的方式:

begining,从头开始

实时更新

指定位置,用代码编写

kafka的工作流程:

生产者向主题里面发送数据,主题里的分区保存数据,消费者根据消费方式来消费数据。

生产者写入的topic的数据时持久化,默认7个小时

至少一次语义:只要消费者进入,确保消息至少被消费一次。

实验

所有

cd /opt

tar -xf kafka_2.13-2.7.0.tgz

mv kafka_2.13-2.7.0 kafka

vim /etc/profile

export KAFKA_HOME=/opt/kafka

export PATH=$PATH:$KAFKA_HOME/bin

source /etc/profile

cd /opt/kafka/config

cp server.properties server.properties.bak

10:

vim server.properties

21行:

65行 修改

130行 修改

zookeeper.connect=192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

配置zookeeper的集群

20:

所有:

vim /etc/init.d/kafka

#!/bin/bash

#chkconfig:2345 22 88

#description:Kafka Service Control Script

KAFKA_HOME='/opt/kafka'

case $1 in

start)

echo "---------- Kafka 启动 ------------"

${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties

;;

stop)

echo "---------- Kafka 停止 ------------"

${KAFKA_HOME}/bin/kafka-server-stop.sh

;;

restart)

$0 stop

$0 start

;;

status)

echo "---------- Kafka 状态 ------------"

count=$(ps -ef | grep kafka | egrep -cv "grep|$$")

if [ "$count" -eq 0 ];then

echo "kafka is not running"

else

echo "kafka is running"

fi

;;

*)

echo "Usage: $0 {start|stop|restart|status}"

esac

wq!

chmod +x /etc/init.d/kafka

chkconfig --add kafka

service kafka start

netstat -antp | grep 9092

所有:

cd /opt/kafka/bin

20:

kafka-topics.sh --create --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --replication-factor 2 --partitions 3 --topic test1

kafka-topics.sh --create --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --replication-factor 2 --partitions 3 --topic test1

##

创建主题

1.在kafka的bin目录下创建,是所有kafka可执行命令的文件

2. --zookeeper指定的是zookeeper的地址和端口,保存kafka的元数据

3. --replication-factor 2 定义每个分区的副本数

4.partition 3 指定主题的分区数

5. --topic test1 指定主题的名称

10:

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181

kafka-topics.sh --describe --zookeeper 192.168.233.10:2181,192.168.233.20:2181,192.168.233.30:2181 --topic test1

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181

kafka-topics.sh --describe --zookeeper 192.168.66.15:2181,192.168.66.16:2181,192.168.66.17:2181 --topic test3

所有

vim /etc/hosts

192.168.233.10 test1

192.168.233.20 test2

192.168.233.30 test3

记得改主机名

20:

kafka-console-producer.sh --broker-list 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-producer.sh --broker-list 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

30:

从头开始

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3 --from-beginning

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1 --from-beginning

实时消息

kafka-console-consumer.sh --bootstrap-server 192.168.66.15:9092,192.168.66.16:9092,192.168.66.17:9092 --topic test3

kafka-console-consumer.sh --bootstrap-server 192.168.233.10:9092,192.168.233.20:9092,192.168.233.30:9092 --topic test1

20:创建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi1

kafka-topics.sh --create --zookeeper 192.168.66.16:2181 --partitions 1 --replication-factor 1 --topic guoqi1

30:创建

kafka-topics.sh --create --zookeeper 192.168.233.20:2181 --partitions 1 --replication-factor 1 --topic guoqi2

kafka-topics.sh --create --zookeeper 192.168.66.17:2181 --partitions 1 --replication-factor 1 --topic guoqi2

10:连接

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.233.20:9092 --topic guoqi2

kafka-console-consumer.sh --bootstrap-server 192.168.66.16:9092 --topic guoqi1

kafka-console-consumer.sh --bootstrap-server 192.168.66.17:9092 --topic guoqi2

20:发送

kafka-console-producer.sh --broker-list 192.168.233.20:9092 --topic guoqi1

kafka-console-producer.sh --broker-list 192.168.66.16:9092 --topic guoqi1

30:发送

kafka-console-producer.sh --broker-list 192.168.233.30:9092 --topic guoqi2

kafka-console-producer.sh --broker-list 192.168.66.17:9092 --topic guoqi2

如何修改分区数

20:kafka-topics.sh --zookeeper 192.168.233.20:2181 --alter --topic guoqi1 --partitons 3

删除

kafka-topics.sh --delete --zookeeper 192.168.233.20:2181 --topic guoqi1

查看元数据

cd /opt/zookeeper/bin

./zkCli.sh -server 192.168.233.10:2181

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

【Linux】冯诺依曼体系结构(硬件)、操作系统(软件)、系统调用和库函数 --- 概念篇

👦个人主页:Weraphael ✍🏻作者简介:目前正在学习c和Linux还有算法 ✈️专栏:Linux 🐋 希望大家多多支持,咱一起进步!😁 如果文章有啥瑕疵,希望大佬指点一二 …

Vcenter 6.7 VCSA证书过期问题处理

1. 故障现象 2022年10月25日,登陆VC报错。 按照报错信息,结合官方文档,判断为STS证书过期导致。 vCenter Server Appliance (VCSA) 6.5.x, 6.7.x or vCenter Server 7.0.x 在/var/log/vmware/vpxd-svcs/vpxd-svcs.log看到类似报错: ERRO…

深度探索 Python Pyramid 框架

更多资料获取 📚 个人网站:ipengtao.com Pyramid是一个灵活且强大的Python web框架,广泛用于构建各种规模的Web应用程序。本文将深度探索Pyramid框架,介绍其核心概念、应用场景以及一些高级特性。 安装与基础用法 首先&#xf…

Python if else条件语句详解

if 分支使用布尔表达式或布尔值作为分支条件来进行分支控制。Python 的 if 分支既可作为语句使用,也可作为表达式使用。下面先介绍 if 分支作为语句使用的情形。 if 语句可使用任意表达式作为分支条件来进行分支控制。Python 的 if 语句有如下三种形式:…

通过仿真理解信道化接收机分析过程

概要 信道化从子信道带宽划分上可分为临界抽取和非临界抽取两种,从各子信道中心频率布局上可分为偶型排列和奇型排列,从处理流程上可分为信道化分析与信道化综合过程。本文主要通过仿真来理解偶型排列/临界抽取/信道化分析过程。 基本原理 常规的数字…

maven生命周期回顾

目录 文章目录 **目录**两种最常用打包方法:生命周期: 两种最常用打包方法: 1.先 clean,然后 package2.先 clean,然后install 生命周期: 根据maven生命周期,当你执行mvn install时&#xff0c…

Python中字符串拼接及其应用场景

更多资料获取 📚 个人网站:ipengtao.com 字符串拼接是Python中常见而重要的操作,它涉及到将多个字符串连接成一个字符串。本文将深入探讨Python中字符串拼接的不同方式、性能比较、以及在实际应用中的场景和最佳实践。 常见的字符串拼接方法…

【Pytorch使用自制数据集,Dataloader】

数据集结构 话不多说,直接上核心代码 myDataset.py from collections import Counter from torch.utils.data import Dataset import os from PIL import Imageclass MyDataset(Dataset):"""读取自制的数据集args:- image_dir: 图片的地址- labe…

腾讯云轻量应用服务器怎么安装BT宝塔面板?

腾讯云轻量应用服务器宝塔面板怎么用?轻量应用服务器如何安装宝塔面板?在镜像中选择宝塔Linux面板腾讯云专享版,在轻量服务器防火墙中开启8888端口号,然后远程连接到轻量服务器执行宝塔面板账号密码查询命令,最后登录和…

数据结构 | 查漏补缺之哈希表、最短路径、二叉树与森林的转换

哈希表是什么? 或者说 设图采用邻接表的存储结构,写对图的删除顶点和删除边的算法步骤 删除边 删除点 最短路径问题 判断一个有向图是否有环(回路)可以用下列哪些办法() 存在拓扑序列,就有回路…

JVM之基本概念(一)

(1) 基本概念: JVM 是可运行 Java 代码的假想计算机 ,包括一套字节码指令集、一组寄存器、一个栈、一个垃圾回收,堆 和 一个存储方法域。JVM 是运行在操作系统之上的,它与硬件没有直接的交互。 (2) 运行过程: 我们都…

关于PDE频率的问题

讨论:关于PDE频率的问题 关于PDE频率的问题问题复现讨论解决方法 关于PDE频率的问题 问题 在这篇文章下有人提出一个问题: 在使用物理信息神经网络对固定求解区域内进行物理场的预测,具体来说是在求解传热问题。在定义域为1或者大于0.4的正…

Python Opencv实践 - 简单的AR项目

这个简单的AR项目效果是,通过给定一张静态图片作为要视频中要替换的目标物品,当在视频中检测到图片中的物体时,通过单应矩阵做投影,将视频中的物体替换成一段视频播放。这个项目的所有素材来自自己的手机拍的视频。 静态图片&…

手机电脑同步的时间管理工具

有不少上班族会发现自己有太多的工作要完成,并且在工作中往往会浪费很多时间在无关紧要的事情上,而不是专注于真正重要的任务,因此没有足够的时间来完成所有任务。在这种情况下,我们可以使用时间管理软件来帮助自己优先考虑重要的…

Flink Flink数据写入Kafka

一、环境准备 flink 1.14写入Kafka&#xff0c;首先在pom.xml文件中导入相关依赖 <properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>1.14.6</flink.version><spark.version>2.4.3</spa…

鸿蒙一出,android开发处境再受重创

华为宣布其自研操作系统鸿蒙HarmonyOSNEXT开发者预览版将不再兼容安卓系统&#xff0c;这一消息引起了广泛关注和热议。这一决策标志着华为正式告别安卓&#xff0c;摆脱了外部的制约&#xff0c;开始着手打造一个全新的生态系统。 鸿蒙系统4发布一个月&#xff0c;截至目前&a…

Java LeetCode篇-深入了解二叉树经典解法(三种方式实现:获取二叉树的最大深度)

&#x1f525;博客主页&#xff1a; 【小扳_-CSDN博客】 ❤感谢大家点赞&#x1f44d;收藏⭐评论✍ 文章目录 1.0 对称二叉树 1.1 判断对称二叉树实现思路 1.2 代码实现&#xff1a;判断对称二叉树 2.0 二叉树的最大深度 2.1 使用递归实现获取二叉树的最大深度思路 2.2 代码实…

书-用数组给已知数组插入某个元素(3)

#include<stdio.h> int main(){int i ;int b[8]{99,65,87,23,44,55};//在87前面插入一个数int n6 ;// 长度for (in;i>3;i--)b[i]b[i-1]; //解释&#xff1a;原来23是第三个位置&#xff0c;但是插入数字后变成了第四个位置//就是把b[4]赋值给b[3];b[2]1000;nn1;//一定…

深入理解MySQL事务隔离级别与锁机制

表锁&#xff1a; 行锁&#xff1a; InnoDB和MyISAM最大的不同有两点&#xff1a; InnoDB支持事务&#xff08;TRANSACTION&#xff09;InnoDB支持行锁 MyISAM在执行查询SELECT前&#xff0c;会自动给涉及的所有表加读锁&#xff0c;在执行update、insert、delete操作会自动…

【多线程】线程的三种常见创建方式

文章目录 线程创建方式1——Thread线程创建方式2——Runnable线程创建方式2——匿名内部类线程创建方式3——Callable、FutureTask,带返回值 线程其实是程序中的一条执行路径。 那怎样的程序才是多线程程序呢&#xff1f; 例如12306网站就是支持多线程的&#xff0c;因为同时可…