KAFKA SpringBoot2 Nacos 消息異步發(fā)送和消費消息（進(jìn)階篇）
2022-09-06 22:42:16

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

spring.application.name=springboot-kafka
server.port=8080
# kafka 配置
spring.kafka.bootstrap-servers=node1:9092

# producer 配置
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.IntegerSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 生產(chǎn)者每個批次最多方多少條記錄
spring.kafka.producer.batch-size=16384
# 生產(chǎn)者一端總的可用緩沖區(qū)大小，此處設(shè)置為32M * 1024 * 1024
spring.kafka.producer.buffer-memory=33544432

# consumer 配置
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.group-id=springboot-consumer-02
# earliest - 如果找不到當(dāng)前消費者的有效偏移量，則自動重置向到最開始
spring.kafka.consumer.auto-offset-reset=earliest
# 消費者的偏移量是自動提交還是手動提交，此處自動提交偏移量
spring.kafka.consumer.enable-auto-commit=true
# 消費者偏移量自動提交時間間隔
spring.kafka.consumer.auto-commit-interval=1000

server:
  port: 8002
spring:
  application:
    # 應(yīng)用名稱
    name: ly-kafka
  profiles:
    # 環(huán)境配置
    active: dev
  cloud:
    nacos:
      discovery:
        # 服務(wù)注冊地址
        server-addr: nacos.server.com:8848
      config:
        # 配置中心地址
        server-addr: nacos.server.com:8848
        # 配置文件格式
        file-extension: yml
        # 共享配置
        shared-configs:
          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}

在這里插入代碼片

package com.gblfy.kafka.entity;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.time.LocalDateTime;

@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class Order {
    /**
     * 訂單id
     */
    private long orderId;
    /**
     * 訂單號
     */
    private String orderNum;
    /**
     * 訂單創(chuàng)建時間
     */
    private LocalDateTime createTime;
}

package com.gblfy.lykafka.provider;

import com.alibaba.fastjson.JSONObject;
import com.gblfy.common.constant.KafkaTopicConstants;
import com.gblfy.common.entity.Order;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.time.LocalDateTime;

/**
 * Kafka生產(chǎn)者
 *
 * @author gblfy
 * @date 2021-09-28
 */
@Service
public class KafkaProvider {
    private final static Logger log = LoggerFactory.getLogger(KafkaProvider.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(long orderId, String orderNum, LocalDateTime createTime) {
        // 構(gòu)建一個訂單類
        Order order = Order.builder()
                .orderId(orderId)
                .orderNum(orderNum)
                .createTime(createTime)
                .build();
        // 發(fā)送消息，訂單類的 json 作為消息體
        ListenableFuture<SendResult<String, String>> future =
                kafkaTemplate.send(KafkaTopicConstants.KAFKA_MSG_TOPIC, JSONObject.toJSONString(order));

        // 監(jiān)聽回調(diào)
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable e) {
                log.info("發(fā)送消息失敗: {}", e.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                RecordMetadata metadata = result.getRecordMetadata();
                log.info("發(fā)送的主題:{} ,發(fā)送的分區(qū):{} ,發(fā)送的偏移量:{}  ",
                        metadata.topic(), metadata.partition(), metadata.offset());
            }
        });
    }
}

package com.gblfy.lykafka.controller;

import com.gblfy.lykafka.provider.KafkaProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;

@RestController
@RequestMapping("/kafka")
public class KafkaProviderController {

    @Autowired
    private KafkaProvider kafkaProvider;

    @GetMapping("/sendMQ")
    public String sendMQContent() {
        kafkaProvider.sendMessage(0001, "10", LocalDateTime.now());
        return "OK";
    }
}

package com.gblfy.common.constant;

public class KafkaTopicConstants {
    //kafka發(fā)送消息主題
    public static final String KAFKA_MSG_TOPIC = "topic-springboot-01";

    // kafka消費者組需要和yml文件中的 kafka.consumer.group-id的值保持一致
    public static final String KAFKA_MSG_TOPIC_GROUP = "springboot-consumer-02";
}

package com.gblfy.kafka;

import com.gblfy.kafka.controller.KafkaProvider;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@SpringBootTest
class KafkaSpringbootApplicationTests {

    @Autowired
    private KafkaProvider kafkaProvider;

    @Test
    public void sendMessage() throws InterruptedException {
        // 發(fā)送 1000 個消息
        for (int i = 0; i < 1000; i++) {
            long orderId = i+1;
            String orderNum = UUID.randomUUID().toString();
            kafkaProvider.sendMessage(orderId, orderNum, LocalDateTime.now());
        }
        TimeUnit.MINUTES.sleep(1);
    }
}