SpringBoot+kafka初始配置与使用

/**
 * 初始化默认 kafka 分区
 *
 * @author maxzhao
 * @date 2021-01-26 14:37
 */
@Configuration
public class DefaultTestKafkaInitialConfiguration {
    public static final String TEST_TOPIC = "test-topic";

    /**
     * 创建一个 test-topic 的 Topic，设置分区数为 1,分区副本数为 1
     * <p>修改分区只需要重新配置分区数、分区副本数，然后重启（数量只能增大）</p>
     */
    @Bean
    public NewTopic initialTopic() {
        return new NewTopic(TEST_TOPIC, 1, (short) 1);
    }
}

public interface IKafkaService {
    /**
     * 发送消息
     *
     * @param topic   主题
     * @param message 消息
     */
    void sendMessage(String topic, String message);

}

@Slf4j
@Service
public class KafkaServiceImpl implements IKafkaService {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 消费监听
     */
    @KafkaListener(topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        log.debug("消费：{}-{}-{}", record.topic(), record.partition(), record.value());
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {
    @Autowired
    private IKafkaService kafkaService;

    /**
     * 测试消息发送
     */
    @Test
    public void sendMessage() {
        String message = "这是一个简单的消息！";
        log.debug("发送消息：{}-{}", DefaultTestKafkaInitialConfiguration.TEST_TOPIC, message);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC, message);
        log.debug("发送完成");
    }
}

@Slf4j
@Service("kafkaService")
public class KafkaServiceImpl implements IKafkaService {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message).addCallback(success -> {
            if (success == null) {
                log.debug("消息发送成功，响应数据不存在");
            } else {
                RecordMetadata recordMetadata = success.getRecordMetadata();
                /*消息发送到的topic,消息发送到的分区 partition,消息在分区内的offset*/
                log.debug("消息发送成功：topic:{} partition:{} offset:{}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
            }
        }, throwable -> {
            log.warn("消息发送失败：topic:{} message{} error:{}", topic, message, throwable.getMessage());
        });

        kafkaTemplate.send(topic, message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> success) {
                if (success == null) {
                    log.debug("消息发送成功，响应数据不存在");
                } else {
                    RecordMetadata recordMetadata = success.getRecordMetadata();
                    /*消息发送到的topic,消息发送到的分区 partition,消息在分区内的offset*/
                    log.debug("消息发送成功：topic:{} partition:{} offset:{}", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
                }
            }

            @Override
            public void onFailure(Throwable throwable) {
                log.warn("消息发送失败：topic:{} message{} error:{}", topic, message, throwable.getMessage());
            }
        });

    }
}

public class CustomPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 自定义分区规则
        // ......
        // 0 则是默认发到 0号
        return 0;
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> configs) {
    }
}

