上干货

1. pom

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>

2. runner

import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

@Component
public class KafkaRunner implements ApplicationRunner {

    @Override
    public void run(ApplicationArguments args) throws Exception {
        new Thread(new OriginalConsumer("xxxx", "xxyy")).start();
    }
}

3. consumer

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 官方jar实现消费者
 */
public class OriginalConsumer implements Runnable {

    /**
     * 服务器列表，多个用 , 隔开
     */
    private static final String BROKER_LIST = "localhost:9092";

    /**
     * 客户端收到消息后，在处理消息前自动提交（即消息已消费），增加偏移量
     */
    private static final String ENABLE_AUTO_COMMIT_CONFIG = "true";

    /**
     * 自动提交的频率，1S
     */
    private static final String AUTO_COMMIT_INTERVAL_MS_CONFIG = "1000";
    private static final String TOPIC = "TEST_TOPIC";
    private KafkaConsumer<String, String> consumer;
    private String clientId;

    public OriginalConsumer(String groupId, String clientId) {
        this.clientId = clientId;
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ENABLE_AUTO_COMMIT_CONFIG);
        properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, AUTO_COMMIT_INTERVAL_MS_CONFIG);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumer = new KafkaConsumer<String, String>(properties);
        consumer.subscribe(Arrays.asList(TOPIC));
    }

    private void consume() {
        // 获取消息，超时时间：1S
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2L));
        records.forEach(record -> {
            System.out.println(clientId + "-消费消息：" + record.toString());
        });
    }

    @Override
    public void run() {
        try {
            while (true) {
                consume();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}