单个kafaka
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
spring:
kafka:
consumer:
bootstrap-servers:
- 127.0.0.1:9092
# 消费组
group-id: myGroup
# 消费者是否自动提交偏移量,默认为true
enable-auto-commit: false
# 消费者在读取一个没有偏移量或者偏移量无效的情况下,从起始位置读取partition的记录,默认是latest
auto-offset-reset: earliest
# 单次调用poll方法能够返回的消息数量
max-poll-records: 10
properties:
# 两次poll之间的最大间隔,默认值为5分钟。如果超过这个间隔会触发reBalance
max:
poll:
interval:
ms: 600000
producer:
# Kafka服务器
bootstrap-servers:
- 127.0.0.1:9092
retries: 1
acks: all
batch-size: 16384
buffer-memory: 1024000
@Configuration
@Slf4j
public class Consumer {
@KafkaListener(topics = "testInfoTopic", group = "myGroup")
public void consume(String message) {
System.out.println("接收到消息:" + message);
log.info("正在为 " + message + " 办理注册业务...");
log.info("注册成功");
}
}
@RestController
@RequestMapping("/test")
@Slf4j
public class ProducerController {
@Autowired
private KafkaTemplate<String, String> kafka;
@PostMapping("register")
public String register(@RequestBody User user) {
String message = JSON.toJSONString(user);
log.info("接收到用户信息:" + message);
kafka.send("testInfoTopic", message);
return "OK";
}
@Data
public static class User {
private String username;
}
}
多个
spring:
kafka:
# 第一个kafka的配置
first:
bootstrap-servers: 101:9092
#生产者配置
producer:
# Kafka提供的序列化和反序列化类
key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
value-serializer: org.apache.kafka.common.serialization.StringSerializer
retries: 1 # 消息发送重试次数
#acks = 0:设置成 表示 producer 完全不理睬 leader broker 端的处理结果。此时producer 发送消息后立即开启下 条消息的发送,根本不等待 leader broker 端返回结果
#acks= all 或者-1 :表示当发送消息时, leader broker 不仅会将消息写入本地日志,同时还会等待所有其他副本都成功写入它们各自的本地日志后,才发送响应结果给,消息安全但是吞吐量会比较低。
#acks = 1:默认的参数值。 producer 发送消息后 leader broker 仅将该消息写入本地日志,然后便发送响应结果给producer ,而无须等待其他副本写入该消息。折中方案,只要leader一直活着消息就不会丢失,同时也保证了吞吐量
acks: 1 #应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
batch-size: 16384 #批量大小
properties:
linger:
ms: 0 #提交延迟
buffer-memory: 33554432 # 生产端缓冲区大小
# 消费者配置
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 默认分组id
group-id: DEMO_${random.uuid}
enable-auto-commit: true
#提交offset延时(接收到消息后多久提交offset)
auto-commit-interval: 100ms
#当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
auto-offset-reset: latest
#批量消费每次最多消费多少条消息
#每次拉取一条,一条条消费,当然是具体业务状况设置
max-poll-records: 1
# 第二个kafka的配置
second:
bootstrap-servers: 110:9092
#生产者配置
producer:
# Kafka提供的序列化和反序列化类
key-serializer: org.apache.kafka.common.serialization.StringSerializer #序列化
value-serializer: org.apache.kafka.common.serialization.StringSerializer
retries: 1 # 消息发送重试次数
#acks = 0:设置成 表示 producer 完全不理睬 leader broker 端的处理结果。此时producer 发送消息后立即开启下 条消息的发送,根本不等待 leader broker 端返回结果
#acks= all 或者-1 :表示当发送消息时, leader broker 不仅会将消息写入本地日志,同时还会等待所有其他副本都成功写入它们各自的本地日志后,才发送响应结果给,消息安全但是吞吐量会比较低。
#acks = 1:默认的参数值。 producer 发送消息后 leader broker 仅将该消息写入本地日志,然后便发送响应结果给producer ,而无须等待其他副本写入该消息。折中方案,只要leader一直活着消息就不会丢失,同时也保证了吞吐量
acks: 1 #应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
batch-size: 16384 #批量大小
properties:
linger:
ms: 0 #提交延迟
buffer-memory: 33554432 # 生产端缓冲区大小
# 消费者配置
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
# 默认分组id
group-id: DEMO_${random.uuid}
enable-auto-commit: true
#提交offset延时(接收到消息后多久提交offset)
auto-commit-interval: 100ms
#当kafka中没有初始offset或offset超出范围时将自动重置offset
# earliest:重置为分区中最小的offset;
# latest:重置为分区中最新的offset(消费分区中新产生的数据);
# none:只要有一个分区不存在已提交的offset,就抛出异常;
auto-offset-reset: latest
#批量消费每次最多消费多少条消息
#每次拉取一条,一条条消费,当然是具体业务状况设置
max-poll-records: 1
FirstKafkaConfig.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
/**
* 第一个kafka配置
*
*/
@Configuration
public class FirstKafkaConfig {
/**
* 读取第一个kafka配置
* Primary注解表示默认以这个为准
*
* @return 第一个kafka配置
*/
@Primary
@ConfigurationProperties(prefix = "spring.kafka.first")
@Bean
public KafkaProperties firstKafkaProperties() {
return new KafkaProperties();
}
/**
* 构建第一个kafka的生产者发送template
*
* @param firstKafkaProperties 第一个kafka配置
* @return 第一个kafka的生产者发送template
*/
@Primary
@Bean
public KafkaTemplate<String, String> firstKafkaTemplate(
@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
return new KafkaTemplate<>(firstProducerFactory(firstKafkaProperties));
}
/**
* 构建第一个kafka的消费者监听容器工厂
*
* @param firstKafkaProperties 第一个kafka配置
* @return 第一个kafka的消费者监听容器工厂
*/
@Bean("firstKafkaListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
firstKafkaListenerContainerFactory(@Autowired @Qualifier("firstKafkaProperties") KafkaProperties firstKafkaProperties) {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(firstConsumerFactory(firstKafkaProperties));
return factory;
}
/**
* 新建第一个kafka的消费者工厂
*
* @param firstKafkaProperties 第一个kafka配置
* @return 第一个kafka的消费者工厂
*/
private ConsumerFactory<? super Integer, ? super String> firstConsumerFactory(KafkaProperties firstKafkaProperties) {
return new DefaultKafkaConsumerFactory<>(firstKafkaProperties.buildConsumerProperties());
}
/**
* 新建第一个kafka的生产者工厂
*
* @param firstKafkaProperties 第一个kafka配置
* @return 第一个kafka的生产者工厂
*/
private DefaultKafkaProducerFactory<String, String> firstProducerFactory(KafkaProperties firstKafkaProperties) {
return new DefaultKafkaProducerFactory<>(firstKafkaProperties.buildProducerProperties());
}
}
SecondKafkaConfig.java
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
/**
* 第二个kafka配置
*
*/
@Configuration
public class SecondKafkaConfig {
/**
* 读取第二个kafka配置
*
* @return 第二个kafka配置
*/
@ConfigurationProperties(prefix = "spring.kafka.second")
@Bean
public KafkaProperties secondKafkaProperties() {
return new KafkaProperties();
}
/**
* 构建第二个kafka的生产者发送template
*
* @param secondKafkaProperties 第二个kafka配置
* @return 第二个kafka的生产者发送template
*/
@Bean
public KafkaTemplate<String, String> secondKafkaTemplate(
@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
return new KafkaTemplate<>(secondProducerFactory(secondKafkaProperties));
}
/**
* 构建第二个kafka的消费者监听容器工厂
*
* @param secondKafkaProperties 第二个kafka配置
* @return 第二个kafka的消费者监听容器工厂
*/
@Bean("secondKafkaListenerContainerFactory")
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
secondKafkaListenerContainerFactory(@Autowired @Qualifier("secondKafkaProperties") KafkaProperties secondKafkaProperties) {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(secondConsumerFactory(secondKafkaProperties));
return factory;
}
/**
* 新建第二个kafka的消费者工厂
*
* @param secondKafkaProperties 第二个kafka配置
* @return 第二个kafka的消费者工厂
*/
private ConsumerFactory<? super Integer, ? super String> secondConsumerFactory(KafkaProperties secondKafkaProperties) {
return new DefaultKafkaConsumerFactory<>(secondKafkaProperties.buildConsumerProperties());
}
/**
* 新建第二个kafka的生产者工厂
*
* @param secondKafkaProperties 第二个kafka配置
* @return 第二个kafka的生产者工厂
*/
private DefaultKafkaProducerFactory<String, String> secondProducerFactory(KafkaProperties secondKafkaProperties) {
return new DefaultKafkaProducerFactory<>(secondKafkaProperties.buildProducerProperties());
}
}
/**
* 第二个kafka配置需要指定下名字
*/
@Resource(name = "secondKafkaTemplate")
private KafkaTemplate<String, String> secondKafkaTemplate;
/**
* 测试消费者
* 这里要消费哪一个kafka消息,containerFactory就需要配成上面相对应的消费者监听容器工厂
*
* @param record 消息
*/
@KafkaListener(
containerFactory = "firstKafkaListenerContainerFactory",
topics = {"orip"},
groupId = "first-consumer")
public void testConsumer(ConsumerRecord<?, ?> record) {
String value = record.value() + "";
System.out.println();
}