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

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

package com.example.demo.kafka;import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;import java.util.concurrent.ExecutionException;@Component
public class Producer {@Autowiredprivate 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;}
}

package com.example.demo.kafka;import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;import java.util.List;@Component
public class Consumer {/*** 有消息就读取,只读取消息value*/@KafkaListener(topics = {"test13"})public void receiveMessage(String message){//收到通道的消息之后执行秒杀操作
        System.out.println(message);}/*** 有消息就读取,批量读取消息value*/@KafkaListener(topics = "test12")public void onMessage(List<String> crs) {for(String str : crs){System.out.println("test12:" + str);}}/*** 有消息就读取,读取消息topic，offset，key，value等信息*/@KafkaListener(topics = "test14")public void listenT1(ConsumerRecord<?, ?> cr){System.out.println("listenT1收到消息,topic:>>>" + cr.topic() + "  offset:>>" + cr.offset()+ "  key:>>" + cr.key() + "  value:>>" + cr.value());}
}