record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic, Consumer consumer) {
System.out.println("消费者收到消息:" + record.value() + "; topic:" + topic);
/*
* 如果需要手工提交异步 consumer.commitSync();
* 手工同步提交 consumer.commitAsync()
*/
}
}
```
#### 2.4 测试整合结果
```java
@Slf4j
@RestController
public class SendMsgController {
@Autowired
private KafKaCustomrProducer producer;
@Autowired
private KafkaTemplate kafkaTemplate;
/***
* 发送消息体为基本类型的消息
*/
@GetMapping("sendSimple")
public void sendSimple() {
producer.sendMessage(Topic.SIMPLE, "hello spring boot kafka");
}
}
```
## 三、关于多消费者组的测试
#### 3.1 创建多分区主题
```java
/**
* @author : heibaiying
* @description : kafka 配置类
*/
@Configuration
public class KafkaConfig {
@Bean
public NewTopic groupTopic() {
// 指定主题名称，分区数量，和复制因子
return new NewTopic(Topic.GROUP, 10, (short) 2);
}
}
```
#### 3.2 多消费者组对同一主题的监听
1. 消费者 1-1 监听主题的 0、1 分区
2. 消费者 1-2 监听主题的 2、3 分区
3. 消费者 1-3 监听主题的 0、1 分区
4. 消费者 2-1 监听主题的所有分区
```java
/**
* @author : heibaiying
* @description : kafka 消费者组
*
* 多个消费者群组可以共同读取同一个主题，彼此之间互不影响。
*/
@Component
@Slf4j
public class KafkaGroupConsumer {
// 分组 1 中的消费者 1
@KafkaListener(id = "consumer1-1", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"0", "1"})
})
public void consumer1_1(ConsumerRecord record) {
System.out.println("consumer1-1 收到消息:" + record.value());
}
// 分组 1 中的消费者 2
@KafkaListener(id = "consumer1-2", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"2", "3"})
})
public void consumer1_2(ConsumerRecord record) {
System.out.println("consumer1-2 收到消息:" + record.value());
}
// 分组 1 中的消费者 3
@KafkaListener(id = "consumer1-3", groupId = "group1", topicPartitions =
{@TopicPartition(topic = Topic.GROUP, partitions = {"0", "1"})
})
public void consumer1_3(ConsumerRecord record) {
System.out.println("consumer1-3 收到消息:" + record.value());
}
// 分组 2 中的消费者
@KafkaListener(id = "consumer2-1", groupId = "group2", topics = Topic.GROUP)
public void consumer2_1(ConsumerRecord record) {
System.err.println("consumer2-1 收到消息:" + record.value());
}
}
```
#### 3.2 发送消息时候指定主题的具体分区
```java
/***
* 多消费者组、组中多消费者对同一主题的消费情况
*/
@GetMapping("sendGroup")
public void sendGroup() {
for (int i = 0; i < 4; i++) {
// 第二个参数指定分区，第三个参数指定消息键 分区优先
ListenableFuture> future = kafkaTemplate.send(Topic.GROUP, i % 4, "key", "hello group " + i);
future.addCallback(new ListenableFutureCallback>() {
@Override
public void onFailure(Throwable throwable) {
log.info("发送消息失败:" + throwable.getMessage());
}
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("发送结果:" + sendResult.toString());
}
});
}
}
```
测试结果：
```yaml
# 主要看每次发送结果中的 partition 属性，代表四次消息分别发送到了主题的0,1,2,3分区
发送结果:SendResult [producerRecord=ProducerRecord(topic=spring.boot.kafka.newGroup, partition=1, headers=RecordHeaders(headers = [], isReadOnly = true), key=key, value=hello group 1, timestamp=null), recordMetadata=spring.boot.kafka.newGroup-1@13]
发送结果:SendResult [producerRecord=ProducerRecord(topic=spring.boot.kafka.newGroup, partition=0, headers=RecordHeaders(headers = [], isReadOnly = true), key=key, value=hello group 0, timestamp=null), recordMetadata=spring.boot.kafka.newGroup-0@19]
发送结果:SendResult [producerRecord=ProducerRecord(topic=spring.boot.kafka.newGroup, partition=3, headers=RecordHeaders(headers = [], isReadOnly = true), key=key, value=hello group 3, timestamp=null), recordMetadata=spring.boot.kafka.newGroup-3@13]
发送结果:SendResult [producerRecord=ProducerRecord(topic=spring.boot.kafka.newGroup, partition=2, headers=RecordHeaders(headers = [], isReadOnly = true), key=key, value=hello group 2, timestamp=null), recordMetadata=spring.boot.kafka.newGroup-2@13]
# 消费者组2 接收情况
consumer2-1 收到消息:hello group 1
consumer2-1 收到消息:hello group 0
consumer2-1 收到消息:hello group 2
consumer2-1 收到消息:hello group 3
# 消费者1-1接收情况
consumer1-1 收到消息:hello group 1
consumer1-1 收到消息:hello group 0
# 消费者1-3接收情况
consumer1-3 收到消息:hello group 1
consumer1-3 收到消息:hello group 0
# 消费者1-2接收情况
consumer1-2 收到消息:hello group 3
consumer1-2 收到消息:hello group 2
```
#### 3.4 测试结果
1. 和 kafka 原本的机制一样，多消费者组之间对于同一个主题的消费彼此之间互不影响；
2. 和 kafka 原本机制不一样的是，这里我们消费者 1-1 和消费 1-3 共同属于同一个消费者组，并且监听同样的分区，按照原本 kafka 的机制，群组保证每个分区只能被同一个消费者组的一个消费者使用，但是按照 spring 的声明方式实现的消息监听，这里被两个消费者都监听到了。
## 四、序列化与反序列化
用例采用的是第三方 fastjson 将实体类序列化为 json 后发送。实现如下：
```java
/***
* 发送消息体为 bean 的消息
*/
@GetMapping("sendBean")
public void sendBean() {
Programmer programmer = new Programmer("xiaoming", 12, 21212.33f, new Date());
producer.sendMessage(Topic.BEAN, JSON.toJSON(programmer).toString());
}
```
```java
@Component
@Slf4j
public class KafkaBeanConsumer {
@KafkaListener(groupId = "beanGroup",topics = Topic.BEAN)
public void consumer(ConsumerRecord record) {
System.out.println("消费者收到消息:" + JSON.parseObject(record.value().toString(), Programmer.class));
}
}
```
