1. 在redis项目中,增加kafka依赖,刷新maven

<!-- Kafka For Spring -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.8.1</version>
</dependency>

2. 在application.yml文件中添加kafka配置

放在spring下

kafka:bootstrap-servers: 192.168.109.160:9092,192.168.109.161:9092,192.168.109.162:9092producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializeracks: allconsumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3. 创建两个主题,两组消费者(组1有2个成员)

package com.hahashou.test.redis.service;import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @description: kafka消费者* @author: 哼唧兽* @date: 9999/9/21**/
@Component
@Slf4j
public class KafkaConsumer {public static final String FIRST_TOPIC = "first-topic";public static final String SECOND_TOPIC = "second-topic";@KafkaListener(topics = FIRST_TOPIC, groupId = "first")public void firstTopicFirst1 (ConsumerRecord<?, ?> record) {log.info("First主题第1个消费者 : "+record.topic()+" : "+record.partition()+" : "+record.value());}@KafkaListener(topics = FIRST_TOPIC, groupId = "first")public void firstTopicFirst2 (ConsumerRecord<?, ?> record) {log.info("First主题第2个消费者 : "+record.topic()+" : "+record.partition()+" : "+record.value());}@KafkaListener(topics = SECOND_TOPIC, groupId = "first")public void secondTopicFirst1 (ConsumerRecord<?, ?> record) {log.info("Second主题第1个消费者 : "+record.topic()+" : "+record.partition()+" : "+record.value());}
}

4. TestKafkaController,并启动项目

package com.hahashou.test.redis.controller;import com.hahashou.test.redis.service.KafkaConsumer;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;/*** @description: 测试Kafka* @author: 哼唧兽* @date: 9999/9/21**/
@RestController
@RequestMapping("/kafka")
@Api(tags = "测试Kafka")
@Slf4j
public class TestKafkaController {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;@GetMapping("/send")@ApiOperation(value = "发送消息")public String send(@RequestParam Integer count) throws InterruptedException {for (int i = 0; i < count; i++) {kafkaTemplate.send(KafkaConsumer.FIRST_TOPIC, i+" for first-topic");Thread.sleep(500);kafkaTemplate.send(KafkaConsumer.SECOND_TOPIC, i+"for second-topic");}return "成功";}
}

5. 测试

5.1 先执行5次

消费日志

根据我看到的文档,以及对kafka的了解,推断一下 :

1. 消费记录按时间有序
2. first-topic的第一个消费者监听的是分区 0 和 1, 第二个消费者监听的分区是 2 和 3
3. 多个分区数据由一组消费者消费,只有一个成员肯定自己全消费咯
4. 确实是按照配置文件创建了 4 个分区,官方示例是8
5. 官方给出的生产配置示例,不建议自动创建topic,是因为分区数量和消费者数量最好是比例差不多,对等应该是比较好的。手动创建topic,至少有那么一瞬间是知道自己需要创建topic,以及topic的分区数量、备份数量,以便后续开发人员创建合适数量的消费者。消费者数量肯定是不能多余分区数量的(官方文档应该有说吗,我没看到),下面增加3个消费者,测试一下

5.2 删除掉second-topic,对first-topic再增加3个消费者,重启项目

消费日志

第3个消费者没有抢到任何分区,就是个摆设,项目重启,消费者对应分区会重新分配