Bagaimana SpringBoot menyepadukan kelas alat konfigurasi Kafka

spring-kafka adalah berdasarkan integrasi klien kafka versi Java dan spring Ia menyediakan KafkaTemplate, yang merangkumi pelbagai kaedah untuk operasi mudah Ia merangkum klien kafka apache dan tidak perlu mengimport kebergantungan klien

.

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

Konfigurasi YML

kafka:
    #bootstrap-servers: server1:9092,server2:9093 #kafka开发地址,
    #生产者配置
    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
      # 分组名称
      group-id: web
      enable-auto-commit: false
      #提交offset延时(接收到消息后多久提交offset)
      # auto-commit-interval: 1000ms
      #当kafka中没有初始offset或offset超出范围时将自动重置offset
      # earliest:重置为分区中最小的offset;
      # latest:重置为分区中最新的offset(消费分区中新产生的数据);
      # none:只要有一个分区不存在已提交的offset,就抛出异常;
      auto-offset-reset: latest
      properties:
        #消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
        session.timeout.ms: 15000
        #消费请求超时时间
        request.timeout.ms: 18000
      #批量消费每次最多消费多少条消息
      #每次拉取一条，一条条消费，当然是具体业务状况设置
      max-poll-records: 1
      # 指定心跳包发送频率，即间隔多长时间发送一次心跳包，优化该值的设置可以减少Rebalance操作，默认时间为3秒；
      heartbeat-interval: 6000
      # 发出请求时传递给服务器的 ID。用于服务器端日志记录 正常使用后解开注释，不然只有一个节点会报错
      #client-id: mqtt
    listener:
      #消费端监听的topic不存在时，项目启动会报错(关掉)
      missing-topics-fatal: false
      #设置消费类型 批量消费 batch，单条消费：single
      type: single
      #指定容器的线程数，提高并发量
      #concurrency: 3
      #手动提交偏移量 manual达到一定数据后批量提交
      #ack-mode: manual
      ack-mode: MANUAL_IMMEDIATE #手動確認消息
        # 认证
    #properties:
      #security:
        #protocol: SASL_PLAINTEXT
      #sasl:
        #mechanism: SCRAM-SHA-256
        #jaas:config: 'org.apache.kafka.common.security.scram.ScramLoginModule required username="username" password="password";'

Kelas alat mudah yang boleh memenuhi penggunaan biasa, tema tidak boleh diubah suai

@Component
@Slf4j
public class KafkaUtils<K, V> {
    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] servers;

    /**
     * 获取连接
     * @return
     */
    private Admin getAdmin() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        // 正式环境需要添加账号密码
        return Admin.create(properties);
    }

    /**
     * 增加topic
     *
     * @param name      主题名字
     * @param partition 分区数量
     * @param replica   副本数量
     * @date 2022-06-23 chens
     */
    public R addTopic(String name, Integer partition, Integer replica) {
        Admin admin = getAdmin();
        if (replica > servers.length) {
            return R.error("副本数量不允许超过Broker数量");
        }
        try {
            NewTopic topic = new NewTopic(name, partition, Short.parseShort(replica.toString()));
            admin.createTopics(Collections.singleton(topic));
        } finally {
            admin.close();
        }
        return R.ok();
    }

    /**
     * 删除主题
     *
     * @param names 主题名字集合
     * @date 2022-06-23 chens
     */
    public void deleteTopic(List<String> names) {
        Admin admin = getAdmin();
        try {
            admin.deleteTopics(names);
        } finally {
            admin.close();
        }
    }

    /**
     * 查询所有主题
     *
     * @date 2022-06-24 chens
     */
    public Set<String> queryTopic() {
        Admin admin = getAdmin();
        try {
            ListTopicsResult topics = admin.listTopics();
            Set<String> set = topics.names().get();
            return set;
        } catch (Exception e) {
            log.error("查询主题错误！");
        } finally {
            admin.close();
        }
        return null;
    }

    // 向所有分区发送消息
    public ListenableFuture<SendResult<K, V>> send(String topic, @Nullable V data) {
        return kafkaTemplate.send(topic, data);
    }
    
    // 指定key发送消息，相同key保证消息在同一个分区
    public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, key, data);
    }

    // 指定分区和key发送。
    public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, K key, @Nullable V data) {
        return kafkaTemplate.send(topic, partition, key, data);
    }
}

Hantar mesej Gunakan asynchronous

