<dependencies>
        <!-- springBoot集成kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
        <!-- SpringBoot整合Web組件 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    </dependencies>

# kafka
spring:
  kafka:
    # kafka服務(wù)器地址(可以多個(gè))
#    bootstrap-servers: 192.168.212.164:9092,192.168.212.167:9092,192.168.212.168:9092
    bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094
    consumer:
      # 指定一個(gè)默認(rèn)的組名
      group-id: kafkaGroup1
      # earliest:當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無(wú)提交的offset時(shí)，從頭開始消費(fèi)
      # latest:當(dāng)各分區(qū)下有已提交的offset時(shí)，從提交的offset開始消費(fèi)；無(wú)提交的offset時(shí)，消費(fèi)新產(chǎn)生的該分區(qū)下的數(shù)據(jù)
      # none:topic各分區(qū)都存在已提交的offset時(shí)，從offset后開始消費(fèi)；只要有一個(gè)分區(qū)不存在已提交的offset，則拋出異常
      auto-offset-reset: earliest
      # key/value的反序列化
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      # key/value的序列化
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      # 批量抓取
      batch-size: 65536
      # 緩存容量
      buffer-memory: 524288
      # 服務(wù)器地址
      bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094

@RestController
public class KafkaController {
	/**
	 * 注入kafkaTemplate
	 */
	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	/**
	 * 發(fā)送消息的方法
	 *
	 * @param key
	 *            推送數(shù)據(jù)的key
	 * @param data
	 *            推送數(shù)據(jù)的data
	 */
	private void send(String key, String data) {
		// topic 名稱 key   data 消息數(shù)據(jù)
		kafkaTemplate.send("kaico", key, data);
	}
	// test 主題 1 my_test 3
	@RequestMapping("/kafka")
	public String testKafka() {
		int iMax = 6;
		for (int i = 1; i < iMax; i++) {
			send("key" + i, "data" + i);
		}
		return "success";
	}
}

@Component
public class TopicKaicoConsumer {
    /**
     * 消費(fèi)者使用日志打印消息
     */
    @KafkaListener(topics = "kaico") //監(jiān)聽的主題
    public void receive(ConsumerRecord<?, ?> consumer) {
        System.out.println("topic名稱:" + consumer.topic() + ",key:" +
                consumer.key() + "," +
                "分區(qū)位置:" + consumer.partition()
                + ", 下標(biāo)" + consumer.offset());
        //輸出key對(duì)應(yīng)的value的值
        System.out.println(consumer.value());
    }
}