模拟spring-kafka实现kafka的consumer监听

背景：因为某些原因，无法直接使用springboot提供的@KafkaListener，改为模拟springboot注解的方式搬过来实现

首先创建一个业务处理的service，这个service主要用于消费下来的消息的处理

@Service
public class MsgProcessService {
    public void consumeMsg(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
            System.out.println("receive msg:{}", records);         
            ack.acknowledge();        
    }
}

然后创建消费者的监听配置类，从afterPropertiesSet()开始执行，注解和常用的套路一样

@Configuration
@EnableKafka
public class KafkaClient implements InitializingBean, Closeable {

    @Resource
    private MsgProcessService msgProcessService;
    /**
     * 消费者监听者容器
     */
    private KafkaMessageListenerContainer<String, String> listenerContainer;

    @Override
    public void afterPropertiesSet() {
        new Thread(this::initListeners, "kafkaInitThread").start();
    }

    @Override
    public void close() {
        if (listenerContainer != null) {
            listenerContainer.stop();
        }
    }

    private void initListeners() throws NoSuchMethodException {
        //第一步：构造BatchMessagingMessageListenerAdaper
        Method method = MsgProcessService.class.getMethod("consumeMsg", List.class, Acknowledgment.class);
        InvocableHandlerMethod handlerMethod = new InvocableHandlerMethod(msgProcessService, method);
        handlerMethod.setMessageMethodArgumentResolvers(new HandlerMethodArgumentResolverComposite());
        BatchMessagingMessageListenerAdapter<String, String> messageListener = new BatchMessagingMessageListenerAdapter<>(msgProcessService, method);
        messageListener.setHandlerMethod(new HandlerAdapter(handlerMethod));          
        //第二步：构造ConsumerFactory
        DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(buildConsumerConfigs());
        //第三步：构造ContainerProperties
        ContainerProperties containerProperties = new ContainerProperties("myTopic");
        containerProperties.setMessageListener(messageListener);
        containerProperties.setGroupId("group007");
        containerProperties.setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL); 
        //第四步：构造KafkaMessageListenerContainer
        KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
        this.listenerContainer = container;
        //第五步：启动container
        container.start();
    }

    public Map<String, Object> buildConsumerConfigs() {
        Map<String, Object> props = new HashMap<>(16);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        //从哪儿开始消费，默认为latest
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");       
        //每次只处理一条消息
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        //两次poll的最大时间间隔设置为5s
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,5000);
        //取消自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //设置数据key和value的序列化处理类
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return props;
    }

}