kafka第三次课！

水深无声 2023-09-29 16:43 7阅读 0赞

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

<dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.75</version>
            </dependency>
            
            <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

\#消费者配置

## 默认的消费组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，本章面试题