demo-kafka

springboot kafka的一个学习项目

直接去官网的下载 页面

复制下载链接后直接使用wget下载

wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.2.0/kafka_2.11-2.2.0.tgz

下载完成后解压

tar -xf kafka_2.11-2.2.0.tgz

下图是解压后的目录内容

bin文件夹里面存放的是与kafka相关的脚本文件，删除了Windows文件夹

用到的脚本以及作用

使用kafka时首先要启动zookeeper

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

建议使用以下命令，可以使得直接在后台执行

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

zookeeper可以通过jps命令去查看是否启动,如下图发现有QuorumPeerMain则表明已启动

zookeeper默认的端口是2181也可以直接通过命令查看 lsof -i:2181

zookeeper启动之后就可以去启动kafka了，启动时可以修改配置文件 config/server.properties
如果是想要可以通过远程连接kafka需要放开 advertised.listeners=PLAINTEXT://:9092 的注释。
中间填上自己自己服务器的地址即可。
启动后通过 jps 命令可以看到有 kakfa 如下图所示

然后可以在终端中创建一个Topic

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

如下图

上述命令详细解释如下

1. kafka-topics.sh：Kafka 提供的一个 shell 脚本文件(位于 bin 目录中)，用于创建或查看 topic 信息。
2. --create：shell 脚本的参数，告诉 shell 脚本：要创建一个 topic。
3. --zookeeper localhost:2181：shell 脚本的参数，告诉 shell 脚本 Zookeeper 的地址，用于保存 topic 元数据信息。
4. --partitions 1：shell 脚本参数，告诉 shell 脚本：所创建的这个 topic 的 partition 个数为1
5. --replication-factor 1：shell脚本参数，告诉 shell 脚本：每个 partition 的副本数为1
6. --topic testTopic：shell 脚本参数，告诉 shell 脚本：创建的 topic 的名称为 testTopic。

可以查看topic的list来查看刚刚创建的topic

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

如下图

接下来测试生产者和消费者
首先启动生产者

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

然后新建一个终端用来启动消费者

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic

测试生产者输入内容

消费者接收内容

可以看到消息正确的发送和接收到了

引入spring-kafka版本也还和kafka自身的版本有关，可以在如下网页查看 根据个人版本挑选合适的版本。

这个是我自己选择的版本

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.5.RELEASE</version>
</dependency>

引入依赖之后还需要在配置文件中进行配置

spring:
  kafka:
    # 服务地址
    bootstrap-servers: ip:9092
    consumer:
      # 自动提交
      enable-auto-commit: true
      # 序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

java 端消费者代码如下

@Component
public class KafkaConsumer {
    @KafkaListener(topics = {"testTopic"}, groupId = "test")
    public void receive(String message){
        System.out.println(message);
    }
}

测试结果如下

java 端生产者代码如下

@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send(String data){
        kafkaTemplate.send("testTopic", data);
    }
}

测试结果如下