SpringBoot整合Kafka教程：从零开始实现消息生产和消费

# log文件存储时间，单位为小时，这里设置为1周
log.retention.hours=168

# log文件大小的最大值，这里为1g，超过这个值，则会创建新的segment（也就是新的.index和.log文件）
log.segment.bytes=1073741824

        <dependency>
            <groupid>org.springframework.boot</groupid>
            <artifactid>spring-boot-starter-web</artifactid>
        </dependency>
        <dependency>
            <groupid>org.springframework.kafka</groupid>
            <artifactid>spring-kafka</artifactid>
        </dependency>

server:
  port: 8089
spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092  #指定kafka server的地址，集群配多个，中间，逗号隔开
    producer:
      key-serializer: org.apache.kafka.common.serialization.stringserializer
      value-serializer: org.apache.kafka.common.serialization.stringserializer
    consumer:
      group-id: test-consumer
      auto-offset-reset: latest
      enable-auto-commit: true
      key-deserializer: org.apache.kafka.common.serialization.stringdeserializer
      value-deserializer: org.apache.kafka.common.serialization.stringdeserializer

@restcontroller
@slf4j
public class sendmessagecontroller {

    @autowired
    private kafkatemplate<string, string> kafkatemplate;

    @getmapping("send/{message}")
    public void send(@pathvariable string message) {
        this.kafkatemplate.send("test", message);
    }

}

@restcontroller
@slf4j
public class sendmessagecontroller {

    @autowired
    private kafkatemplate<string, string> kafkatemplate;

    @getmapping("send/{message}")
    public void send(@pathvariable string message) {
        listenablefuture<sendresult<string, string>> future = this.kafkatemplate.send("test", message);
        future.addcallback(new listenablefuturecallback<sendresult<string, string>>() {
            @override
            public void onsuccess(sendresult<string, string> result) {
                log.info("成功发送消息：{}，offset=[{}]", message, result.getrecordmetadata().offset());
            }

            @override
            public void onfailure(throwable ex) {
                log.error("消息：{} 发送失败，原因：{}", message, ex.getmessage());
            }
        });
    }
}

@component
@slf4j
public class kafkamessagelistener {
    @kafkalistener(topics = "test", groupid = "test-consumer")
    public void listen(consumerrecord<?, ?> record) {
        log.info("接收消息: topic is {}, offset is {}, partition is {}, value is {} ", record.topic(), record.offset(), record.partition(), record.value());
    }
}

@kafkalistener(topics = “topic1, topic2”)

@kafkalistener(topics = "test", groupid = "test-consumer")
public void listen(@payload string message,
                   @header(kafkaheaders.received_partition_id) int partition) {
    logger.info("接收消息: {}，partition：{}", message, partition);
}

@component
@slf4j
public class kafkamessagelistener {
    @kafkalistener(topics = "test", groupid = "test-consumer")
    public void listen(consumerrecord<?, ?> record) {
        log.info("接收消息: topic is {}, offset is {}, partition is {}, value is {} ", record.topic(), record.offset(), record.partition(), record.value());
    }
}

@kafkalistener(groupid = "test-consumer",
        topicpartitions = @topicpartition(topic = "test",
                partitionoffsets = {
                        @partitionoffset(partition = "0", initialoffset = "0")
            }))

public void listen(@payload string message,
                   @header(kafkaheaders.received_partition_id) int partition) {
    logger.info("接收消息: {}，partition：{}", message, partition);
}

@kafkalistener(groupid = "test-consumer", 
	topicpartitions = @topicpartition(topic = "test", partitions = { "0", "1" }))

@configuration
public class kafkainitialconfig {

    // 监听器工厂
    @autowired
    private consumerfactory consumerfactory;

    // 配置一个消息过滤策略
    @bean
    public concurrentkafkalistenercontainerfactory<string, string> kafkalistenercontainerfactory() {
        concurrentkafkalistenercontainerfactory<string, string> factory
                = new concurrentkafkalistenercontainerfactory<>();
        factory.setconsumerfactory(consumerfactory);
        // ------- 过滤配置 --------
        factory.setrecordfilterstrategy(
                r -> r.value().contains("test")
        );
        return factory;
    }

}

