1. pom.xml
Kafka并没有提供对应的Starter,而是使用的spring-kafka
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.2</version>
</dependency>
2. application.properties
#============== kafka ===================
# 指定kafka 代理地址,可以多个
spring.kafka.bootstrap-servers=localhost:9092
#=============== provider =======================
spring.kafka.producer.retries=0
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
# 指定消息key和消息体的编解码方式
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
#=============== consumer =======================
# 指定默认消费者group id
spring.kafka.consumer.group-id=test-consumer-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100
# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
3. KafkaTemplate
Kafka使用KafkaTemplate发送消息,使用@KafkaListener消费指定主题的消息。
@Data
@AllArgsConstructor
public class Message {
private String id; // id
private String msg; // 消息
private Long timestamp; // 时间戳
}
生产消息(发送消息)
@RunWith(SpringRunner.class)
@SpringBootTest
public class KafkaApplicationTests {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Test
public void testKafka() throws Exception{
Gson gson = new GsonBuilder().create();
String topic = "topic1";
for (int i = 0; i < 3; i ++) {
String key = "key-" + i;
Message data = new Message(UUID.randomUUID().toString(), "hello kafka: " + i, new Date().getTime());
// 默认情况下,如果主题不存在会自动帮你创建注意,是否允许创建不存在的主题可以在kafka中配置
kafkaTemplate.send(topic, key, gson.toJson(data));
Thread.sleep(1000);
}
}
}
消费消息(接收消息)
@Component
public class KafkaListener {
public static Logger logger = LoggerFactory.getLogger(KafkaListener.class);
@org.springframework.kafka.annotation.KafkaListener(topics = {"topic1", "topic2"})
public void listenerForTopic1(ConsumerRecord<?, ?> record) {
logger.info("topic={} key={} value={}", record.topic(), record.key(), record.value());
}
}
4. 启动kafka
# 首先启动zookeeper
cd /usr/local/Cellar/kafka/1.0.0
./bin/zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties
# 再启动kafka
cd /usr/local/Cellar/kafka/1.0.0
./bin/kafka-server-start /usr/local/etc/kafka/server.properties