Logo

菜单

legegecoder
legegecoder
发布于 2024-09-13 / 3 阅读 / 0 评论 / 0 点赞

springboot 引入kafka

单个kafaka

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

spring:
  kafka:
    consumer:
      bootstrap-servers:
        - 127.0.0.1:9092
      # 消费组
      group-id: myGroup
      # 消费者是否自动提交偏移量，默认为true
      enable-auto-commit: false
      # 消费者在读取一个没有偏移量或者偏移量无效的情况下，从起始位置读取partition的记录，默认是latest
      auto-offset-reset: earliest
      # 单次调用poll方法能够返回的消息数量
      max-poll-records: 10
    properties:
      # 两次poll之间的最大间隔，默认值为5分钟。如果超过这个间隔会触发reBalance
      max:
        poll:
          interval:
            ms: 600000

    producer:
      # Kafka服务器
      bootstrap-servers:
        - 127.0.0.1:9092
      retries: 1
      acks: all
      batch-size: 16384
      buffer-memory: 1024000
@Configuration
@Slf4j
public class Consumer {

    @KafkaListener(topics = "testInfoTopic", group = "myGroup")
    public void consume(String message) {
        System.out.println("接收到消息：" + message);
        log.info("正在为 " + message + " 办理注册业务...");
        log.info("注册成功");
    }

}


@RestController
@RequestMapping("/test")
@Slf4j
public class ProducerController {

    @Autowired
    private KafkaTemplate<String, String> kafka;


    @PostMapping("register")
    public String register(@RequestBody User user) {
        String message = JSON.toJSONString(user);
        log.info("接收到用户信息：" + message);
        kafka.send("testInfoTopic", message);
        return "OK";
    }

    @Data
    public static class User {
        private String username;
    }

}

多个



spring:
  kafka:
    # 第一个kafka的配置
    first:
      bootstrap-servers: 101:9092
      #生产者配置
      producer:
        # Kafka提供的序列化和反序列化类
        key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
        value-serializer: org.apache.kafka.common.serialization.StringSerializer
        retries: 1 # 消息发送重试次数
        #acks = 0：设置成 表示 producer 完全不理睬 leader broker 端的处理结果。此时producer 发送消息后立即开启下 条消息的发送，根本不等待 leader broker 端返回结果
        #acks= all 或者－1 ：表示当发送消息时， leader broker 不仅会将消息写入本地日志，同时还会等待所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给，消息安全但是吞吐量会比较低。
        #acks = 1：默认的参数值。 producer 发送消息后 leader broker 仅将该消息写入本地日志，然后便发送响应结果给producer ，而无须等待其他副本写入该消息。折中方案，只要leader一直活着消息就不会丢失，同时也保证了吞吐量
        acks: 1 #应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
        batch-size: 16384 #批量大小
        properties:
          linger:
            ms: 0 #提交延迟
        buffer-memory: 33554432 # 生产端缓冲区大小
      # 消费者配置
      consumer:
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        # 默认分组id
        group-id: DEMO_${random.uuid}
        enable-auto-commit: true
        #提交offset延时(接收到消息后多久提交offset)
        auto-commit-interval: 100ms
        #当kafka中没有初始offset或offset超出范围时将自动重置offset
        # earliest:重置为分区中最小的offset;
        # latest:重置为分区中最新的offset(消费分区中新产生的数据);
        # none:只要有一个分区不存在已提交的offset,就抛出异常;
        auto-offset-reset: latest
        #批量消费每次最多消费多少条消息
        #每次拉取一条，一条条消费，当然是具体业务状况设置
        max-poll-records: 1
    # 第二个kafka的配置
    second:
      bootstrap-servers: 110:9092
      #生产者配置
      producer:
        # Kafka提供的序列化和反序列化类
        key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
        value-serializer: org.apache.kafka.common.serialization.StringSerializer
        retries: 1 # 消息发送重试次数
        #acks = 0：设置成 表示 producer 完全不理睬 leader broker 端的处理结果。此时producer 发送消息后立即开启下 条消息的发送，根本不等待 leader broker 端返回结果
        #acks= all 或者－1 ：表示当发送消息时， leader broker 不仅会将消息写入本地日志，同时还会等待所有其他副本都成功写入它们各自的本地日志后，才发送响应结果给，消息安全但是吞吐量会比较低。
        #acks = 1：默认的参数值。 producer 发送消息后 leader broker 仅将该消息写入本地日志，然后便发送响应结果给producer ，而无须等待其他副本写入该消息。折中方案，只要leader一直活着消息就不会丢失，同时也保证了吞吐量
        acks: 1 #应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
        batch-size: 16384 #批量大小
        properties:
          linger:
            ms: 0 #提交延迟
        buffer-memory: 33554432 # 生产端缓冲区大小
      # 消费者配置
      consumer:
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        # 默认分组id
        group-id: DEMO_${random.uuid}
        enable-auto-commit: true
        #提交offset延时(接收到消息后多久提交offset)
        auto-commit-interval: 100ms
        #当kafka中没有初始offset或offset超出范围时将自动重置offset
        # earliest:重置为分区中最小的offset;
        # latest:重置为分区中最新的offset(消费分区中新产生的数据);
        # none:只要有一个分区不存在已提交的offset,就抛出异常;
        auto-offset-reset: latest
        #批量消费每次最多消费多少条消息
        #每次拉取一条，一条条消费，当然是具体业务状况设置
        max-poll-records: 1