@Slf4j
@Service("kafkaService")
public class KafkaServiceImpl implements IKafkaService {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Override
    public void sendMessageWithTransaction(String topic, String message) {
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(topic, message).addCallback(success -> {
                /*没有发出去，就没有响应*/
            }, throwable -> {
                /*没有发出去，就没有响应*/
            });
            /*这里抛出错误，则不会发出消息*/
            throw new RuntimeException();
        });
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 消费监听
     * <p>可以监听多个</p>
     */
    @KafkaListener(topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC})
    public void onMessage1(ConsumerRecord<?, ?> record) {
        log.debug("消费：{}-{}-{}", record.topic(), record.partition(), record.value());
    }

    /**
     *  指定topic、partition、offset消费
     * <p>同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8</p>
     **/
    @KafkaListener(id = "consumer1", groupId = "consumer-group",
            topicPartitions = {
                    @TopicPartition(topic = DefaultTestKafkaInitialConfiguration.TEST_TOPIC, partitions = {"0"}),
                    @TopicPartition(topic = DefaultTestKafkaInitialConfiguration.TEST_TOPIC_2, partitions = "0",
                            partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
            })
    public void onMessage2(ConsumerRecord<?, ?> record) {
        log.debug("消费2：topic:{} partition:{} value:{} offset:{}", record.topic(), record.partition(), record.value(), record.offset());
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /******************************************************************/
    /**
     * 批量消费
     **/
    @KafkaListener(id = "consumer2", groupId = "consumer-group",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3})
    public void onMessage3(List<ConsumerRecord<?, ?>> records) {
        for (ConsumerRecord<?, ?> record : records) {
            log.debug("批量消费：topic:{} partition:{} value:{} offset:{}", record.topic(), record.partition(), record.value(), record.offset());
        }
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {
    @Autowired
    private IKafkaService kafkaService;

    /**
     * 测试发送多个消息发送
     */
    @Test
    public void sendMessage3() {
        String message = "这是一个简单的消息！";
        String message2 = "这是一个简单的消息！_2";
        String message3 = "这是一个简单的消息！_3";
        String message4 = "这是一个简单的消息！_4";
        log.debug("发送消息：{}-{}", DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3, message);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3, message);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3, message2);
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
        }
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3, message3);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_3, message4);
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
        }
        log.debug("发送完成");
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /********************************************************/
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
        return (message, exception, consumer) -> {
            log.debug("批量消费异常：topic:{} message:{}",
                    consumer.listTopics(),
                    message.getPayload());
            return null;
        };
    }

    /**
     * 批量消费
     **/
    @KafkaListener(id = "consumer4", groupId = "consumer-group",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_4},
            errorHandler = "consumerAwareErrorHandler")
    public void onMessage4(List<ConsumerRecord<?, ?>> records) throws Exception {
        throw new Exception("批量消费-模拟异常");
    }

}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {
    /**
     * 测试异常处理
     */
    @Test
    public void sendMessage4() {
        String message = "这是一个简单的消息_4！";
        log.debug("发送消息：{}-{}", DefaultTestKafkaInitialConfiguration.TEST_TOPIC_4, message);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_4, message);
        log.debug("发送完成");
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /********************************************************/

    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * 消息过滤器
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory filterContainerFactory() {
        ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        // 消息过滤策略
        factory.setRecordFilterStrategy(consumerRecord -> {
            if (consumerRecord.value().toString().length() > 20) {
                return false;
            }
            //返回true消息则被过滤
            return true;
        });
        return factory;
    }

    /**
     * 消息过滤
     */
    @KafkaListener(id = "consumer5", groupId = "consumer-group",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_5},
            containerFactory = "filterContainerFactory")
    public void onMessage5(String record) {
        log.debug("批量消费-拦截长度小于20的字符串： value:{} ", record);
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {
    /**
     * 消息过滤
     * <p>需要吧提交延时改为0</p>
     */
    @Test
    public void sendMessage5() {
        String message = "这是一个简单的消息_5！";
        String message2 = "这是一个简单的消息_5！__________________________________________________";
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_5, message);
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_5, message2);
        log.debug("发送完成");
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 消息转发给 5
     */
    @KafkaListener(id = "consumer6", groupId = "consumer-group",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_6})
    @SendTo(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_5)
    public String onMessage6(List<ConsumerRecord<?, ?>> records) {
        log.debug("消息转发 6：topic:{} partition:{} value:{} offset:{}",
                records.get(0).topic(),
                records.get(0).partition(),
                records.get(0).value(),
                records.get(0).offset());
        return "消息转发：" + records.get(0).value().toString();
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    /**
     * 监听器容器工厂(设置禁止KafkaListener自启动)
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory container = new ConcurrentKafkaListenerContainerFactory();
        container.setConsumerFactory(consumerFactory);
        /*禁止 KafkaListener 自启动*/
        container.setAutoStartup(false);
        return container;
    }

    /**
     * 消息监听的起停控制
     */
    @KafkaListener(id = "consumer7", groupId = "consumer-group",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_7},
            containerFactory = "delayContainerFactory")
    public void onMessage7(String record) {
        log.debug("消息监听的起停控制-启动后可以接收到的消息： value:{} ", record);
    }
}

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {
    /**
     * 消息监听的起停控制 7
     */
    @Test
    public void sendMessage7() {
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_7, "监听 7 默认未未启动，这是一个简单的消息_7！");
        /*让启动后收到消息更明显*/
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        if (!registry.getListenerContainer("consumer7").isRunning()) {
            log.debug("监听 7 还未启动，准备启动监听7");
            registry.getListenerContainer("consumer7").start();
        }
        log.debug("监听 7 启动");
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_7, "监听 7 已启动——————————！");
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        registry.getListenerContainer("consumer7").pause();
        log.debug("监听 7 暂停");
        try { Thread.sleep(5000); } catch (InterruptedException e) { }
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_7, "监听 7 已暂停——————————————！");
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        registry.getListenerContainer("consumer7").stop();
        log.debug("监听 7 停止");
        try { Thread.sleep(1000); } catch (InterruptedException e) { }
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_7, "监听 7 已停止——————————————！");
        log.debug("发送完成");
        try { Thread.sleep(5000); } catch (InterruptedException e) { }
    }
}

@Slf4j
@Component
public class KafkaConsumer {
    @KafkaListener(id = "consumer8", groupId = "consumer-group8",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_8})
    public void onMessage8(String record) {
        log.debug("分组测试消费：group8-consumer8 收到消息： value:{} ", record);
    }
    @KafkaListener(id = "consumer8_1", groupId = "consumer-group8",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_8})
    public void onMessage8_1(String record) {
        log.debug("分组测试消费：group8-consumer8_1 收到消息： value:{} ", record);
    }
    @KafkaListener(id = "consumer8_2", groupId = "consumer-group8_2",
            topics = {DefaultTestKafkaInitialConfiguration.TEST_TOPIC_8})
    public void onMessage8_2(String record) {
        log.debug("分组测试消费：group8_2-consumer8_2 收到消息： value:{} ", record);
    }
}
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = start.DemoApplication.class)
public class KafkaTest {   
    /**
     * 分组测试消费 8
     */
    @Test
    public void sendMessage8() {
        kafkaService.sendMessage(DefaultTestKafkaInitialConfiguration.TEST_TOPIC_8, "这是一个简单的消息_8！");
        log.debug("发送完成");
    }
}