kafka---springboot

Posted 2022-09-19 Raabbit_Cunk

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了kafka---springboot相关的知识，希望对你有一定的参考价值。

消息队列的两种模式

1.点对点
这种跟rabbitmq很像，消费者监听消息队列（但是 rabbitmq 可以设置ack 不删除数据…）

2.发布订阅
rocketmq 就是这样的，（主题标签），但是rocketmq 消费者是轮询收到消息的，也可以采取广播.

kafka 分区

分区策略

在生成环境中，将订单发送到某一个分区，可以只写key，这样都发送到一个分区了
消息不符合规定的也可以不发送

docker 安装kafka

添加链接描述

操作kafka参数

异步消息发送

依赖

   <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>

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

消费者


@Component
public class KafkaConsumer 

    @KafkaListener(topics = "hello-world")
    public void onoMsg(ConsumerRecord<?,?> record) 
        System.out.println("简单消费： " + record.topic() + "-" + record.partition() + "- " + record.value());

同步异步发送消息

生产者

    @Test
    void contextLoads() 
        // 创建生产者 发送数据
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"43.143.30.231:9092") ;
       properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName() );
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // kafka 发送数据
        kafkaProducer.send(new ProducerRecord<>("hello-world","hello~~~"));

        // kafka 异步回调发送数据
        kafkaProducer.send(new ProducerRecord<>("hello-world", "hello~~~"), new Callback() 
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) 
                System.out.println("主题是："+recordMetadata.topic()+" 的消息发送成功哦~~");
            

        ) ;

        System.out.println("主线程正在运行 ");

        kafkaProducer.close();

同步发送


        for (int i = 0; i < 3; i++) 
            // 生产者 发送消息加上 .get() 就是同步发送消息了
            kafkaProducer.send(new ProducerRecord<>("hello-world", "hello~~~"), new Callback() 
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) 
                    System.out.println("消息发送成功");
                
            ).get();
        
        
        System.out.println("主线程正在运行 ");

res:

消息发送成功
消息发送成功
消息发送成功
主线程正在运行

自定义分区器


// 自定义分区策略
public class MyPartitioner implements Partitioner 

    @Override
    public int partition(String topic, Object key, byte[] bytes, Object value, byte[] bytes1, Cluster cluster) 
        String msgValue = value.toString();
        // 将value包含 hello 的消息发送 0 号分区

        if (msgValue.contains("hello"))
            System.out.println("消息发送 0 号分区 ");
            return 0 ;
        

        return 0;
    



    @Override
    public void configure(Map<String, ?> map) 

    

    @Override
    public void close()

生产者配置分区器

 // 自定义分区器
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.example.kafka.config.MyPartitioner") ;

提高生产者吞吐量

springboot配置

###########【Kafka集群】###########
spring.kafka.bootstrap-servers=43.143.30.231:9092 
###########【初始化生产者配置】###########
# 重试次数
spring.kafka.producer.retries=0
# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
spring.kafka.producer.acks=1
# 批量大小
spring.kafka.producer.batch-size=16384
# 提交延时
spring.kafka.producer.properties.linger.ms=0
# 当生产端积累的消息达到batch-size或接收到消息linger.ms后,生产者就会将消息提交给kafka
# linger.ms为0表示每接收到一条消息就提交给kafka,这时候batch-size其实就没用了

# 生产端缓冲区大小
spring.kafka.producer.buffer-memory = 33554432
# Kafka提供的序列化和反序列化类
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
# 自定义分区器
# spring.kafka.producer.properties.partitioner.class=com.example.kafka.config.MyPartitioner

###########【初始化消费者配置】###########
# 默认的消费组ID
spring.kafka.consumer.properties.group.id=defaultConsumerGroup
# 是否自动提交offset
spring.kafka.consumer.enable-auto-commit=true
# 提交offset延时(接收到消息后多久提交offset)
spring.kafka.consumer.auto.commit.interval.ms=1000
# 当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
spring.kafka.consumer.auto-offset-reset=latest
# 消费会话超时时间(超过这个时间consumer没有发送心跳,就会触发rebalance操作)
spring.kafka.consumer.properties.session.timeout.ms=120000
# 消费请求超时时间
spring.kafka.consumer.properties.request.timeout.ms=180000
# Kafka提供的序列化和反序列化类
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消费端监听的topic不存在时，项目启动会报错(关掉)
spring.kafka.listener.missing-topics-fatal=false
# 设置批量消费
# spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
# spring.kafka.consumer.max-poll-records=50

消息可靠性

0 ：可靠性低，不适用，但是保证消息不会重复发送
-1：效率低，配上充实策略有可能导致消息重复发送

对故障节点采取降级（服务降级一般是从整体系统负荷考虑，由调用方控制）

ack =1 降级为 0 的起因

##== 数据完全可靠性==

ack总结

数据去重

消息幂等性（精确发送一次，这是rocketmq 和rabbitmq 所没有的【数据库防重】）

生产者事务

开启幂等性才能开启事务，幂等性默认开启

事务消息会造成异步发送消息变成同步发送消息


       // 指定事务id
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transactional_id_01") ;

       KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 开启事务  必须指定事务 id , 此时消息不能发送消息异步！！ 

        kafkaProducer.initTransactions();
        kafkaProducer.beginTransaction();


       try 
           // kafka 异步回调发送数据
           kafkaProducer.send(new ProducerRecord<>("hello-world", "hello~~~"), new Callback() 
               @SneakyThrows
               @Override
               public void onCompletion(RecordMetadata recordMetadata, Exception e) 
                  Thread.sleep(2000);
                   System.out.println("主题是："+recordMetadata.topic()+" 的消息发送成功哦~~"+"分区是："+recordMetadata.partition());
               
           ) ;

           // 提交事务
           kafkaProducer.commitTransaction();
       catch (Exception e)
         // 出错 事务回滚
           kafkaProducer.abortTransaction();
       finally 
           kafkaProducer.close();
       

        System.out.println("主线程正在运行 ");