kafka第三次课！

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:WindowsSystem32driversetchosts

#生产者配置

#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;

//启动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;

@Component
@Slf4j
public class KafkaConsumer {

@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

        
        
            com.alibaba
            fastjson
            1.2.75
        
        
        
            org.projectlombok
            lombok
            1.18.18

注意：这里的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:WindowsSystem32driversetchosts
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

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

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 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;

@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，本章面试题

kafka第三次课！

Java相关栏目本月热门文章