欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

SpringBoot如何集成Kafka低版本和高版本

 更新時間：2024年03月21日 08:42:28   作者：誰不想飛舞青春  
這篇文章主要介紹了SpringBoot如何集成Kafka低版本和高版本問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

說明

這里之所以集成低版本和新版本，是因為在企業(yè)開發(fā)中，有的SpringBoot項目版本很低，像我這個項目版本就很低，是1.4.2.RELEASE版本，而新版本即高版本就是用來自己學習的。

這里主要告訴大家，版本一定要根據(jù)自己的項目版本選擇對應的kafka版本。

地址

官網(wǎng)地址:https://spring.io/projects/spring-kafka#overview

maven倉庫spring-kafka地址:https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka

官網(wǎng)對應版本圖:

低版本SpringBoot集成Kafka代碼

linux本地服務器zookeeper和kafka使用版本：

springboot版本和使用的spring版本:

使用的spring-kafka版本:

這里我SpringBoot版本是1.4.2.RELEASE版本，版本很低,官網(wǎng)顯示的SpringBoot版本最低是1.5.x，可以使用1.3.x的版本，很明顯我的這個不在官網(wǎng)給的范圍內(nèi)，然后我的spring版本是4.3.9.RELEASE，這里我在上面這個maven倉庫spring-kafka地址里面看了一個1.3.0版本，如下:

直到我往下繼續(xù)找，終于發(fā)現(xiàn)1.2.2.RELEASE這個版本是與我項目對應的。

剛好這個版本對應的spring版本是4.3.9.RELEASE與我項目的spring版本一致，于是我就使用了這個spring-kafka版本。

好了，這里怎么選擇版本就說到這里，下面是代碼。

代碼

這里之所以是在Java類里面寫生產(chǎn)者和消費者配置，是因為springboot和kafka集成版本太低，不支持直接在application.yml里面配置，好像springboot高版本至少2.幾的版本可以直接在application.yml里面配置，至于2.幾的版本才支持我給忘記了，有知道的小伙伴麻煩告訴下我，謝謝了。

kafka生產(chǎn)者配置

這里是帶用戶名密碼協(xié)議配置，最下面三個就是，協(xié)議類型為:SASL/SCRAM-SHA-256，如果你們那里的kafka配置沒有設置這個，可以不需要配置最下面三個。

企業(yè)開發(fā)一般需要進行認證才能發(fā)送消息。

package com.gmcc.project.controllers.kafka;

import lombok.Data;
import org.springframework.context.annotation.Configuration;

//kafka生產(chǎn)者參數(shù)配置
@Data
@Configuration
public class KafkaProducerProperties {
    //指定kafka 代理地址,多個地址用英文逗號隔開
    private String bootstrapServers="192.168.11.111:9092,192.168.11.112:9093";//本地測試kafka使用
    //消息重發(fā)次數(shù),如果配置了事務,則不能為0,改為1
    private int retries=0;
    //每次批量發(fā)送消息的數(shù)量
    private String batchSize="16384";
    //默認值為0,意思就是說消息必須立即被發(fā)送,但這樣會影響性能
    //一般設置10毫秒左右,這個消息發(fā)送完后會進入本地的一個batch,如果10毫秒內(nèi)這個batch滿了16kb就會隨batch一起發(fā)送出去
    private String lingerMs="10";
    //生產(chǎn)者最大可發(fā)送的消息大小,內(nèi)有多個batch,一旦滿了,只有發(fā)送到kafka后才能空出位置,否則阻塞接收新消息
    private String bufferMemory="33554432";
    //指定消息key和消息體的編解碼方式
    private String keySerializer="org.apache.kafka.common.serialization.StringSerializer";
    private String valueSerializer="org.apache.kafka.common.serialization.StringSerializer";
    //確認等級ack,kafka生產(chǎn)端最重要的選項,如果配置了事務,那必須是-1或者all
    //acks=0,生產(chǎn)者在成功寫入消息之前不會等待任何來自服務器的響應
    //acks=1,只要集群的首領節(jié)點收到消息,生產(chǎn)者就會收到一個來自服務器成功響應
    //acks=-1,表示分區(qū)leader必須等待消息被成功寫入到所有的ISR副本(同步副本)中才認為product請求成功。這種方案提供最高的消息持久性保證,但是理論上吞吐率也是最差的
    private String acks="1";
    //協(xié)議類型,為SASL類型
    private String securityProtocol="SASL_PLAINTEXT";
    //協(xié)議
    private String saslMechanism="SCRAM-SHA-256";
    //用戶名密碼配置
    private String saslJaas="org.apache.kafka.common.security.scram.ScramLoginModule required username=root password=123456;";
}

