kafka maven 依赖_kafka安装及与springboot集成

docker 安装kafka

1、下载镜像

这里使用了wurstmeister/kafka和wurstmeister/zookeeper这两个版本的镜像

docker pull wurstmeister/zookeeper

docker pull wurstmeister/kafka

在命令中运行docker images验证两个镜像已经安装完毕

2.启动

启动zookeeper容器

docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper

启动kafka容器

docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --env KAFKA_ADVERTISED_HOST_NAME=192.168.59.101 --env KAFKA_ADVERTISED_PORT=9092 --volume /etc/localtime:/etc/localtime wurstmeister/kafka:latest

192.168.59.101 改为宿主机器的IP地址，如果不这么设置，可能会导致在别的机器上访问不到kafka。

3. 测试kafka

进入kafka容器的命令行

运行 docker ps，找到kafka的 CONTAINER ID，运行 docker exec -it ${CONTAINER ID} /bin/bash，进入kafka容器。

进入kafka默认目录 /opt/kafka_2.11-0.10.1.0

kafka命令行工具使用

Docker的服务起来之后，我们就可以测试服务，这里我们可以使用docker自带的cli工具进行基本的测试，后续有机会我们再使用API调用方式测试服务。

首先我们需要检查已经启动的容器

1. [root@somehost kafka]# docker ps -a
2. CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
3. 2c3e1aacdd56 wurstmeister/kafka "start-kafka.sh" 2 hours ago Up 2 minutes 0.0.0.0:9092->9092/tcp kafka_kafka_1
4. 20cd04496a9c wurstmeister/zookeeper "/bin/sh -c '/usr/sb…" 2 days ago Up 2 minutes 22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp kafka_zookeeper_1
5. [root@somehost kafka]#

从上面可以看到Kafka的container id为2c3e1aacdd56，我们可以采用下述命令行的方式进入该容器中：

docker exec -it {container id} /bin/bash

Apache Kafka命令行工具(Command Line Interface,CLI)，下文简称CLI。

1. 启动Kafka

启动Kafka需要两步：

1.1. 启动ZooKeeper

[root@Server1 kafka_2.12-0.11.0.0]# bin/zookeeper-server-start.sh config/zookeeper.properties

1.2. 启动Kafka Server

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-server-start.sh config/server.properties

2. 列出Topic

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --zookeeper localhost:2181 --list

HelloWorld

3. 创建Topic

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Demo1

Created topic "Demo1".

上述命令会创建一个名为Demo1的Topic，并指定了replication-factor和partitions分别为1。其中replication-factor控制一个Message会被写到多少台服务器上，因此这个值必须小于或者

等于Broker的数量。

4. 描述Topic

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Demo1

Topic:Demo1 PartitionCount:1 ReplicationFactor:1 Configs:

Topic: Demo1 Partition: 0 Leader: 0 Replicas: 0 Isr: 0

5. 发布消息到指定的Topic

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Demo1

>this

>is

>the

>firest

>input

可以在控制台逐行输入任意消息。命令的终止符是：control + C组合键。

6. 消费指定Topic上的消息

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic Demo1

this

is

the

firest

input

7. 修改Topic

7.1 增加指定Topic的partition，在第3步中创建的Demo1的partition是1。如下命令将增加10个partition

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --partitions 11 --topic Demo1

WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected

Adding partitions succeeded!

7.2. 删除指定Topic

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic Demo1

Topic Demo1 is marked for deletion.

Note: This will have no impact if delete.topic.enable is not set to true.

Note中指出该Topic并没有真正的删除，如果真删除，需要把server.properties中的delete.topic.enable置为true

7.3 给指定的Topic增加配置项，如给一个增加max message size值为128000

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-topics.sh --alter --zookeeper localhost:2181 --topic Demo1 --config max.message.bytes=128000

WARNING: Altering topic configuration from this script has been deprecated and may be removed in future releases.

Going forward, please use kafka-configs.sh for this functionality

Updated config for topic "Demo1".

warning中指出该命令已经过期，将来可能被删除，替代的命令是使用kafka-config.sh。新命令如下：

[root@Server1 kafka_2.12-0.11.0.0]# bin/kafka-configs.sh --alter --zookeeper localhost:2181 --entity-type topics --entity-name Demo1 --add-config max.message.bytes=12800

Completed Updating config for entity: topic 'Demo1'.

需要使用entity-type置为topics，并在entity-name中指定对应的名称

springboot 与 kafka集成

kafka的maven依赖

org.springframework.kafka

spring-kafka

application.yml配置

spring:

kafka:

bootstrap-servers: 192.168.239.128:9092

producer:

key-serializer: org.apache.kafka.common.serialization.StringSerializer

value-serializer: org.apache.kafka.common.serialization.StringSerializer

consumer:

group-id: test

enable-auto-commit: true

auto-commit-interval: 1000

key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

生产者代码

public class TestKafkaProducerController {

@Autowired

private KafkaTemplate kafkaTemplate;

@RequestMapping("send")

public String send(String msg){

kafkaTemplate.send("test_topic