引入依赖

<!--kafka支持-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!--kafka支持-->

配置yml

spring:
  kafka:
    bootstrap-servers: 192.168.1.63:9092 #kafka地址
    template:
      default-topic: mykafka #topic 名
    listener:
      concurrency: 5 #指定listener 容器中的线程数，用于提高并发量
    consumer:
      group-id: myGroup # 指定默认消费者group id
      client-id: 200
      max-poll-records: 200
      auto-offset-reset: earliest # 最早未被消费的offset
    producer:
      batch-size: 1000 # 每次批量发送消息的数量
      retries: 3
      client-id: 200

构建生产者

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import java.util.concurrent.ExecutionException;
/** * spring 2020/6/9 */
@Service
@Component
public class Producer { 
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    /** * 发送消息到kafka */
    public RecordMetadata sendChannelMess(String topic, String message) { 
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        RecordMetadata recordMetadata = null;
        try { 
            recordMetadata = future.get().getRecordMetadata();
        } catch (InterruptedException | ExecutionException e) { 
            e.printStackTrace();
            System.out.println("发送失败");
        }
        System.out.println("发送成功");
        System.out.println("partition:" + recordMetadata.partition());
        System.out.println("offset:" + recordMetadata.offset());
        System.out.println("topic:" + recordMetadata.topic());
        return recordMetadata;
    }
}

构建消费者

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
/** * spring 2020/6/9 */
@Service
@Component
public class Consumer { 
    private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
    /** * 有消息就读取,只读取消息value */
    @KafkaListener(topics = { "mykafka"})
    public void receiveMessage(String message){ 
        logger.error("***********************************************************");
        logger.error(message);
        logger.error("***********************************************************");
    }
}