@GetMapping("/{topic}")
    public String test(@PathVariable String topic, @PathVariable Long index) throws ExecutionException, InterruptedException {

        ListenableFuture future = null;
        Chenshuang user = new Chenshuang(i, "陈爽", "123456", new Date());
        String s = JSON.toJSONString(user);
        KafkaUtils utils = new KafkaUtils();
        future = kafkaUtils.send(topic, s);
        // 异步回调，同步get,会等待 不推荐同步！
        future.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送失败");
            }
            @Override
            public void onSuccess(Object result) {
                System.out.println("发送成功：" + result);
            }
        });
        return "发送成功";
    }

untuk mencipta topik

Jika broker mengkonfigurasi auto.create.topics.enable menjadi benar (lalai adalah benar), ia akan digunakan apabila menerima permintaan metadata daripada klien Cipta topik.

Menghantar dan menggunakan topik yang tidak wujud akan mencipta topik baharu Dalam kebanyakan kes, penciptaan topik yang tidak dijangka akan membawa kepada banyak masalah yang tidak dijangka.

Topik topik digunakan untuk membezakan jenis mesej yang berbeza, sebenarnya, ia sesuai untuk senario perniagaan yang berbeza Secara lalai, mesej disimpan selama satu minggu

Di bawah tema Topik yang sama. lalai ialah partition. Maksudnya, hanya ada satu pengguna untuk penggunaan Jika anda ingin meningkatkan kapasiti penggunaan, anda perlu menambah partition tiga cara untuk mengedarkan mesej (kunci, nilai) kepada Partition yang berbeza, nyatakan partition, laluan HASH, lalai, ID mesej dalam partition yang sama adalah unik dan teratur

Apabila pengguna menggunakan mesej dalam partition partition , mereka menggunakan offset untuk mengenal pasti lokasi mesej;

GroupId digunakan untuk menyelesaikan masalah penggunaan berulang di bawah Topik yang sama Contohnya, jika penggunaan perlu diterima oleh berbilang pengguna, ia boleh dicapai dengan menetapkan GroupId yang berbeza

Mesej sebenar disimpan dalam satu salinan. Ia hanya dibezakan dengan menetapkan logo secara logik mengenal pasti sama ada ia telah dimakan.

Ketersediaan tinggi untuk menghantar mesej—

Mod kluster, pelaksanaan berbilang salinan; penghantaran berjaya ; Apabila =1, induk hanya akan bertindak balas apabila =semua, lebih separuh daripada jawapan akan OK (ketersediaan tinggi sebenar)

Ketersediaan tinggi mesej yang digunakan—

<.>Pengenalan automatik boleh dimatikan mod Offsert, mula-mula tarik mesej, selepas penggunaan selesai, kemudian tetapkan kedudukan offset untuk menyelesaikan ketersediaan tinggi penggunaan

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaTopic {
    // yml自定义主题，项目启动就创建，
    @Value("${spring.kafka.topic}")
    String topic;
    @Value("${spring.kafka.bootstrap-servers}")
    String[] server;
    /**
     * 项目启动 初始化主题,如果存在不会覆盖主题的
     */
    @Bean
    public NewTopic batchTopic() {
        // 最大复制因子 <= 经纪人broker数量.
        return new NewTopic(topic, 10, (short) server.length);
    }
}

Kelas mendengar, satu mesej, hanya satu pengguna dalam setiap kumpulan menggunakannya sekali, jika Mesej berada dalam zon 1. Jika anda menentukan partition 1 untuk memantau, ia juga akan digunakan

Anda juga boleh memantau topik yang berbeza dengan kaedah yang sama dan menentukan monitor anjakan

The kumpulan yang sama akan mengambil sama rata, dan kumpulan yang berbeza akan mengambil berulang kali.

1 Mod Unicast, hanya terdapat satu kumpulan pengguna

(1) Topik hanya mempunyai satu partition, dan apabila terdapat berbilang pengguna dalam kumpulan, mesej dalam partition yang sama akan Hanya boleh dimakan oleh seorang pengguna dalam kumpulan. Apabila bilangan pengguna melebihi bilangan partition, lebihan pengguna akan terbiar, seperti yang ditunjukkan dalam Rajah 1. Topik dan ujian hanya mempunyai satu partition dan hanya satu kumpulan, G1 Terdapat berbilang pengguna dalam kumpulan ini dan hanya boleh digunakan oleh salah seorang daripada mereka, manakala yang lain melahu.

Rajah 1