然后再創(chuàng)建一個config使kafka生產(chǎn)者配置生效。

如果kafka配置文件沒有設置用戶名密碼協(xié)議，注釋掉最下面三個即可。

package com.gmcc.project.controllers.config;

import com.gmcc.project.controllers.kafka.KafkaProducerProperties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaProductConfig {
    @Autowired
    private KafkaProducerProperties producerProperties;

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, producerProperties.getBootstrapServers());
        props.put(ProducerConfig.RETRIES_CONFIG, producerProperties.getRetries());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, producerProperties.getBatchSize());
        props.put(ProducerConfig.LINGER_MS_CONFIG, producerProperties.getLingerMs());
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, producerProperties.getBufferMemory());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, producerProperties.getKeySerializer());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, producerProperties.getValueSerializer());
        props.put(ProducerConfig.ACKS_CONFIG, producerProperties.getAcks());
        //props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, producerProperties.getSecurityProtocol());
        //props.put(SaslConfigs.SASL_MECHANISM, producerProperties.getSaslMechanism());
        //props.put(SaslConfigs.SASL_JAAS_CONFIG, producerProperties.getSaslJaas());
        return props;
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

kafka消費者配置

如果kafka配置文件沒有配置用戶名密碼協(xié)議，認證后才能消費消息，可以將最下面的三個注釋掉不使用。

package com.gmcc.project.controllers.kafka;

import lombok.Data;
import org.springframework.context.annotation.Configuration;

//kafka消費者配置
@Data
@Configuration
public class KafkaConsumerProperties {
    //指定kafka 代理地址,多個地址用英文逗號隔開
    private String bootstrapServers="192.168.11.111:9092,192.168.11.112:9093";//本地測試kafka使用
    //指定默認消費者group id,消費者監(jiān)聽到的也是這個
    private String groupId="test-consumer-group";//本地測試使用
    //消費者在讀取一個沒有offset的分區(qū)或者offset無效時的策略,默認earliest是從頭讀,latest不是從頭讀
    private String autoOffsetReset="earliest";
    //是否自動提交偏移量offset,默認為true,一般是false,如果為false,則auto-commit-interval屬性就會無效
    private boolean  enableAutoCommit=true;
    //自動提交間隔時間,接收到消息后多久會提交offset,前提需要開啟自動提交,也就是enable-auto-commit設置為true,默認單位是毫秒(ms)，如果寫10s,最后加載的顯示值為10000ms,需要符合特定時間格式:1000ms,1S,1M,1H,1D(毫秒,秒,分,小時,天)
    private String autoCommitInterval="1000";
    //指定消息key和消息體的編解碼方式
    private String keyDeserializerClass="org.apache.kafka.common.serialization.StringDeserializer";
    private String valueDeserializerClass ="org.apache.kafka.common.serialization.StringDeserializer";
    //批量消費每次最多消費多少條信息
    private String maxPollRecords="50";
    //協(xié)議類型,為SASL類型
    private String securityProtocol="SASL_PLAINTEXT";
    //協(xié)議
    private String saslMechanism="SCRAM-SHA-256";
    //用戶名密碼配置
    private String saslJaas="org.apache.kafka.common.security.scram.ScramLoginModule required username=root password=123456;";
}

然后再創(chuàng)建一個config使kafka消費者配置生效。如果kafka沒有設置用戶名密碼協(xié)議，注釋掉最下面三個即可。

package com.gmcc.project.controllers.config;

import com.gmcc.project.controllers.kafka.KafkaConsumerProperties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConsumerConfig {
    @Autowired
    private KafkaConsumerProperties consumerProperties;

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        //設置為批量消費，每個批次數(shù)量在Kafka配置參數(shù)中設置ConsumerConfig.MAX_POLL_RECORDS_CONFIG
        factory.setBatchListener(false);//這里為true的時候,KafkaConsumer那里需要使用批量消費方法,不然報錯
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumerProperties.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerProperties.getGroupId());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, consumerProperties.getAutoOffsetReset());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, consumerProperties.isEnableAutoCommit());
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, consumerProperties.getAutoCommitInterval());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, consumerProperties.getKeyDeserializerClass());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, consumerProperties.getValueDeserializerClass());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, consumerProperties.getMaxPollRecords());
        //props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, consumerProperties.getSecurityProtocol());
        //props.put(SaslConfigs.SASL_MECHANISM, consumerProperties.getSaslMechanism());
        //props.put(SaslConfigs.SASL_JAAS_CONFIG, consumerProperties.getSaslJaas());
        return props;
    }
}

