kafka与springboot 版本冲突问题
1.问题描述：
kafka服务器的版本 需要与我们springboot 版本对应，否则由于kafka版本差异，发送消息时可能出现问题；
kafka 与 springboot 对应关系：
官网：kafka 与 springboot 对应关系.
更老版本的对应如下图：
简而言之：kafka服务器安装的版本 与 我们使用的springboot 版本要对应；不对应就可能出问题，发消息报错、消息发出去主题里没有呀。。。。。。
2.解决方案：
- 更换kafka服务器的版本，与springboot版本匹配；
- 更换springboot版本，与kafka服务器安装的kafka版本匹配；
- springboot项目中，使用 kafka生产者时，使用原生API发送消息；监听的消费者，使用springboot的监听者；
方案1和方案2 代价较大，方案3 实测即可不受版本影响；
3.实际操作：
3.1 yml 配置：
kafka:
bootstrap-servers: 000.000.000.001:9092,000.000.000.002:9092 #<可根据配置修改>
producer:
# 重试次数，默认Integer.MAX_VALUE
retries: 3
# 同一批次内存大小（默认16K）
batch-size: 16384
# 生产者内存缓存区大小(32M)
buffer-memory: 33554432
#ack 0 不等待队列确认； ack 1等待leader 确认； ack -1 等待所有节点确认完成；
acks: 1
consumer:
group-id: test #可根据需求修改
# 关闭自动提交，消费后手动提交
enable-auto-commit: false
# earliest：从头开始消费 latest：从最新的开始消费 默认latest
auto-offset-reset: latest
# key和value反序列化（默认，可以不设置）
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
# # 消费者并发能力
concurrency: 1 #默认值
# # 设置手动提交的时候，需要设置ackMode
ack-mode: MANUAL
3.2 生产者：
3.2.1 配置 KafkaProducer：
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
import java.util.Properties;
//获取配置参数；
@Configuration
public class KafkaConfig {
@Value("#{'${spring.kafka.bootstrap-servers}'.split(',')}")
private List<String> bootstrapServers;
@Value("${spring.kafka.producer.retries}")
private int retries;
@Value("${spring.kafka.producer.batch-size}")
private int batchSize;
@Value("${spring.kafka.producer.buffer-memory}")
private int bufferMemory;
@Value("${spring.kafka.producer.acks}")
private String acks;
@Bean
public KafkaProducer<String, String> initKafkaTemplate() {
Properties props = new Properties();
//设置接入点，请通过控制台获取对应Topic的接入点
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//Kafka消息的序列化方式
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(ProducerConfig.ACKS_CONFIG,acks);
//构造Producer对象，注意，该对象是线程安全的，一般来说，一个进程内一个Producer对象即可；
//如果想提高性能，可以多构造几个对象，但不要太多，最好不要超过5个
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
return producer;
}
}
3.2.2 配置 MessageSenderClient ：
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.concurrent.Future;
/** * kafka 配置 * 由于kafka服务端使用的是 0.10.2版本，而我们的框架使用的spring boot版本为2.5.0， * kafka starter版本中kafka-clients版本为2.7.1；kafka客户端作为生产者发送消息 * 时，会增加请求头，而服务端不支持，导致无法正常生产消息，故kafka作为生产者，不能直 * 接使用boot starter，需使用kafka原生生产者; */
@Slf4j
@Component
public class MessageSenderClient {
@Resource
private KafkaProducer kafkaProducer;
/** * 消息发送至 kafka * * @param topic 路由 * @param data 消息内容 */
public void send(String topic, Object data) {
if (StringUtils.isEmpty(topic) || data == null) {
throw new IllegalStateException("The send message parameter cannot be null");
}
try {
Gson gson=new Gson();
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic,gson.toJson(data));
Future<RecordMetadata> metadataFuture = kafkaProducer.send(kafkaMessage);
RecordMetadata recordMetadata = metadataFuture.get();
log.info("MessageSenderClient kafka send Produce ok:{}", gson.toJson(data));
} catch (Exception e) {
log.info("MessageSenderClient kafka send error,", e);
}
}
/** * 消息发送至 kafka * * @param topic * @param partition 分区 * @param data */
public void send(String topic, Integer partition, Object data) {
if (StringUtils.isEmpty(topic) || data == null) {
throw new IllegalStateException("The send message parameter cannot be null");
}
if (partition == null) {
send(topic, data);
}
try {
Gson gson=new Gson();
ProducerRecord<String, String> kafkaMessage = new ProducerRecord<>(topic, partition, null,gson.toJson(data));
Future<RecordMetadata> metadataFuture = kafkaProducer.send(kafkaMessage);
RecordMetadata recordMetadata = metadataFuture.get();
log.info("MessageSenderClient kafka send Produce ok:{}", recordMetadata.toString());
} catch (Exception e) {
log.info("MessageSenderClient kafka send error,", e);
}
}
}
3.2.3 发送消息示例 ：
@Slf4j
@SpringBootTest
public class KafkaTest {
@Autowired
private MessageSenderClient kafkaProducer;
/** * 发消息 */
@Test
void testKafkaProducer() {
kafkaProducer.send("GroupQueue", "payment发送一条新消息");
}
}
结果：
发送成功；
3.3 消费者
3.3.1 消费者示例：
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
/** * kafka 监听者 */
@Slf4j
@Component
public class KafkaConsumerTest {
@KafkaListener(topics = { "GroupQueue"})
void onMessage1(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
// 消费的哪个topic、partition的消息,打印出消息内容
System.out.println("kafka consumer：" + record.topic() + "-" + record.partition() + "-" + record.value());
String value = record.value();
System.out.println("value:"+value);
/** * 业务处理代码。。。。。。 */
//处理完成后，手动提交ack
acknowledgment.acknowledge();
}
}
3.3.2 消费者监听测试：
注意：由于我们kafka配置的从最新的开始消费，所以测试时，记得先启动消费者开启监听，生产者再发次消息；消费者才能收到最新的那条消息；
结果ok:
4.小结：
本文不如不当之处，请各位大佬及时指正；