(2) Topik mempunyai berbilang partition dan terdapat berbilang pengguna dalam kumpulan Contohnya, test mempunyai 3 partition adalah dua pengguna dalam kumpulan, maka C0 mungkin sepadan dengan menggunakan data dalam p0 dan p1, dan c1 sepadan dengan menggunakan data dalam p2, jika terdapat tiga pengguna, seorang pengguna sepadan dengan menggunakan data dalam satu partition. Gambar rajah ditunjukkan dalam Rajah 2 dan Rajah 3. Mod ini sangat biasa dalam mod kluster Sebagai contoh, kita boleh memulakan 3 perkhidmatan dan menetapkan 3 partition untuk topik yang sepadan, supaya penggunaan selari dapat dicapai dan kecekapan pemprosesan mesej. boleh dipertingkatkan dengan ketara.

Gambar 2

Gambar 3

2 >

Jika anda ingin melaksanakan mod siaran, anda perlu menyediakan beberapa kumpulan pengguna, supaya selepas satu kumpulan pengguna menggunakan mesej, ia tidak akan menjejaskan penggunaan pengguna dalam kumpulan lain sama sekali daripada penyiaran.

(1) Berbilang kumpulan pengguna, 1 partition

Data dalam topik ini digunakan oleh berbilang kumpulan pengguna pada masa yang sama Apabila kumpulan pengguna mempunyai berbilang pengguna Ia hanya boleh digunakan oleh satu pengguna, seperti yang ditunjukkan dalam Rajah 4:

Rajah 4

(2) Berbilang kumpulan pengguna, berbilang partition

Data dalam topik ini boleh digunakan berbilang kali oleh berbilang kumpulan pengguna Dalam kumpulan pengguna, setiap pengguna boleh menggunakan secara selari dengan satu atau lebih partition dalam topik, seperti yang ditunjukkan dalam rajah Lima:

<.>

注意： 消费者的数量并不能决定一个topic的并行度。它是由分区的数目决定的。
再多的消费者，分区数少，也是浪费！
一个组的最大并行度将等于该主题的分区数。

@Component
@Slf4j
public class Consumer {
    // 监听主题 分组a
    @KafkaListener(topics =("${spring.kafka.topic}") ,groupId = "a")
    public  void  getMessage(ConsumerRecord message, Acknowledgment ack) {
        //确认收到消息
        ack.acknowledge();
    }
    // 监听主题 分组a
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "a")
    public  void getMessage2(ConsumerRecord message, Acknowledgment ack) {
        //确认收到消息
        ack.acknowledge();
    }
    // 监听主题 分组b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage3(ConsumerRecord message, Acknowledgment ack) {
        //确认收到消息//确认收到消息
        ack.acknowledge();
    }
    // 监听主题 分组b
    @KafkaListener(topics = ("${spring.kafka.topic}"),groupId = "b")
    public  void getMessage4(ConsumerRecord message, Acknowledgment ack) {
        //确认收到消息//确认收到消息
        ack.acknowledge();
    }

    // 指定监听分区1的消息
    @KafkaListener(topicPartitions = {@TopicPartition(topic = ("${spring.kafka.topic}"),partitions = {"1"})})
    public void getMessage5(ConsumerRecord message, Acknowledgment ack) {
        Long id = JSONObject.parseObject(message.value().toString()).getLong("id");
        //确认收到消息//确认收到消息
        ack.acknowledge();
    }
    
    /**
     * @Title 指定topic、partition、offset消费
     * @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
     * 注意：topics和topicPartitions不能同时使用；
     **/
    @KafkaListener(id = "c1",groupId = "c",topicPartitions = {
            @TopicPartition(topic = "t1", partitions = { "0" }),
            @TopicPartition(topic = "t2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))})
    public void getMessage6(ConsumerRecord record,Acknowledgment ack) {
        //确认收到消息
        ack.acknowledge();
    }
    
    /**        
     * 批量消费监听goods变更消息
     * yml配置listener:type 要改为batch
     * ymk配置consumer：max-poll-records: ？？（每次拉取多少条数据消费）
     * concurrency = "2" 启动多少线程执行，应小于等于broker数量，避免资源浪费
     */
    @KafkaListener(id="sync-modify-goods", topics = "${spring.kafka.topic}",concurrency = "4")
    public void getMessage7(List<ConsumerRecord<String, String>> records){
        for (ConsumerRecord<String, String> msg:records) {
            GoodsChangeMsg changeMsg = null;
            try {
                changeMsg = JSONObject.parseObject(msg.value(), GoodsChangeMsg.class);
                syncGoodsProcessor.handle(changeMsg);
            }catch (Exception exception) {
                log.error("解析失败{}", msg, exception);
            }
        }
    }
}

Atas ialah kandungan terperinci Bagaimana SpringBoot menyepadukan kelas alat konfigurasi Kafka. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!