發(fā)送消息給kafka的Controller代碼

這里使用addCallback這個方法，是可以在生產(chǎn)者發(fā)送消息給kafka時，如果生產(chǎn)者配置有問題或者服務有問題，我可以直接看到接口返回結(jié)果，所以沒有直接這樣kafkaTemplate.send(“first”,data);寫。

package com.gmcc.project.controllers.kafka;

import com.gmcc.project.core.utils.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

//kafka生產(chǎn)者
@RestController
@RequestMapping("kafkaProducer")
public class KafkaProducerController {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;

    //向kafka發(fā)送消息
    @RequestMapping(value = "/sendFileMd5", method = RequestMethod.POST)
    public Map<String, Object> sendFileMd5(@RequestParam(value = "fileMd5", required = false) String fileMd5,
                                           @RequestParam(value = "uuid", required = false) String uuid){
        Map<String, Object> returnMap = new HashMap<>();
        //寫在success里面只會返回一次,第二次就給你返回一個空map對象
        returnMap.put("message", "發(fā)送消息成功!");
        returnMap.put("result", null);
        returnMap.put("status", "200");
        //非空判斷
        if(StringUtils.isBlank(fileMd5)) {
            returnMap.put("message", "fileMd5不能為空!");
            returnMap.put("result", "");
            returnMap.put("status", "999");
            return returnMap;
        }
        if(StringUtils.isBlank(uuid)) {
            returnMap.put("message", "uuid不能為空!");
            returnMap.put("result", "");
            returnMap.put("status", "999");
            return returnMap;
        }
        try{
            //需要發(fā)送的消息
            String data="{\"file_md5\":\""+fileMd5+"\",\"uuid\":\""+uuid+"\",\"vendor\":\"etone\",\"model\":\"5g信令回放\"}";
            //pro環(huán)境使用topic為test_sample_get
            //本地測試使用,向topic為first發(fā)送消息
            kafkaTemplate.send("first",data).addCallback(success -> {
                // 消息發(fā)送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息發(fā)送到的分區(qū)
                int partition = success.getRecordMetadata().partition();
                // 消息在分區(qū)內(nèi)的offset
                long offset = success.getRecordMetadata().offset();
                System.out.println("發(fā)送消息成功："+data+"，主題："+topic+"，分區(qū)："+partition+"，偏移量："+offset);
            }, failure -> {
                returnMap.put("message", "發(fā)送消息失敗:" + failure.getMessage());
                returnMap.put("result", null);
                returnMap.put("status", "500");
            });
        }catch (Exception e){
            returnMap.put("message", e.getMessage());
            returnMap.put("result", null);
            returnMap.put("status", "500");
        }
        return returnMap;
    }
}

消費者消費代碼

