1、引入kafka的依賴

        <dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-kafka</artifactId></dependency>

2、配置kafka

spring:kafka:bootstrap-servers: 156.65.20.76:9092,156.65.20.77:9092,156.65.20.78:9092 #指定Kafka集群的地址，這里有三個地址，用逗號分隔。listener:ack-mode: manual_immediate #設置消費者的確認模式為manual_immediate，表示消費者在接收到消息后立即手動確認。concurrency: 3  #設置消費者的并發數為5missing-topics-fatal: false  #設置為false，表示如果消費者訂閱的主題不存在，不會拋出異常。producer:key-serializer: org.apache.kafka.common.serialization.StringSerializer  # 設置消息鍵的序列化器value-serializer: org.apache.kafka.common.serialization.StringSerializer #設置消息值的序列化器acks: 1  #一般就是選擇1，兼顧可靠性和吞吐量 ，如果想要更高的吞吐量設置為0，如果要求更高的可靠性就設置為-1consumer:auto-offset-reset: earliest #設置為"earliest"表示將從最早的可用消息開始消費，即從分區的起始位置開始讀取消息。enable-auto-commit: false #禁用了自動提交偏移量的功能，為了避免出現重復數據和數據丟失，一般都是手動提交key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  # 設置消息鍵的反序列化器value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #設置消息值的反序列化器

3、創建主題

• 自動創建(不推薦)

  在kafka的安裝目錄conf目錄下找到該配置文件server.properties，添加如下配置：
  num.partitions=3 #默認3個分區
  auto.create.topics.enable=true #開啟自動創建主題
  default.replication.factor=3 #默認3個副本
  

• 手動創建

在kafka的安裝目錄bin目錄下，執行如下命令：
//創建一個有三個分區和三個副本，名為zhuoye的主題
./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3  --topic xxxx

4、生產者代碼

@Slf4j
@Component
public class ALiYunServiceImpl implents IALiYunService {@Autowiredprivate KafkaTemplate kafkaTemplate;@Autowiredprivate ExecutorService executorService;String topicName = "xxxx";@Overridepublic void queryInfo() {List<Message> messages = Collections.synchronizedList(new ArrayList<>());boolean flag = true;//獲取上次查詢時間Long startTime = Long.valueOf(queryTimeRecordMapper.selectTimeByBelongId(3)) * 1000;Long endTime = System.currentTimeMillis();try {List<CloudInstanceAssetDto> cloudInstances = cloudInstanceAssetMapper.queryAllRunningInstance(1, "Running");if (CollectionUtils.isEmpty(cloudInstances)) {return;}//定義計數器CountDownLatch latch = new CountDownLatch(cloudInstances.size());//遍歷查詢for (CloudInstanceAssetDto instance : cloudInstances) {executorService.submit(() -> {try {dealMetricDataToMessage(ALiYunConstant.ECS_INTRANET_OUT_RATE, ALiYunConstant.INTRANET_OUT_RATE_NAME, ALiYunConstant.LW_INTRANET_OUT_RATE_CODE,startTime, endTime, instance, messages);} catch (Exception e) {} finally {latch.countDown();}});}//等待任務執行完畢latch.await();//將最終的消息集合發送到kafkaif (CollectionUtils.isNotEmpty(messages)) {for (int i = 0; i < messages.size(); i++) {if (StringUtils.isNotBlank(messages.get(i).getValue())&& "noSuchInstance".equals(messages.get(i).getValue())) {continue;}kafkaTemplate.send(topicName,  messages.get(i));}}} catch (Exception e) {flag = false;}}

這個時候，如果你想看有沒有把消息發送到kafka的指定主題可以使用如下命令：

kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic xxxx

5、消費者代碼

@Slf4j
@Component
public class KafkaConsumer {// 消費監聽@KafkaListener(topics = "xxxx",groupId ="aliyunmetric")public void consumeExtractorChangeMessage(ConsumerRecord<String, String> record, Acknowledgment ack){try {String value = record.value();//處理數據，存入openTsDb.................................ack.acknowledge();//手動提交}catch (Exception e){log.error("kafa-topic【zhuoye】消費阿里云指標源消息【失敗】");log.error(e.getMessage());}}
}

6、常用Kafka的命令

//創建主題
./kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3  --topic zhuoye
//查看kafka是否接收對應的消息kafka-console-consumer.sh --bootstrap-server localhost:9093 --topic xxxx
// 修改kafka-topic分區數
./kafka-topics.sh --zookeeper localhost:2181 -alter --partitions 6 --topic xxxx
// 查看topic分區數
./kafka-topics.sh --zookeeper localhost:2181 --describe --topic xxxx
// 查看用戶組消費情況
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group zhuoye-aliyunmetric --describe