public interface recordfilterstrategy<k, v> {
    boolean filter(consumerrecord<k, v> var1);
}

spring.kafka.admin.client-id= # id to pass to the server when making requests. used for server-side logging.
spring.kafka.admin.fail-fast=false # whether to fail fast if the broker is not available on startup.
spring.kafka.admin.properties.*= # additional admin-specific properties used to configure the client.
spring.kafka.admin.ssl.key-password= # password of the private key in the key store file.
spring.kafka.admin.ssl.key-store-location= # location of the key store file.
spring.kafka.admin.ssl.key-store-password= # store password for the key store file.
spring.kafka.admin.ssl.key-store-type= # type of the key store.
spring.kafka.admin.ssl.protocol= # ssl protocol to use.
spring.kafka.admin.ssl.trust-store-location= # location of the trust store file.
spring.kafka.admin.ssl.trust-store-password= # store password for the trust store file.
spring.kafka.admin.ssl.trust-store-type= # type of the trust store.
spring.kafka.bootstrap-servers= # comma-delimited list of host:port pairs to use for establishing the initial connections to the kafka cluster. applies to all components unless overridden.
spring.kafka.client-id= # id to pass to the server when making requests. used for server-side logging.
spring.kafka.consumer.auto-commit-interval= # frequency with which the consumer offsets are auto-committed to kafka if 'enable.auto.commit' is set to true.
spring.kafka.consumer.auto-offset-reset= # what to do when there is no initial offset in kafka or if the current offset no longer exists on the server.
spring.kafka.consumer.bootstrap-servers= # comma-delimited list of host:port pairs to use for establishing the initial connections to the kafka cluster. overrides the global property, for consumers.
spring.kafka.consumer.client-id= # id to pass to the server when making requests. used for server-side logging.
spring.kafka.consumer.enable-auto-commit= # whether the consumer's offset is periodically committed in the background.
spring.kafka.consumer.fetch-max-wait= # maximum amount of time the server blocks before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by "fetch-min-size".
spring.kafka.consumer.fetch-min-size= # minimum amount of data the server should return for a fetch request.
spring.kafka.consumer.group-id= # unique string that identifies the consumer group to which this consumer belongs.
spring.kafka.consumer.heartbeat-interval= # expected time between heartbeats to the consumer coordinator.
spring.kafka.consumer.key-deserializer= # deserializer class for keys.
spring.kafka.consumer.max-poll-records= # maximum number of records returned in a single call to poll().
spring.kafka.consumer.properties.*= # additional consumer-specific properties used to configure the client.
spring.kafka.consumer.ssl.key-password= # password of the private key in the key store file.
spring.kafka.consumer.ssl.key-store-location= # location of the key store file.
spring.kafka.consumer.ssl.key-store-password= # store password for the key store file.
spring.kafka.consumer.ssl.key-store-type= # type of the key store.
spring.kafka.consumer.ssl.protocol= # ssl protocol to use.
spring.kafka.consumer.ssl.trust-store-location= # location of the trust store file.
spring.kafka.consumer.ssl.trust-store-password= # store password for the trust store file.
spring.kafka.consumer.ssl.trust-store-type= # type of the trust store.
spring.kafka.consumer.value-deserializer= # deserializer class for values.
spring.kafka.jaas.control-flag=required # control flag for login configuration.
spring.kafka.jaas.enabled=false # whether to enable jaas configuration.
spring.kafka.jaas.login-module=com.sun.security.auth.module.krb5loginmodule # login module.
spring.kafka.jaas.options= # additional jaas options.
spring.kafka.listener.ack-count= # number of records between offset commits when ackmode is "count" or "count_time".
spring.kafka.listener.ack-mode= # listener ackmode. see the spring-kafka documentation.
spring.kafka.listener.ack-time= # time between offset commits when ackmode is "time" or "count_time".
spring.kafka.listener.client-id= # prefix for the listener's consumer client.id property.
spring.kafka.listener.concurrency= # number of threads to run in the listener containers.
spring.kafka.listener.idle-event-interval= # time between publishing idle consumer events (no data received).
spring.kafka.listener.log-container-config= # whether to log the container configuration during initialization (info level).
spring.kafka.listener.monitor-interval= # time between checks for non-responsive consumers. if a duration suffix is not specified, seconds will be used.
spring.kafka.listener.no-poll-threshold= # multiplier applied to "polltimeout" to determine if a consumer is non-responsive.
spring.kafka.listener.poll-timeout= # timeout to use when polling the consumer.
spring.kafka.listener.type=single # listener type.
spring.kafka.producer.acks= # number of acknowledgments the producer requires the leader to have received before considering a request complete.
spring.kafka.producer.batch-size= # default batch size.
spring.kafka.producer.bootstrap-servers= # comma-delimited list of host:port pairs to use for establishing the initial connections to the kafka cluster. overrides the global property, for producers.
spring.kafka.producer.buffer-memory= # total memory size the producer can use to buffer records waiting to be sent to the server.
spring.kafka.producer.client-id= # id to pass to the server when making requests. used for server-side logging.
spring.kafka.producer.compression-type= # compression type for all data generated by the producer.
spring.kafka.producer.key-serializer= # serializer class for keys.
spring.kafka.producer.properties.*= # additional producer-specific properties used to configure the client.
spring.kafka.producer.retries= # when greater than zero, enables retrying of failed sends.
spring.kafka.producer.ssl.key-password= # password of the private key in the key store file.
spring.kafka.producer.ssl.key-store-location= # location of the key store file.
spring.kafka.producer.ssl.key-store-password= # store password for the key store file.
spring.kafka.producer.ssl.key-store-type= # type of the key store.
spring.kafka.producer.ssl.protocol= # ssl protocol to use.
spring.kafka.producer.ssl.trust-store-location= # location of the trust store file.
spring.kafka.producer.ssl.trust-store-password= # store password for the trust store file.
spring.kafka.producer.ssl.trust-store-type= # type of the trust store.
spring.kafka.producer.transaction-id-prefix= # when non empty, enables transaction support for producer.
spring.kafka.producer.value-serializer= # serializer class for values.
spring.kafka.properties.*= # additional properties, common to producers and consumers, used to configure the client.
spring.kafka.ssl.key-password= # password of the private key in the key store file.
spring.kafka.ssl.key-store-location= # location of the key store file.
spring.kafka.ssl.key-store-password= # store password for the key store file.
spring.kafka.ssl.key-store-type= # type of the key store.
spring.kafka.ssl.protocol= # ssl protocol to use.
spring.kafka.ssl.trust-store-location= # location of the trust store file.
spring.kafka.ssl.trust-store-password= # store password for the trust store file.
spring.kafka.ssl.trust-store-type= # type of the trust store.
spring.kafka.streams.application-id= # kafka streams application.id property; default spring.application.name.
spring.kafka.streams.auto-startup=true # whether or not to auto-start the streams factory bean.
spring.kafka.streams.bootstrap-servers= # comma-delimited list of host:port pairs to use for establishing the initial connections to the kafka cluster. overrides the global property, for streams.
spring.kafka.streams.cache-max-size-buffering= # maximum memory size to be used for buffering across all threads.
spring.kafka.streams.client-id= # id to pass to the server when making requests. used for server-side logging.
spring.kafka.streams.properties.*= # additional kafka properties used to configure the streams.
spring.kafka.streams.replication-factor= # the replication factor for change log topics and repartition topics created by the stream processing application.
spring.kafka.streams.ssl.key-password= # password of the private key in the key store file.
spring.kafka.streams.ssl.key-store-location= # location of the key store file.
spring.kafka.streams.ssl.key-store-password= # store password for the key store file.
spring.kafka.streams.ssl.key-store-type= # type of the key store.
spring.kafka.streams.ssl.protocol= # ssl protocol to use.
spring.kafka.streams.ssl.trust-store-location= # location of the trust store file.
spring.kafka.streams.ssl.trust-store-password= # store password for the trust store file.
spring.kafka.streams.ssl.trust-store-type= # type of the trust store.
spring.kafka.streams.state-dir= # directory location for the state store.
spring.kafka.template.default-topic= # default topic to which messages are sent.