FirstKafkaConfig.java
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 
/**
 * 第一个kafka配置
 *
 */
@Configuration
public class FirstKafkaConfig {
 
    /**
     * 读取第一个kafka配置
     * Primary注解表示默认以这个为准
     *
     * @return 第一个kafka配置
     */
    @Primary
    @ConfigurationProperties(prefix = "spring.kafka.first")
    @Bean
    public KafkaProperties firstKafkaProperties() {
        return new KafkaProperties();
    }
 
    /**
     * 构建第一个kafka的生产者发送template
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的生产者发送template
     */
    @Primary
    @Bean
    public KafkaTemplate<String, String> firstKafkaTemplate(
            @Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
        return new KafkaTemplate<>(firstProducerFactory(firstKafkaProperties));
    }
 
    /**
     * 构建第一个kafka的消费者监听容器工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的消费者监听容器工厂
     */
    @Bean("firstKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    firstKafkaListenerContainerFactory(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(firstConsumerFactory(firstKafkaProperties));
        return factory;
    }
 
    /**
     * 新建第一个kafka的消费者工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的消费者工厂
     */
    private ConsumerFactory<? super Integer, ? super String> firstConsumerFactory(KafkaProperties firstKafkaProperties) {
        return new DefaultKafkaConsumerFactory<>(firstKafkaProperties.buildConsumerProperties());
    }
 
    /**
     * 新建第一个kafka的生产者工厂
     *
     * @param firstKafkaProperties 第一个kafka配置
     * @return 第一个kafka的生产者工厂
     */
    private DefaultKafkaProducerFactory<String, String> firstProducerFactory(KafkaProperties firstKafkaProperties) {
        return new DefaultKafkaProducerFactory<>(firstKafkaProperties.buildProducerProperties());
    }
}
 
 

 
SecondKafkaConfig.java
 

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 
/**
 * 第二个kafka配置
 *
 */
@Configuration
public class SecondKafkaConfig {
 
    /**
     * 读取第二个kafka配置
     *
     * @return 第二个kafka配置
     */
    @ConfigurationProperties(prefix = "spring.kafka.second")
    @Bean
    public KafkaProperties secondKafkaProperties() {
        return new KafkaProperties();
    }
 
    /**
     * 构建第二个kafka的生产者发送template
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的生产者发送template
     */
    @Bean
    public KafkaTemplate<String, String> secondKafkaTemplate(
            @Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
        return new KafkaTemplate<>(secondProducerFactory(secondKafkaProperties));
    }
 
    /**
     * 构建第二个kafka的消费者监听容器工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的消费者监听容器工厂
     */
    @Bean("secondKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
    secondKafkaListenerContainerFactory(@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(secondConsumerFactory(secondKafkaProperties));
        return factory;
    }
 
    /**
     * 新建第二个kafka的消费者工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的消费者工厂
     */
    private ConsumerFactory<? super Integer, ? super String> secondConsumerFactory(KafkaProperties secondKafkaProperties) {
        return new DefaultKafkaConsumerFactory<>(secondKafkaProperties.buildConsumerProperties());
    }
 
    /**
     * 新建第二个kafka的生产者工厂
     *
     * @param secondKafkaProperties 第二个kafka配置
     * @return 第二个kafka的生产者工厂
     */
    private DefaultKafkaProducerFactory<String, String> secondProducerFactory(KafkaProperties secondKafkaProperties) {
        return new DefaultKafkaProducerFactory<>(secondKafkaProperties.buildProducerProperties());
    }
}
 
 
 
 
/**
     * 第二个kafka配置需要指定下名字 
     */
    @Resource(name = "secondKafkaTemplate")
    private KafkaTemplate<String, String> secondKafkaTemplate;
 
    /**
     * 测试消费者
     * 这里要消费哪一个kafka消息，containerFactory就需要配成上面相对应的消费者监听容器工厂
     *
     * @param record 消息
     */
    @KafkaListener(
            containerFactory = "firstKafkaListenerContainerFactory",
            topics = {"orip"},
            groupId = "first-consumer")
    public void testConsumer(ConsumerRecord<?, ?> record) {
        String value = record.value() + "";
        System.out.println();
    }

写在前面
Mybatis-Plus 相关踩坑

评论