Spring下实现kafka

2021-10-19

Spring下实现kafka

原理

kafka下载

地址：http://kafka.apache.org/downloads

JDK前提

Window

下载地址：https://www.java.com/zh-CN/download/

配置环境变量：https://www.runoob.com/java/java-environment-setup.html

必须通过JAVA_HOME方式配置

Linux

命令行安装openJDK即可

启动kafka

Spring项目依赖于zookeeper和kafka的开启

Window

下载成功后解压到电脑指定目录

里面自带有zookeeper、配置文件 server.properties(地址、ip等配置信息)，本地简单版本默认就行。

cmd进入到kafka解压后目录，执行以下命令。

注意启动zookeeper后不要关闭窗口，再新开一个cmd窗口启动kafka。

// 启动zookeeper
bin\windows\zookeeper-server-start.bat config\zookeeper.properties

// 启动kafka，新开cmd
bin\windows\kafka-server-start.bat config\server.properties

// 创建topic，test是主题名，可自行定义
bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

// 查看topic
bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

// 启动生产端，新开窗口方便测试
bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test

// 启动消费端，新开窗口方便测试
bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test--from-beginning

// 本地测试
在生产端输入信息，消费端消费信息。

Linux

# 启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 启动kafka
bin/kafka-server-start.sh config/server.properties &

# 创建topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

# 查看Topic
bin/kafka-topics.sh --list --zookeeper localhost:2181

# 发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

# 消费消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

# 关闭zookeeper
bin/zookeeper-server-stop.sh config/zookeeper.properties &

# 关闭kafka
bin/kafka-server-stop.sh config/server.properties &

kafka服务与spring服务在不同机子

kafka服务所在的机子需要配置config/server.properties

1 2	listeners=PLAINTEXT://IP地址:9092 IP地址是kafka服务所在机子的IP地址

如果不配置具体IP地址，则默认使用localhost，不在kafka所在机子的spring建立连接时就会报错误（Connection to node 1 (localhost/127.0.0.1:9092) could not be established），因为kafka在zookeeper中注册时使用的是localhost。

Gradle

1 2	// https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka implementation group: 'org.springframework.kafka', name: 'spring-kafka', version: '2.7.6'

resources下application.yml

# kafka配置
spring:
  kafka:
    # broker地址:端口
    # IP为kafka服务IP
    bootstrap-servers: IP:9092

    # 生产者
    producer:
      retries: 0
      # 发送消息序列化时使用的类
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      enable-auto-commit: false
      auto-commit-interval: 100ms
      # 发送消息反序列化时使用的类
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: latest
      group-id: test-consumer-group

kafka包下KafkaProducer

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Component
public class KafkaProducer {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    // log
    private final Logger logger = LoggerFactory.getLogger(KafkaProducer.class);

    /**
     * Send msg.
     *
     * @param topicName the topic name
     * @param object    the object
     */
    public void send(String topicName, Object object) {
        String msg = JsonUtils.objectToJsonString(object);
        logger.info("ready send msg, data is: {}", msg);

        kafkaTemplate.send(topicName, msg)
                .addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onFailure(@Nullable Throwable throwable) {
                        // send fail process...
                        logger.info("log:发送数据失败...,{}", throwable.getMessage());
                    }

                    @Override
                    public void onSuccess(SendResult<String, String> sendResult) {
                        RecordMetadata rmd = sendResult.getRecordMetadata();
                        logger.info("log:发送数据成功,--topic:【{}】,--partition:【{}】,--offset:【{}】,--timestamp:【{}】",
                                rmd.topic(), rmd.partition(), rmd.offset(), rmd.timestamp());
                    }
                });
    }
}

kafka包下KafkaConsumer

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class KafkaConsumer {

    // log
    private final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    /**
     * Listener topic.
     *
     * @param record the record
     * @param topic  the topic
     */
    @KafkaListener(groupId = "consumer-group", topicPattern = "topic.*")
    public void listenerTopic(ConsumerRecord<String, String> record,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        try {
            Optional<Entity> optionMsg = Optional.ofNullable(JsonUtils.jsonStringToPojo(record.value(), Entity.class));
            if (optionMsg.isPresent()) {
                Entity msg = optionMsg.get();
                logger.info("log:成功消费Topic:【{}】的消息,数据为:【{}】", topic, msg);
                // 消费的业务逻辑处理
                // ......
                // end
            }
        } catch (Exception e) {
            logger.error("log:接受数据失败...{}", e.getMessage());
            e.printStackTrace();
        }
    }
}

KafkaController

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Autowired
    private KafkaProducer kafkaProducer;

    @PostMapping("/send/{topic}")
    public void sendMsg(@RequestBody Entity entity, @PathVariable("topic") String topic) {
        kafkaProducer.send(topic, entity);
    }
}

Entity

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class Entity {

    private Integer id;

    private String name;
}

公共类JsonUtils

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

public class JsonUtils {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    public static String objectToJsonString(Object data) {
        try {
            return MAPPER.writeValueAsString(data);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            return null;
        }
    }

    public static <T> T jsonStringToPojo(String jsonData, Class<T> beanType) {
        try {
            return MAPPER.readValue(jsonData, beanType);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            return null;
        }
    }
}

应用

带参请求controller，发布主题topic的消息。

KafkaConsumer组件监听主题topic消息，获取数据执行相应的业务操作。KafkaConsumer监听所有前缀为 “topic.” 的主题消息。