Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。官网下载地址：http://kafka.apache.org/downloads

安装

下载地址

wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.6.0/kafka_2.13-2.6.0.tgz
tar -zxvf kafka_2.13-2.6.0.tgz
mv kafka_2.13-2.6.0 /usr/local/kafka
cd /usr/local/kafka

启动

    Kafka使用ZooKeeper，所以您需要先启动一个ZooKeeper服务器。如果没有安装。您可以使用随Kafka一起打包的便捷脚本来获取一个快速但是比较粗糙的单节点ZooKeeper实例。

1、启动ZooKeeper

    查看ZooKeeper配置

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# the directory where the snapshot is stored.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181 ########## zookeeper的默认端口 2181，在后面会用到
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080

    启动命令：

bin/zookeeper-server-start.sh config/zookeeper.properties

或者后台静默启动：

bin/zookeeper-server-start.sh -daemon  config/zookeeper.properties

2、Kafka基本配置

    Kafka在config目录下提供了一个基本的配置文件。为了保证可以远程访问Kafka，我们需要修改两处配置。

        打开config/server.properties文件，在很靠前的位置有listeners和 advertised.listeners两处配置的注释，去掉这两个注释，并且根据当前服务器的IP修改如下：

vim config/server.properties

listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://127.0.0.1:9092 #你需要修改为外网或局域网可以访问到的服务器IP

3、启动Kafka

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

或者使用后台启动

bin/kafka-server-start.sh -daemon config/server.properties

这里碰到几个问题，我的云服务器只有1G内存：

java内存不足的问题

vim jvm.options
#修改为
-Xms512m
-Xmx512m

kafaka启动时日式内存不足

vim bin/kafka-server-start.sh
#修改为
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx512m -Xms512m"
fi

3、创建 Topic

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

查看 topic 列表

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

查看描述 topics 信息

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

返回信息说明：

    第一行给出了所有分区的摘要，每个附加行给出了关于一个分区的信息。 由于我们只有一个分区，所以只有一行。

    “Leader”: 是负责给定分区的所有读取和写入的节点。 每个节点将成为分区随机选择部分的领导者。

    “Replicas”: 是复制此分区日志的节点列表，无论它们是否是领导者，或者即使他们当前处于活动状态。

    “Isr”: 是一组“同步”副本。这是复制品列表的子集，当前活着并被引导到领导者。

4、启动生产者

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

5、启动消费者

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

--from-beginning 表示从起始位读取

测试

在生产者窗口输入，可以在消费者端查看输入的内容

生产端：

消费端：

特别提示：

一定要先启动ZooKeeper 再启动Kafka 顺序不可以改变。

先关闭kafka ，再关闭zookeeper。

本文地址：https://www.stayed.cn/item/251

转载请注明出处。

本站部分内容来源于网络,如侵犯到您的权益,请 联系我