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]1 1
tar -zxvf zookeeper-3.4.9.tar.gz
1
1

配置 zoo.cfg
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
1
2
1
2

配置 dataDir、dataLogDir
dataDir=/data/www/data/zookeeper
dataLogDir = /data/www/logs/zookeeper
1
2
1
2

启动zk
./zkServer.sh ../conf/zoo.cfg
zkServer.sh start
1
1

安装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=1zookeeper 

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

zookeeper.connection.timeout.ms=6000

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

测试
创建Topic
启动producer
启用consumer

创建Topic
./kafka-topic.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic
1
1

查看创建的 Topic
./kafka-topics.sh --list --zookeeper localhost:2181
1
1



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

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

发送消息

producer 



consumer




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

host.name=xx.xx.xxxx.xx
hostname 为内网地址


代码

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