网站html有趣代码,做暖暖视频网站大全,广州网站排名优化服务,网站程序超市目录
消息队列
产生背景
消息队列介绍
常见的消息队列产品
应用场景 消息队列的消息模型
Kafka的基本介绍
简介
Kafka的架构
Kafka的使用
Kafka的shell命令
Kafka的Python API的操作
完成生产者代码
完成消费者代码 消息队列
产生背景 消息队列:指数据在一个容器…目录
消息队列
产生背景
消息队列介绍
常见的消息队列产品
应用场景 消息队列的消息模型
Kafka的基本介绍
简介
Kafka的架构
Kafka的使用
Kafka的shell命令
Kafka的Python API的操作
完成生产者代码
完成消费者代码 消息队列
产生背景 消息队列:指数据在一个容器中,从容器中一端传递到另一端过程 消息:指的数据,只不过这个这个数据存在一定流动状态 队列:指的容器,可以存储数据,这个容器具备FIFO(先进先出)特性 公共容器的特点: 1.公共性:各个程序都可以与之对接 2.FIFO特性:先进先出 3.具备高效的并发能力:能够承载海量数据 4.具备一定的容错能力:比如支持重新读取消息方案 消息队列介绍
常见的消息队列产品 MQmessage queue消息队列 activeMQ: 出现时期比较早的一款消息队列的中间件产品在早期使用人群是非常多目前整个社区活跃度严重下降使用人群基本很少 rabbitMQ: 此款是目前使用人群比较多的一款消息队列的中间件的产品社区活跃度比较高主要是应用传统业务领域中 rocketMQ: 是阿里推出的一款消息队列的中间件的产品目前主要是在阿里系环境中使用目前支持的客户端比较少主要是Java中应用较多 Kafka: Apache旗下的顶级开源消息,是一款消息队列的中间件产品,项目来源于领英,是大数据体系中目前为止最为常用的一款消息队列产品 应用场景 消息队列的应用场景: 1.应用解耦合 2.异步处理 3.限流削峰 4.消息驱动系统 消息队列的消息模型 在Java中, 为了能够集成消息队列的产品, 专门提供了一个消息队列的协议: JMS(Java Message Server) java消息服务 消息队列中两个角色:生产者(producer)和消费者(consumer) 生产者:生产/发送消息到消息队列中 消费者:从消息队列中获取消息 在JMS规范中,专门规定了两种消息消费类型: 1.点对点消费类型:一条消息最终只能被一个消费所消费,微信聊天的私聊 2.发布订阅消费模型:指一条消息最终被多个消费者所消费,微信聊天的群聊 Kafka的基本介绍
简介
Kafka是一款消息队列中间件产品,来源于领英公司,后期贡献给了Apache,目前是Apache旗下的顶级开源项目,采用语言是Scala Kafka的特点: 1.可靠性:Kafka集群是分布式的,有多副本机制,数据可以自动复制 2.可扩展性:Kafka集群可以灵活的调整,在线扩容 3.耐用性:Kafka数据保存在磁盘上,数据有多副本机制,数据持久化,一定程度上防止数据丢失 4.高性能:Kafka可以存储海量的数据,虽然是使用磁盘进行存储,但是Kafka有各种优化手段(例如磁盘的顺序读写,零拷贝等)提高数据的读写速度(吞吐量) Kafka的架构 1. Kafka中集群节点叫broker,节点与节点之间没有主从之分,地位是完全一样 2.Topic:主题/话题,是业务层面对消息进行分类的 3.一个Topic可以设置多个分区 4.同一个partition分区可以设置多个副本,但是副本数不能超过()集群broker节点的个数 5.broker节点间没有主从之分,但是同一个partition分区的不同副本间有主从之分,分为Leader主副本和Follwer从副本 6.生产者将数据首先发送给到Leader主副本,接着是Leader主副本主动往Follower从副本上同步消息 7.Zookeeper用来管理集群,以及管理元数据信息 8.ISR同步列表,该列表中存放的是与Leader主副本消息同步程度最接近的Follower从副本,也就是消息最小的一个列表,该列表的作用是当Leader主副本无法对外提供服务的时候,会从该ISR列表中选择一个Follower从副本变成Leader主副本,对外提供服务 相关名词 Kafka Cluster : kafka集群 Topic : 主题/话题 Broker : Kafka中的节点 Producer : 生产者,负责生产/发送消息到Kafka中 Consumer : 消费者,负责从Kafka中获取消息 Partition : 分区,一个Topic可以设置多个分区,没有数量限制 Kafka的使用
Kafka的shell命令 Kafka本质上是一个消息队列中间件产品,主要负责消息数据的传递,也就说学习Kafka 也就是学习如何使用Kafka生产数据以及如何使用Kafka来消费数据
创建Topic ./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --create --topic test02 --partitions 4 --replication-factor 2 参数说明: --bootstrap-server:kafka集群中broker连接信息 --create:指定操作类型,这里是新建Topic --topic:指定要新建的Topic名称 --partitions:设置Topic的分区数 --replicattion-factor:设置Topic分区的的副本数 注意:如果副本数超过了集群broker节点个数,会报错 查看Topic ./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --list 参数说明: --bootstrap-server: Kafka集群中broker连接信息 --list: 指定操作类型。这里是查看Kafka集群上所有可用的Topic列表 查看具体Topic ./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --describe --topic test04 参数说明: --bootstrap-server: Kafka集群中broker连接信息 --describe: 指定操作类型。这里是查看具体Topic信息 模拟生产者Producer ./kafka-console-producer.sh --broker-list node1.itcast.cn:9092,node2.itcast.cn:9092 --topic test04 参数说明: --broker-list: Kafka集群中broker连接信息 --topic: 指定要将消息发送到哪个具体的Topic 模拟消费者Consumer ./kafka-console-consumer.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --topic test04 参数说明: --bootstrap-server: Kafka集群中broker连接信息 --topic: 指定要从哪个Topic中消费消息 --from-beginning: 指定该参数以后会从最旧的地方开始消费 latest: 消费者默认从最新的地方开始消费 --max-messages: 最多消费的条数。满足条数后就会自动结束 --group: 指定消费组名称。一个消费者只能属于一个消费组一个消费组里面可以有多个消费者。同一个Topic中的同一条数据只能被同一个消费组中的一个消费者所消费 在工作中的参数一般如何使用 答: 推荐latest、--max-messages、--group一同使用。因为实际企业中Topic的数据量是特别大的消费、打印都需要消耗服务器的资源如果不限定消费的最大条数可能造成服务器宕机。 修改Topic ./kafka-topics.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --alter --topic test01 --partitions 10 分区: 只能增大不能减小。而且没有数量限制 副本: 既不能增大也不能减小 查看消费组中有多少个消费者 ./kafka-consumer-groups.sh --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092 --group g_01 --members --describe Kafka的Python API的操作 准备工作在服务器的节点上安装一个python用于操作Kafka的库 安装命令 python -m pip install kafka-python -i https://pypi.tuna.tsinghua.edu.cn/simple API使用的参考文档: https://kafka-python.readthedocs.io/en/master/usage.html#kafkaproducer 完成生产者代码
import timefrom kafka import KafkaProducer# 同步发送
def sync_send():global topic, partition, offset# 2.1- 同步发送数据/消息metadata producer.send(test01, valuefhello_java_{i}.encode(UTF-8)).get()# metadata producer.send(test03,valuefhello_spark_{i}.encode(UTF-8)).get()# 2.2- 获取元信息中的内容topic metadata.topicpartition metadata.partitionoffset消息偏移量从0开始编号。也就是一条消息在分区中的序号/索引在不同分区间消息偏移量是无序在同一个分区里面消息偏移量是有序offset metadata.offsetprint(f{topic},{partition},{offset},{metadata})if __name__ __main__:# 1- 创建生产者producer KafkaProducer(bootstrap_servers[node1.itcast.cn:9092,node2.itcast.cn:9092])# 2- 发送消息for i in range(10):# 同步发送# sync_send()# 2.3- 异步发送异步发送需要等待一下或者明确关闭Producer生产者producer.send(test01, valuefhello_hive_{i}.encode(UTF-8))time.sleep(1)# 3- 释放资源/关闭生产者# producer.close() 完成消费者代码
from kafka import KafkaConsumerif __name__ __main__:# 1- 创建消费者consumer KafkaConsumer(test01,bootstrap_servers[node1.itcast.cn:9092, node2.itcast.cn:9092])# 2- 消费消息for msg in consumer:topic msg.topicpartition msg.partitionoffset msg.offset# key和value消费出来都是bytes数据类型需要进行解码key msg.keyvalue msg.valueprint(f{topic}{partition}{offset}{key}{value.decode(UTF-8)}{msg})