package com.gmcc.project.controllers.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    //逐條消費
    @KafkaListener(topics = "first")
    //@KafkaListener(topics = "test_sample_return")
    public void onMessage(ConsumerRecord<?,?> record){

        try{
            //消費的哪個topic、partition的消息，打印出消息內(nèi)容
            System.out.println("消費："+record.topic()+"-"+record.partition()+"-"+record.value());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    //批量消費方法
    /*@KafkaListener(topics = "first")
    public void onMessage(List<ConsumerRecord<?,?>> records){
        System.out.println("消費數(shù)量="+records.size());
        for(ConsumerRecord<?,?> record:records){
            //消費的哪個topic、partition的消息，打印出消息內(nèi)容
            System.out.println("消費："+record.topic()+"-"+record.partition()+"-"+record.value());
        }
    }*/
}

消費到的消息:

這里面的uuid是集成了websocket需要用到,這里怎么集成websocket將消費到的消息返回給客戶端等以后有時間了在另寫一個博客說明。

高版本SpringBoot集成Kafka代碼

這里高版本可以供自己學習。高版本集成很簡單，沒有低版本那么麻煩。

代碼結(jié)構:

pom.xml文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.6.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.hjl</groupId>
    <artifactId>kafka-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>kafka-demo</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.6.2</version>
            </plugin>
        </plugins>
    </build>

</project>

這里我的SpringBoot版本是2.6.2版本,spring-kafka版本是2.8.1版本。符合官網(wǎng)給的版本推薦。

如下:

application.yml文件

這里之所以可以在application.yml直接配置kafka，是因為springboot和spring-kafka版本很高。

這里生產(chǎn)者配置和消費者配置都在里面。

server:
  port: 8080

spring:
  kafka:
    # 指定kafka 代理地址,多個地址用英文逗號隔開
    bootstrap-servers: 192.168.11.111:9092
    #初始化生產(chǎn)者配置
    producer:
      #消息重發(fā)次數(shù),如果配置了事務,則不能為0,改為1
      retries: 0
      # 每次批量發(fā)送消息的數(shù)量
      batch-size: 16384
      #生產(chǎn)者最大可發(fā)送的消息大小,內(nèi)有多個batch,一旦滿了,只有發(fā)送到kafka后才能空出位置,否則阻塞接收新消息
      buffer-memory: 33554432
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      #確認等級ack,kafka生產(chǎn)端最重要的選項,如果配置了事務,那必須是-1或者all
      #acks=0,生產(chǎn)者在成功寫入消息之前不會等待任何來自服務器的響應
      #acks=1,只要集群的首領節(jié)點收到消息,生產(chǎn)者就會收到一個來自服務器成功響應
      #acks=-1,表示分區(qū)leader必須等待消息被成功寫入到所有的ISR副本(同步副本)中才認為product請求成功。這種方案提供最高的消息持久性保證,但是理論上吞吐率也是最差的
      acks: all
      #配置事務，名字隨便起
      #transaction-id-prefix: hbz-transaction-

    #初始化消費者配置
    consumer:
      # 指定默認消費者group id,消費者監(jiān)聽到的也是這個
      group-id: test-consumer-group
      #消費者在讀取一個沒有offset的分區(qū)或者offset無效時的策略,默認earliest是從頭讀,latest不是從頭讀
      auto-offset-reset: earliest
      #是否自動提交偏移量offset,默認為true,一般是false,如果為false,則auto-commit-interval屬性就會無效
      enable-auto-commit: true
      #自動提交間隔時間,接收到消息后多久會提交offset,前提需要開啟自動提交,也就是enable-auto-commit設置為true,默認單位是毫秒(ms)，如果寫10s,最后加載的顯示值為10000ms,需要符合特定時間格式:1000ms,1S,1M,1H,1D(毫秒,秒,分,小時,天)
      auto-commit-interval: 1000
      # 指定消息key和消息體的編解碼方式
      key-serializer: org.apache.kafka.common.serialization.StringDeserializer
      value-serializer: org.apache.kafka.common.serialization.StringDeserializer
      #批量消費每次最多消費多少條信息
      max-poll-records: 50

    #監(jiān)聽器設置
    listener:
      #消費端監(jiān)聽的topic不存在時,項目啟動會報錯(關掉)
      missing-topics-fatal: false
      #設置消費類型 批量消費batch,單條消費single
      type: batch
      #指定容器的線程數(shù),提高并發(fā)量，默認為1
      #concurrency: 3
      #手動提交偏移量,當enable-auto-commit為true自動提交時,不需要設置改屬性
      #ack-mode: manual

生產(chǎn)者發(fā)送消息代碼

package com.project.kafkademo.kafkaproduct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

//kafka生產(chǎn)者
@RestController
@RequestMapping("kafka")
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @RequestMapping(value = "/send", method = RequestMethod.GET)
    public String send(@RequestParam(value = "message", required = false) String message){
        kafkaTemplate.send("first",message);
        return "success";
    }
}

消費者消費消息代碼

package com.project.kafkademo.kafkaconsumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class KafkaConsumer {

    //消費監(jiān)聽,topics=監(jiān)聽的主題名,groupId=分組,consumer.properties里面的group.id配置
    //如果在這里直接寫groupId="test-consumer-group"會導致application.yml里面設置的group-id不起效
    //最終會被這里的設置直接覆蓋掉，所以這里不應該加groupId="test-consumer-group"這個屬性
    //@KafkaListener(topics = "first",groupId="test-consumer-group")
    //這樣寫的話,application.yml里面設置的group-id就會生效,監(jiān)控的就是application.yml里面的了
    //逐條消費
    /*@KafkaListener(topics = "first")
    public void onMessage(ConsumerRecord<?,?> record){
        //消費的哪個topic、partition的消息，打印出消息內(nèi)容
        System.out.println("消費："+record.topic()+"-"+record.partition()+"-"+record.value());
    }*/

    //批量消費,用List批量接收消息,ConsumerRecord<?,?>只能單條消費消息
    /*@KafkaListener(topics = "first")
    public void onMessage(List<ConsumerRecord<?,?>> records){
        System.out.println("消費數(shù)量="+records.size());
        for(ConsumerRecord<?,?> record:records){
            //消費的哪個topic、partition的消息，打印出消息內(nèi)容
            System.out.println("消費："+record.topic()+"-"+record.partition()+"-"+record.value());
        }
    }*/

    //批量消費,ConsumerRecords<?,?>用于批量消費消息
    @KafkaListener(topics = "first")
    public void onMessage(ConsumerRecords<?,?> records){
        System.out.println("消費數(shù)量="+records.count());
        for(ConsumerRecord<?,?> record:records){
            //消費的哪個topic、partition(哪個分區(qū))的消息，打印出消息內(nèi)容
            System.out.println("消費："+record.topic()+"-"+record.partition()+"-"+record.key()+"-"+record.value());
        }
    }
}

