kafka第三次课！

Posted 2022-07-12 teayear

1，课程回顾
2，本章重点
springboot整合kafka
springcloud整合kafka
3，具体内容
3.1 springboot整合kafka
3.1.1 pom.xml添加jar


org.springframework.kafka
spring-kafka
2.8.1

com.alibaba fastjson 1.2.79 注意：此处使用的springboot版本为2.4.1 kafka是编写课件时最新版本2.6.6，不是任意版本都兼容 3.1.2 配置文件application.properties #Kafka集群配置，注意如果集群搭建时用的是名字 需要配置对应的主机名称 C:\\Windows\\System32\\drivers\\etc\\hosts

#生产者配置

#spring整合kafka配置
#连接集群配置
spring.kafka.bootstrap-servers=cluster1:9092,cluster2:9092,cluster3:9092

重试次数

spring.kafka.producer.retries=3

应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)

spring.kafka.producer.acks=-1

批量大小

spring.kafka.producer.batch-size=16384

提交延时

spring.kafka.producer.properties.linger.ms=10

当生产端积累的消息达到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.felix.kafka.producer.CustomizePartitioner

#自定义topic名称
topicName=topic-deptinfo

#消费者配置
#springboot整合 kafka
#消费者配置
#连接集群配置
spring.kafka.bootstrap-servers=cluster1:9092,cluster2:9092,cluster3:9092

默认的消费组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

#自定义topic名称
topicName=topic-deptinfo

3.1.3 生成者代码
package com.aaa.sbm.task;
import com.aaa.sbm.entity.Dept;
import com.aaa.sbm.service.DeptService;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**

• @ fileName:TimedSendDeptInfoTask
• @ description:
• @ author:zhz
• @ createTime:2022/1/13 9:37
• @ version:1.0.0
  /
  @Component //不在3层之内，交给IOC处理
  @EnableScheduling //开启定时任务
  @EnableAsync //开启异步处理 可以在任务方法上使用@Async 该方法多线程处理时，可以异步处理，提高执行效率
  @Slf4j
  public class TimedSendDeptInfoTask
  //spring用到了哪些涉及模式 模板模式（封装出来一个通用的工具模板，供你完成什么功能，简化整个操作流程）
  @Resource
  private KafkaTemplate kafkaTemplate;
  //juc包下线程安全的类，可以实现多线程同步自增
  private AtomicInteger atomicInteger =new AtomicInteger();
  //注入topic名称
  @Value(“$topicName”)
  private String topicN;
  / @Resource
  private DeptService deptService;/
  / @Resource
  private RestTemplate restTemplate; / //HttpClient
  /@Resource //模板模式
  private RedisTemplate redisTemplate;*/
  /**
  • 定制执行任务
  • 每隔3秒 使用多线程发送5条部门信息到kafka中
    /
    // cron=“秒 分 时 日 月 周” / 每 - 范围（3-10 每分钟的3秒到10秒每秒执行一次） , 选项 (3,10,15 每分钟第3秒第10秒第15秒执行)
    @Scheduled(cron = "/3 * * * * ?")
    public void timedExecute()
    //实例化固定线程池
    ExecutorService executorService = Executors.newFixedThreadPool(5);
    //lambda表达式 简化代码写法 ()->sendDeptInfo() 左边是参数 右边是执行业务
    /* executorService.execute(new Runnable()
    @Override
    public void run()
    sendDeptInfo();
    System.out.println(“1”);
    System.out.println(“2”);
    
    );/
    / executorService.execute(()->
    sendDeptInfo();
    System.out.println(“1”);
    System.out.println(“2”);
    );*/
    //启动5个线程执行
    executorService.execute(()->sendDeptInfo());
    executorService.execute(()->sendDeptInfo());
    executorService.execute(()->sendDeptInfo());
    executorService.execute(()->sendDeptInfo());
    executorService.execute(()->sendDeptInfo());
    //关闭线程池
    executorService.shutdown();
    
    /**
  • 发送部门信息
    */
    // @Async //异步处理，提高效率
    public void sendDeptInfo()
    log.info(“线程信息为：”+Thread.currentThread().getName()+“,正在执行。。。。。。。。。。。。。。”);
    int getAndIncrement = atomicInteger.getAndIncrement();
    Dept dept =new Dept(getAndIncrement,“dev”+getAndIncrement,“zz”+getAndIncrement);
    log.info(“要发送的部门信息为：”+dept);
    //发送部门信息到kafka中 一定要是字符串格式
    kafkaTemplate.send(topicN, JSON.toJSONString(dept));
    
    

