SpringBoot -- Kafka(一) 环境搭建

Kafka

分布式发布-订阅类型的MQ

高吞吐量

消息无序，同一个 partition中消息有序

版本号 kafka_2.11-0.10.1.1.tgz： 2.11为scala版本号，0.10.1.1 为kafka版本

准备好JVM

概念

Broker : Kafka服务器，一个服务器被称为一个Broker

Topic : 每一类消息可以定义一个Topic

Partition : 每个Topic都有1个或者多个partition，属于物理上的分隔

offset : 偏移量每个partition中的消息唯一标识

Producer : 消息发布者

Consumer ： 消息订阅者

Consumer Group : 属于订阅者独有的概念，默认为统一的group

更多参考 官网

CentOS下安装Kafka

apache.org项目列表

安装Zookeeper

使用外部zookeeper，不使用kafka自带的zookeeper，使用wget 命令进行下载

使用wget命令下载

使用tar 进行解压

使用默认port： 2181

wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.9/zookeeper-3.4.9.tar.gz[/code] 
tar -zxvf zookeeper-3.4.9.tar.gz


配置 zoo.cfg

cp zoo_sample.cfg zoo.cfg
vim zoo.cfg


配置 dataDir、dataLogDir

dataDir=/data/www/data/zookeeper
dataLogDir = /data/www/logs/zookeeper


启动zk

./zkServer.sh ../conf/zoo.cfg


安装Kafka
使用wget命令下载

使用tar 进行解压

使用默认port： 9092

修改位于config下的 server.properties

使用默认端口号：9092

对外网开启服务： advertised.host.name = xxx.xx.xxx.xx (服务器外网地址)

xxx.xx.xxx.xx (服务器外网地址),我这里服务器在远端所以需要配置

log.dirs： log.dirs=/tmp/kafka-logs

partitions：num.partitions=1

zookeeper

zookeeper.connect=localhost:2181 (zk和kafka我放在了同一台服务器上)

zookeeper.connection.timeout.ms=6000

启动kafka

# &为后台运行
./kafka-server-start.sh ../config/server.properties &


测试
创建Topic

启动producer

启用consumer

创建Topic

./kafka-topic.sh --create --zookeeper localhost:2181 --relication-factor 1 --partitions 1 --topic testTopic


查看创建的 Topic

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




启动producer

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


启用consumer

./kafka-console-consumer.sh --zookeeper localhost:2181 --topic testTopic --from-beginning


发送消息

producer



consumer



可能的错误
当kafka服务与项目在不同服务器上时，可能会出现连接错误，这个时候需要绑定kafka的host

host.name=xx.xx.xxxx.xx

hostname 为内网地址

代码
代码请移步 Github参考地址
如有疑问请加公众号（K171），如果觉得对您有帮助请 github start