Kafka安装与配置

1.下载Kafka2.13-3.1.0

最新版为 kafka_2.13-3.1.0.tgz

下载Zookeper

最新版为 zookeeper-3.8.0

2.单机安装zookeper

Kafka依赖于zookeeper,官方承诺将来会移除.

解压文件:

tar zxvf apache-zookeeper-3.8.0-bin.tar.gz -C /opt/
mv /opt/apache-zookeeper-3.8.0-bin/ /opt/zookeeper

在/opt/zookeeper/ 目录下创建数据文件目录和日志文件目录

mkdir /opt/zookeeper/zkData
mkdir /opt/zookeeper/zkLog

# 复制一份配置文件并修改

cd /opt/zookeeper/conf/
cp zoo_sample.cfg zoo.cfg
vi zoo.cfg

# 修改如下内容
dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog

启动

cd /opt/zookeeper/bin/
# 启动zookeeper
./zkServer.sh start
# 查看进程是否启动
jps
# 查看状态
./zkServer.sh status
# 停止zookeeper
./zkServer.sh stop

3.安装Kafka

解压到指定目录

cd /home
$ tar -xzf kafka_2.13-3.1.0.tgz
$ cd kafka_2.13-3.1.0

修改config目录下vi server.propertie文件

listeners = PLAINTEXT://192.168.2.40:9092

#多个可用逗号分隔

#zookeeper.connect=server1:2181,server2:2181,server3:2181

zookeeper.connect=localhost:2181

启动命令:

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

此方式可以实时查看日志.

后台启动方式:

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

查询进程和关闭命令

jps
./kafka-server-stop.sh

登录zookeeper客户端，查看/brokers/ids

cd /opt/zookeeper/bin/
zkCli.sh
# 查询结果如下：
[zk: localhost:2181(CONNECTED) 0] ls /brokers/ids
[0]
[zk: localhost:2181(CONNECTED) 1] quit

kafka常见命令

#创建主题 主题名是 quickstart-events
$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092
#查询主题
$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
#主题中写入消息
 bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
This is my first event
This is my second event
#主题中读取消息
 bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

kafka集群

假如现在有两台服务器192.168.2.40,192.168.2.41

kafka的安装与配置如上,两台服务器唯一不同的地方就是配置文件中的broker.id

 修改config目录下vi server.propertie文件

192.168.2.40

broker.id=0

192.168.2.41

broker.id=1

bin目录启动命令都添加

vi kafka-server-start.sh
#添加  export JMX_PORT="9999"
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export JMX_PORT="9999"
fi

 登录zookeeper客户端，查看/brokers/ids

4.可视化工具kafka-eagle

下载:kafka-eaglev2.1.0.tar.gz

解压

cd /home
tar -zxvf kafka-eagle-web-2.1.0-bin.tar.gz 

在/etc/profile文件中添加环境变量KE_HOME 

vi /etc/profile
# 在profile文件中添加
export KE_HOME=/mydata/kafka/kafka-eagle-web-2.0.5
export PATH=$PATH:$KE_HOME/bin
# 使修改后的profile文件生效
. /etc/profile

安装MySQL并添加数据库ke，kafka-eagle之后会用到它；
修改配置文件$KE_HOME/conf/system-config.properties，主要是修改Zookeeper的配置和数据库配置，注释掉sqlite配置，改为使用MySQL.

######################################
kafka.eagle.zk.cluster.alias=cluster1
cluster1.zk.list=localhost:2181
 ######################################
# kafka eagle webui port
######################################
kafka.eagle.webui.port=8048
 ######################################
# kafka mysql jdbc driver address
######################################
kafka.eagle.driver=com.mysql.cj.jdbc.Driver
kafka.eagle.url=jdbc:mysql://localhost:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
kafka.eagle.username=root
kafka.eagle.password=123456

 启动后会自动使用上面的数据库连接，创建并初始化数据库.名称ke.

# 停止服务
bin/ke.sh stop
# 重启服务
bin/ke.sh restart
# 查看服务运行状态
bin/ke.sh status
# 查看服务状态
bin/ke.sh stats
# 动态查看服务输出日志
tail -f /logs/ke_console.out 

 ​
启动成功可以直接访问，输入账号密码admin:123456，访问地址：http://localhost:8048/

注意观察 brokers,topics的数量。brokers为0的话没有连接成功.

可视化工具自然少不了监控，如果你想开启kafka-eagle对Kafka的监控功能的话，需要修改Kafka的启动脚本，暴露JMX的端口.

vi kafka-server-start.sh
#添加  export JMX_PORT="9999"
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
    export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    export JMX_PORT="9999"
fi

kafka集群图示:

5.SpringBoot整合Kafka.

在pom.xml中添加

    org.springframework.kafka
    spring-kafka

 在application.yml中spring节点下添加

spring
    kafka:
    bootstrap-servers: 192.168.2.40:9092
    producer: # 生产者配置
      retries: 3 # 设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384 #16K
      buffer-memory: 33554432 #32M
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: zhTestGroup # 消费者组
      enable-auto-commit: false # 关闭自动提交
      auto-offset-reset: earliest # 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    listener:
      # 当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # RECORD
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
      # BATCH
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
      # TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
      # COUNT
      # TIME |　COUNT　有一个条件满足时提交
      # COUNT_TIME
      # 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # MANUAL
      # 手动调用Acknowledgment.acknowledge()后立即提交，一般使用这种
      # MANUAL_IMMEDIATE
      ack-mode: manual_immediate

生产者:

@RestController
public class KafkaProducer {

	@Autowired
	private KafkaTemplate kafkaTemplate;
	// 发送消息
	@GetMapping("/kafka/normal/{message}")
	public void sendMessage1(@PathVariable("message") String normalMessage) {
		kafkaTemplate.send("quickstart-events", normalMessage);
	}

}

 消费者:

@Component
public class KafkaConsumer {
	// 消费监听
	@KafkaListener(topics = {"quickstart-events"})
	public void onMessage1(ConsumerRecord record){
		// 消费的哪个topic、partition的消息,打印出消息内容
		System.out.println("简单消费："+record.topic()+"-"+record.partition()+"-"+record.value());
	}
}

 可以用postman进行测试，观察结果.