3.1.4 消费者代码
package com.aaa.sbm.util;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.Optional;
/**

• @ fileName:KafkaConsumer

• @ description:工具类用来监控topic-deptinof，不停的获取message

• @ author:zhz

• @ createTime:2022/1/13 10:26

• @ version:1.0.0
  */
  @Component
  @Slf4j
  public class KafkaConsumer

  /**

  • 消费消息方法 借助 @KafkaListener指定消费的topic 如果该topic有信息都回被拉取pull 到参数中
  • @param record
    */
    @KafkaListener(topics = “$topicName”) //监听注解 监听指定的topic
    public void pullKafkaMsg(ConsumerRecord<?,?> record)
    //jdk8之后封装的专门处理空值一个类，有效防止空指针异常
    Optional<?> optional = Optional.ofNullable(record.value());
    // isPresent等同于if(record!=null)
    if(optional.isPresent())
    log.info(“接受到的信息为：”+ record);
    log.info(“接受到的部门信息为：”+ optional.get());
    
    
    

3.1.5 测试
消费消息
启动消费者项目，观察控制台
生产信息
直接启动生成者，观察控制台

3.2 springcloud整合kafka(以121讲课项目为例子)
3.2.1 添加jar
父项目:


org.springframework.kafka
spring-kafka
2.2.14.RELEASE

        <!-- fastjson的jar包-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.75</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.18</version>
        </dependency>

注意：这里的springboot2.1.11只可以和spring-kafka2.2.*的版本匹配，否则会报异常
微服务:

org.projectlombok
lombok



org.springframework.kafka
spring-kafka



com.alibaba
fastjson

3.2.2 生成者配置application.properties
图片: https://shimo.im/fake.png

#springboot 整合kafka
#Kafka集群配置，注意如果集群搭建时用的是名字 需要配置对应的主机名称 C:\\Windows\\System32\\drivers\\etc\\hosts
spring.kafka.bootstrap-servers=kafka1:9092,kafka2:9092,kafka3: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.felix.kafka.producer.CustomizePartitioner

3.2.3 消费者配置application.properties

#springboot 整合kafka
#Kafka集群配置，注意如果集群搭建时用的是名字 需要配置对应的主机名称 C:\\Windows\\System32\\drivers\\etc\\hosts
spring.kafka.bootstrap-servers=kafka1:9092,kafka2:9092,kafka3:9092

#消费者配置

默认的消费组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

3.2.4 生产者代码
图片: https://shimo.im/fake.png
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
//发送消息方法
@GetMapping(“productOrder”)
public String send()
Order order =new Order();
order.setId(100);
order.setMemberUsername(“测试生产者”);
order.setShopId(1001);
//log.info(“+++++++++++++++++++++ message = ”, JSON.toJSONString(dept));
//topic-dept为主题
kafkaTemplate.send(“topic-order”, JSON.toJSONString(order));
return “suc”;

3.2.5 消费者代码
图片: https://shimo.im/fake.png
package com.aaa.ss.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**

• @ fileName:KafkaConsumer

• @ description:

• @ author:zhz

• @ createTime:2021/2/20 17:20
  */
  @Component
  @Slf4j
  public class KafkaConsumer
  @KafkaListener(topics = “topic-order”)
  public void consumer(ConsumerRecord<?, ?> record)
  Optional<?> kafkaMessage = Optional.ofNullable(record.value());
  if (kafkaMessage.isPresent())
  Object message = kafkaMessage.get();
  log.info(“----------------- record =” + record);
  log.info(“------------------ message =” + message);
  
  
  

      3.2.6  测试
              1，生产者，地址栏请求（具体要看业务需求，讲课只位了测试效果）
               http://localhost:2221/order/productOrder
              2，观察消费者项目
  

图片: https://shimo.im/fake.png
3，还可以使用命令查看后台topic
图片: https://shimo.im/fake.png

4，知识点总结
5，本章面试题