效果

項目啟動后，會打印出你配置的參數(shù)以及默認配置的參數(shù)

postman接口測試:

后臺結(jié)果打印:

總結(jié)

好了，我的記錄就先到這里。

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

相關文章

  • 詳解Java的日期時間新特性

    詳解Java的日期時間新特性

    隨著時間的不斷推移，現(xiàn)實的需求也在不斷更新，原先的一些API已經(jīng)難以滿足開發(fā)需求了，從JDK?8之后，為了滿足更多的開發(fā)需求，Java給我們增加了不少關于日期時間的新特性，接下來就帶各位來看看這些新特性有哪些,需要的朋友可以參考下
    2023-06-06
  • Spring時間戳(日期)格式轉(zhuǎn)換方式

    Spring時間戳(日期)格式轉(zhuǎn)換方式

    這篇文章主要介紹了Spring時間戳(日期)格式轉(zhuǎn)換方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
    2025-03-03
  • Java函數(shù)式編程(八)：字符串及方法引用

    Java函數(shù)式編程(八)：字符串及方法引用

    這篇文章主要介紹了Java函數(shù)式編程(八)：字符串及方法引用,本文是系列文章的第8篇,其它文章請參閱本文底部的相關文章,需要的朋友可以參考下
    2014-09-09
  • Spring Boot 中的 @ConditionalOnBean 注解作用及基本用法

    Spring Boot 中的 @ConditionalOnBean 注解作用及基

    在 Spring Boot 中,@ConditionalOnBean 可以幫助我們根據(jù) 是否存在特定 Bean 來 動態(tài)注冊 Bean,廣泛用于 按需加載、自動配置 等場景,本文給大家介紹Spring Boot 中的 @ConditionalOnBean 注解,感興趣的朋友一起看看吧
    2025-04-04
  • Java結(jié)構型設計模式之組合模式Composite Pattern詳解

    Java結(jié)構型設計模式之組合模式Composite Pattern詳解

    組合模式，又叫部分整體模式，它創(chuàng)建了對象組的數(shù)據(jù)結(jié)構組合模式使得用戶對單個對象和組合對象的訪問具有一致性。本文將通過示例為大家詳細介紹一下組合模式，需要的可以參考一下
    2022-11-11
  • Spring Boot中數(shù)據(jù)庫操作Druid和HikariDataSource的詳細過程

    Spring Boot中數(shù)據(jù)庫操作Druid和HikariDataSource的詳細過程

    這篇文章主要介紹了Spring Boot中數(shù)據(jù)庫操作Druid和HikariDataSource的詳細過程,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下
    2023-06-06
  • Springboot項目長時間不進行接口操作,提示HikariPool-1警告的解決

    Springboot項目長時間不進行接口操作,提示HikariPool-1警告的解決

    這篇文章主要介紹了Springboot項目長時間不進行接口操作,提示HikariPool-1警告的解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教
    2023-12-12
  • Java命名規(guī)范

    Java命名規(guī)范

    本文主要介紹了Java命名規(guī)范。具有一定的參考價值，下面跟著小編一起來看下吧
    2017-01-01
  • Java實戰(zhàn)之實現(xiàn)OA辦公管理系統(tǒng)

    Java實戰(zhàn)之實現(xiàn)OA辦公管理系統(tǒng)

    這篇文章主要介紹了如何通過Java實現(xiàn)OA辦公管理系統(tǒng)，文章采用到了JSP、JQuery、Ajax等技術，文中的示例代碼講解詳細，感興趣的小伙伴可以了解一下
    2022-02-02
  • 利用Java8 Optional類優(yōu)雅如何地解決空指針問題

    利用Java8 Optional類優(yōu)雅如何地解決空指針問題

    這篇文章主要給大家介紹了關于如何利用Java8 Optional類優(yōu)雅解決空指針問題的相關資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
    2020-11-11

最新評論