简易教程 |
您所在的位置：网站首页 › kafka配置参数详解 › 简易教程 |
简易教程
专注Python、AI、大数据 @七步编程 Kafka简介 Kafka是一款流行分布式消息分布订阅系统，除Kafka之外还有MQ、Redis等。我们可以把消息队列视为一个管道，管道的两端分别是消息生产者(producer)和消息消费者(consumer)，消息生产者产生日志等消息后可以发送到管道中，这时消息队列可以驻留在内存或者磁盘上，直到消费者来把它读走为止。 上述就是Kafka的一个概括，我们只需要了解一下Kafka的架构和一些专业术语即可，下面就来介绍一下Kafka 中一些专业术语。 Producer：消息生产者，负责把产生的消息发送到Kafka服务器上。 Consumer：消息消费者，从Kafka服务器读取消息。 Consumer Group：消费者群组，每个消息消费者可以划分为一个特定的群组。 Topic：这是Kafka使用中非常重要的一个术语，它相当于消息的"身份标识"，消息生产者产生消息时会给它贴上一个Topic标签，当消息消费者需要读取消息时，可以根据这个Topic读取特定的数据。 Broker：Kafka集群中包含的服务器。 Kafka配置下载Kafka 打开链接https://www.apache.org/dyn/closer.cgi?path=/kafka/2.4.0/kafka_2.12-2.4.0.tgz下载kafka的tgz安装包，  然后解压并进入kafka路径下， > tar -xzf kafka_2.12-2.4.0.tgz > cd kafka_2.12-2.4.0这里需要简单的介绍一下ZooKeeper，它是一个开源的分布式协调框架，它可以将那些容易出错的分布式一致性服务封装起来，构成一个高效、可靠、简单易用的结构提供给用户使用。 之所以提到ZooKeeper是因为Kafka就是使用ZooKeeper作为协调框架，因为前面下载的压缩包中已经包含了ZooKeeper，因此不需要在单独去下载它的安装包。 启动服务 使用Kafka之前需要先启动一个ZooKeeper服务，这里直接使用Kafka中包含的脚本即可， > bin/zookeeper-server-start.sh config/zookeeper.properties启动ZooKeeper服务之后再启动Kafka服务， > bin/kafka-server-start.sh config/server.properties需要强调一下，config/server.properties是Kafka的配置文件，可以用于配置监听的host、port、broker等。 默认的ZooKeeper连接服务为localhost:2181， # Zookeeper connection string (see zookeeper docs for details). # This is a comma separated host:port pairs, each corresponding to a zk # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002". # You can also append an optional chroot string to the urls to specify the # root directory for all kafka znodes. zookeeper.connect=localhost:2181另外，producer和consumer的监听端口为9092，如果需要更改server的host和port端口可以通过修改config/server.properties进行配置。 Kafka使用这里主要介绍2种使用方式， 命令行Python需要注意的是，无论是命令行方式还是Python方式，都是建议在前面的基础之上的，也就是说需要首先在集群上启动ZooKeeper和Kafka服务。 命令行 命令行的使用方式相对简单，通过前面的Kafka配置之后可以直接在命令行下进行使用。 1. 创建Topic 使用Kafka，我们首先需要创建一个Topic，这样后续消息生产者和消息消费者才能针对性的发送和消费数据， > bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test这样我们就创建了一个名为test的Topic。 我们也可以通过命令来查看我们已经创建的 Topic， > bin/kafka-topics.sh --list --zookeeper localhost:2181 test2. 发送消息 前面介绍Kafka专业术语中已经阐述，Kafka使用过程中首先需要消息生产者发送消息，那么消费者才可以读取到消息。 启动一个终端A，执行下面命令， > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test >hello world >当执行producer脚本后，会出现消息输入提示符，这是我们可以输入消息(数据)，然后它会发送到对应的服务器(Broker)。 3. 消费消息 现在管道中已经有了数据，接下来我就可以使用消费者去读取数据。 另外启动一个终端B，执行下面命令， > ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning hello world可以看到，消息消费者一直处于监听状态，每当在终端A输入一条消息，终端B也会更新一条消息。 Python-Kafka Python是目前比较热门、流行的一种编程语言，我们在开发过程中有时需要直接在项目中直接调用Kafka的接口，这样使用命令行方式就不够方便了。好在Kafka提供了Python的工具包，我们只需要安装Kafka工具包就可以直接调用它的API进行使用，下面来介绍一下在Python中如何使用Kafka。 1. 安装Kafka工具包 > pip install kafka2. 创建Producer脚本 导入Kafka生产者模块，创建一个Producer脚本， import json from kafka import KafkaProducer data = {"a": 1, "b": 2, "c":3} producer = KafkaProducer(bootstrap_servers=["localhost:9092"], api_version=(0, 10, 0), value_serializer=lambda v: json.dumps(v).encode('utf-8')) for i in range(1000): producer.send("test", data, partition=0)这里data是用于发送到服务器的示例数据，然后调用KafkaProducer接口，分别传入server、api_version等参数。 3. 创建消费者脚本 执行前面Python脚本，Producer就可以发送消息到指定的服务器，下面就可以创建一个Consumer脚本，从服务器订阅消息， from kafka import KafkaConsumer consumer = KafkaConsumer('test', group_id="group2", bootstrap_servers=["localhost:9092"]) for msg in consumer: print(msg.value) # 输出 b'{"a": 1, "b": 2, "c": 3}' b'{"a": 1, "b": 2, "c": 3}' b'{"a": 1, "b": 2, "c": 3}' b'{"a": 1, "b": 2, "c": 3}' b'{"a": 1, "b": 2, "c": 3}' b'{"a": 1, "b": 2, "c": 3}' ......这样就可以输出从消息队列中读取到的数据。 备注Kafka定位为分布式消息发布-订阅系统，提及分布式就可以想象，只有当在多节点环境下才能最大的发挥它的价值，前面所介绍的Kafka配置方式都是基于单节点的配置，由于本文主要是为了系统的梳理一下Kafka的配置及使用，因此对于配置这一块不再花费大的篇幅去详细介绍，如果需要到多节点配置Kafka可以自行查阅其他资料。 干货最近，为了方便大家，我花费了半个月的时间把这几年来收集的各种技术干货整理到一起，其中内容包括但不限于Python、机器学习、深度学习、计算机视觉、推荐系统、Linux、工程化、Java，内容多达5T+，我把各个资源下载链接整理到一个文档内，目录如下：  所有干货送给大家，希望能够点赞支持一下！ https://pan.baidu.com/s/1eks7CUyjbWQ3A7O9cmYljA （提取码：0000） 作品精选Jackpop：目录 | 精选CV、Python等系列教程 Jackpop：大数据处理 | Spark&HDFS集群配置及基本使用 Jackpop：实用工具 | 推荐3款令人惊艳的截图工具 Jackpop：强烈推荐 | 这将会成为一个优质的github项目 Jackpop：计算机小白如何开始机器学习的学习，有入门课程推荐吗? Jackpop：C盘快满了，该如何清理？ |
【本文地址】
今日新闻 |
推荐新闻 |