聊聊如何实现一个带幂等模板的Kafka消费者

前言

不知道大家有没有这样的体验，你跟你团队的成员，宣导一些开发时注意事项，比如在使用消息队列时，在消费端处理消息时，需根据业务场景，考虑一下幂等。后面走查代码的时，会发现一些资浅的开发，在需要幂等判断的场景的情况下，仍然没做幂等判断。既然宣导无效，就干脆实现一个带幂等模板的消费者，然后开发基于这个模板进行消费端业务处理。本文就以spring-Kafka举例，聊聊如何实现一个带幂等模板的kafka消费者

实现步骤

1、kafka自动提交改为手动提交

spring:    kafka:        consumer:            #  是否自动提交偏移量，默认值是true,为了避免出现重复数据和数据丢失，可以把它设置为false,然后手动提交偏移量            enable-auto-commit: ${KAFKA_CONSUMER_ENABLE_AUTO_COMMIT:false}

2、定义消费端模板抽象基类

@Slf4jpublic abstract class BaseComusmeListener {    @KafkaHandler    public final void receive(@Payload String data, @Header(value = KafkaHeaders.RECEIVED_TOPIC,required = false) String receivedTopic,                              @Header(value = KafkaHeaders.RECEIVED_MESSAGE_KEY,required = false) String receivedMessageKey, @Header(value = KafkaHeaders.RECEIVED_TIMESTAMP,required = false) long receivedTimestamp, Acknowledgment ack){        KafkaComsumePayLoad kafkaComsumePayLoad = buildKafkaComsumePayLoad(data,receivedTimestamp,receivedTopic,receivedMessageKey);        boolean isRepeateConsume = isRepeateConsume(kafkaComsumePayLoad);        if(isRepeateConsume){            log.warn("messageKey:【{}】，topic：【{}】存在重复消息数据-->【{}】",receivedMessageKey,receivedTopic,data);            //手工确认            ack.acknowledge();            return;        }        if(doBiz(kafkaComsumePayLoad)){            //手工确认            ack.acknowledge();        }    }    /**     * 是否重复消费     * @param kafkaComsumePayLoad     * @return     */    public abstract boolean isRepeateConsume(KafkaComsumePayLoad kafkaComsumePayLoad);    /**     * 业务处理     * @param kafkaComsumerPayLoad     */    public abstract boolean doBiz(KafkaComsumePayLoad kafkaComsumerPayLoad);    private KafkaComsumePayLoad buildKafkaComsumePayLoad(String data, long receivedTimestamp, String receivedTopic, String receivedMessageKey){        return KafkaComsumePayLoad.builder()                .data(data)                .receivedTimestamp(receivedTimestamp)                .receivedTopic(receivedTopic)                .receivedMessageKey(receivedMessageKey)                .build();    }}

3、自定义监听注解【可选】

@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })@Retention(RetentionPolicy.RUNTIME)@KafkaListener@Documented@Componentpublic @interface LybGeekKafkaListener {    @AliasFor(annotation = KafkaListener.class, attribute = "id")    String id() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "containerFactory")    String containerFactory() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "topics")    String[] topics() default {};    @AliasFor(annotation = KafkaListener.class, attribute = "topicPattern")    String topicPattern() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "topicPartitions")    TopicPartition[] topicPartitions() default {};    @AliasFor(annotation = KafkaListener.class, attribute = "containerGroup")    String containerGroup() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "errorHandler")    String errorHandler() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "groupId")    String groupId() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "idIsGroup")    boolean idIsGroup() default true;    @AliasFor(annotation = KafkaListener.class, attribute = "clientIdPrefix")    String clientIdPrefix() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "beanRef")    String beanRef() default "__listener";    @AliasFor(annotation = KafkaListener.class, attribute = "concurrency")    String concurrency() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "autoStartup")    String autoStartup() default "";    @AliasFor(annotation = KafkaListener.class, attribute = "properties")    String[] properties() default {};    @AliasFor(annotation = Component.class, attribute = "value")    String value() default "";}

3、重写KafkaListener注解后置处理器【可选】

注： 因示例项目的springboot版本比较低，直接使用@LybGeekKafkaListener不起作用

public class LybGeekKafkaListenerAnnotationBeanPostProcessor<K, V>        implements BeanPostProcessor, Ordered, BeanFactoryAware, SmartInitializingSingleton {    private static final String GENERATED_ID_PREFIX = "org.springframework.kafka.KafkaListenerEndpointContainer#";    /**     * The bean name of the default {@link KafkaListenerContainerFactory}.     */    public static final String DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME = "kafkaListenerContainerFactory";    private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap<>(64));    private final Log logger = LogFactory.getLog(getClass());    private final ListenerScope listenerScope = new ListenerScope();    private KafkaListenerEndpointRegistry endpointRegistry;    private String defaultContainerFactoryBeanName = DEFAULT_KAFKA_LISTENER_CONTAINER_FACTORY_BEAN_NAME;    private DefaultListableBeanFactory beanFactory;    private final KafkaHandlerMethodFactoryAdapter messageHandlerMethodFactory =            new KafkaHandlerMethodFactoryAdapter();    private final KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar();    private final AtomicInteger counter = new AtomicInteger();    private BeanExpressionResolver resolver = new StandardBeanExpressionResolver();    private BeanExpressionContext expressionContext;    private Charset charset = StandardCharsets.UTF_8;    @Override    public int getOrder() {        return LOWEST_PRECEDENCE;    }    /**     * Set the {@link KafkaListenerEndpointRegistry} that will hold the created     * endpoint and manage the lifecycle of the related listener container.     * @param endpointRegistry the {@link KafkaListenerEndpointRegistry} to set.     */    public void setEndpointRegistry(KafkaListenerEndpointRegistry endpointRegistry) {        this.endpointRegistry = endpointRegistry;    }    /**     * Set the name of the {@link KafkaListenerContainerFactory} to use by default.     * <p>If none is specified, "kafkaListenerContainerFactory" is assumed to be defined.     * @param containerFactoryBeanName the {@link KafkaListenerContainerFactory} bean name.     */    public void setDefaultContainerFactoryBeanName(String containerFactoryBeanName) {        this.defaultContainerFactoryBeanName = containerFactoryBeanName;    }    /**     * Set the {@link MessageHandlerMethodFactory} to use to configure the message     * listener responsible to serve an endpoint detected by this processor.     * <p>By default, {@link DefaultMessageHandlerMethodFactory} is used and it     * can be configured further to support additional method arguments     * or to customize conversion and validation support. See     * {@link DefaultMessageHandlerMethodFactory} Javadoc for more details.     * @param messageHandlerMethodFactory the {@link MessageHandlerMethodFactory} instance.     */    public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory messageHandlerMethodFactory) {        this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(messageHandlerMethodFactory);    }    /**     * Making a {@link BeanFactory} available is optional; if not set,     * {@link KafkaListenerConfigurer} beans won't get autodetected and an     * {@link #setEndpointRegistry endpoint registry} has to be explicitly configured.     * @param beanFactory the {@link BeanFactory} to be used.     */    @Override    public void setBeanFactory(BeanFactory beanFactory) {        this.beanFactory = (DefaultListableBeanFactory) beanFactory;        if (beanFactory instanceof ConfigurableListableBeanFactory) {            this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver();            this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory,                    this.listenerScope);        }    }    /**     * Set a charset to use when converting byte[] to String in method arguments.     * Default UTF-8.     * @param charset the charset.     * @since 2.2     */    public void setCharset(Charset charset) {        Assert.notNull(charset, "'charset' cannot be null");        this.charset = charset;    }    @Override    public void afterSingletonsInstantiated() {        this.registrar.setBeanFactory(this.beanFactory);        if (this.beanFactory instanceof ListableBeanFactory) {            Map<String, KafkaListenerConfigurer> instances =                    ((ListableBeanFactory) this.beanFactory).getBeansOfType(KafkaListenerConfigurer.class);            for (KafkaListenerConfigurer configurer : instances.values()) {                configurer.configureKafkaListeners(this.registrar);            }        }        if (this.registrar.getEndpointRegistry() == null) {            if (this.endpointRegistry == null) {                Assert.state(this.beanFactory != null,                        "BeanFactory must be set to find endpoint registry by bean name");                this.endpointRegistry = this.beanFactory.getBean(                        KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,                        KafkaListenerEndpointRegistry.class);            }            this.registrar.setEndpointRegistry(this.endpointRegistry);        }        if (this.defaultContainerFactoryBeanName != null) {            this.registrar.setContainerFactoryBeanName(this.defaultContainerFactoryBeanName);        }        // Set the custom handler method factory once resolved by the configurer        MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();        if (handlerMethodFactory != null) {            this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(handlerMethodFactory);        }        else {            addFormatters(this.messageHandlerMethodFactory.defaultFormattingConversionService);        }        // Actually register all listeners        this.registrar.afterPropertiesSet();        beanFactory.removeBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME);    }    @Override    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {        return bean;    }    @Override    public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {        if (!this.nonAnnotatedClasses.contains(bean.getClass())) {            Class<?> targetClass = AopUtils.getTargetClass(bean);            Collection<LybGeekKafkaListener> classLevelListeners = findListenerAnnotations(targetClass);            final boolean hasClassLevelListeners = classLevelListeners.size() > 0;            final List<Method> multiMethods = new ArrayList<>();            Map<Method, Set<LybGeekKafkaListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,                    (MethodIntrospector.MetadataLookup<Set<LybGeekKafkaListener>>) method -> {                        Set<LybGeekKafkaListener> listenerMethods = findListenerAnnotations(method);                        return (!listenerMethods.isEmpty() ? listenerMethods : null);                    });            if (hasClassLevelListeners) {                Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass,                        (ReflectionUtils.MethodFilter) method ->                                AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);                multiMethods.addAll(methodsWithHandler);            }            if (annotatedMethods.isEmpty()) {                this.nonAnnotatedClasses.add(bean.getClass());                if (this.logger.isTraceEnabled()) {                    this.logger.trace("No @LybGeekKafkaListener annotations found on bean type: " + bean.getClass());                }            }            else {                // Non-empty set of methods                for (Map.Entry<Method, Set<LybGeekKafkaListener>> entry : annotatedMethods.entrySet()) {                    Method method = entry.getKey();                    for (LybGeekKafkaListener listener : entry.getValue()) {                        processKafkaListener(listener, method, bean, beanName);                    }                }                if (this.logger.isDebugEnabled()) {                    this.logger.debug(annotatedMethods.size() + " @LybGeekKafkaListener methods processed on bean '"                            + beanName + "': " + annotatedMethods);                }            }            if (hasClassLevelListeners) {                processMultiMethodListeners(classLevelListeners, multiMethods, bean, beanName);            }        }        return bean;    }    /*     * AnnotationUtils.getRepeatableAnnotations does not look at interfaces     */    private Collection<LybGeekKafkaListener> findListenerAnnotations(Class<?> clazz) {        Set<LybGeekKafkaListener> listeners = new HashSet<>();        LybGeekKafkaListener ann = AnnotationUtils.findAnnotation(clazz, LybGeekKafkaListener.class);        if (ann != null) {            listeners.add(ann);        }        return listeners;    }    /*     * AnnotationUtils.getRepeatableAnnotations does not look at interfaces     */    private Set<LybGeekKafkaListener> findListenerAnnotations(Method method) {        Set<LybGeekKafkaListener> listeners = new HashSet<>();        LybGeekKafkaListener ann = AnnotatedElementUtils.findMergedAnnotation(method, LybGeekKafkaListener.class);        if (ann != null) {            listeners.add(ann);        }        return listeners;    }    private void processMultiMethodListeners(Collection<LybGeekKafkaListener> classLevelListeners, List<Method> multiMethods,            Object bean, String beanName) {        List<Method> checkedMethods = new ArrayList<>();        Method defaultMethod = null;        for (Method method : multiMethods) {            Method checked = checkProxy(method, bean);            KafkaHandler annotation = AnnotationUtils.findAnnotation(method, KafkaHandler.class);            if (annotation != null && annotation.isDefault()) {                final Method toAssert = defaultMethod;                Assert.state(toAssert == null, () -> "Only one @KafkaHandler can be marked 'isDefault', found: "                        + toAssert.toString() + " and " + method.toString());                defaultMethod = checked;            }            checkedMethods.add(checked);        }        for (LybGeekKafkaListener classLevelListener : classLevelListeners) {            MultiMethodKafkaListenerEndpoint<K, V> endpoint =                    new MultiMethodKafkaListenerEndpoint<>(checkedMethods, defaultMethod, bean);            processListener(endpoint, classLevelListener, bean, bean.getClass(), beanName);        }    }    protected void processKafkaListener(LybGeekKafkaListener kafkaListener, Method method, Object bean, String beanName) {        Method methodToUse = checkProxy(method, bean);        MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();        endpoint.setMethod(methodToUse);        processListener(endpoint, kafkaListener, bean, methodToUse, beanName);    }    private Method checkProxy(Method methodArg, Object bean) {        Method method = methodArg;        if (AopUtils.isJdkDynamicProxy(bean)) {            try {                // Found a @LybGeekKafkaListener method on the target class for this JDK proxy ->                // is it also present on the proxy itself?                method = bean.getClass().getMethod(method.getName(), method.getParameterTypes());                Class<?>[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces();                for (Class<?> iface : proxiedInterfaces) {                    try {                        method = iface.getMethod(method.getName(), method.getParameterTypes());                        break;                    }                    catch (NoSuchMethodException noMethod) {                    }                }            }            catch (SecurityException ex) {                ReflectionUtils.handleReflectionException(ex);            }            catch (NoSuchMethodException ex) {                throw new IllegalStateException(String.format(                        "@LybGeekKafkaListener method '%s' found on bean target class '%s', " +                                "but not found in any interface(s) for bean JDK proxy. Either " +                                "pull the method up to an interface or switch to subclass (CGLIB) " +                                "proxies by setting proxy-target-class/proxyTargetClass " +                                "attribute to 'true'", method.getName(),                        method.getDeclaringClass().getSimpleName()), ex);            }        }        return method;    }    protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, LybGeekKafkaListener kafkaListener,            Object bean, Object adminTarget, String beanName) {        String beanRef = kafkaListener.beanRef();        if (StringUtils.hasText(beanRef)) {            this.listenerScope.addListener(beanRef, bean);        }        endpoint.setBean(bean);        endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);        endpoint.setId(getEndpointId(kafkaListener));        endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));        endpoint.setTopicPartitions(resolveTopicPartitions(kafkaListener));        endpoint.setTopics(resolveTopics(kafkaListener));        endpoint.setTopicPattern(resolvePattern(kafkaListener));        endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));        String group = kafkaListener.containerGroup();        if (StringUtils.hasText(group)) {            Object resolvedGroup = resolveExpression(group);            if (resolvedGroup instanceof String) {                endpoint.setGroup((String) resolvedGroup);            }        }        String concurrency = kafkaListener.concurrency();        if (StringUtils.hasText(concurrency)) {            endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));        }        String autoStartup = kafkaListener.autoStartup();        if (StringUtils.hasText(autoStartup)) {            endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));        }        resolveKafkaProperties(endpoint, kafkaListener.properties());        KafkaListenerContainerFactory<?> factory = null;        String containerFactoryBeanName = resolve(kafkaListener.containerFactory());        if (StringUtils.hasText(containerFactoryBeanName)) {            Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");            try {                factory = this.beanFactory.getBean(containerFactoryBeanName, KafkaListenerContainerFactory.class);            }            catch (NoSuchBeanDefinitionException ex) {                throw new BeanInitializationException("Could not register Kafka listener endpoint on [" + adminTarget                        + "] for bean " + beanName + ", no " + KafkaListenerContainerFactory.class.getSimpleName()                        + " with id '" + containerFactoryBeanName + "' was found in the application context", ex);            }        }        endpoint.setBeanFactory(this.beanFactory);        String errorHandlerBeanName = resolveExpressionAsString(kafkaListener.errorHandler(), "errorHandler");        if (StringUtils.hasText(errorHandlerBeanName)) {            endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, KafkaListenerErrorHandler.class));        }        this.registrar.registerEndpoint(endpoint, factory);        if (StringUtils.hasText(beanRef)) {            this.listenerScope.removeListener(beanRef);        }    }    private void resolveKafkaProperties(MethodKafkaListenerEndpoint<?, ?> endpoint, String[] propertyStrings) {        if (propertyStrings.length > 0) {            Properties properties = new Properties();            for (String property : propertyStrings) {                String value = resolveExpressionAsString(property, "property");                if (value != null) {                    try {                        properties.load(new StringReader(value));                    }                    catch (IOException e) {                        this.logger.error("Failed to load property " + property + ", continuing...", e);                    }                }            }            endpoint.setConsumerProperties(properties);        }    }    private String getEndpointId(LybGeekKafkaListener kafkaListener) {        if (StringUtils.hasText(kafkaListener.id())) {            return resolveExpressionAsString(kafkaListener.id(), "id");        }        else {            return GENERATED_ID_PREFIX + this.counter.getAndIncrement();        }    }    private String getEndpointGroupId(LybGeekKafkaListener kafkaListener, String id) {        String groupId = null;        if (StringUtils.hasText(kafkaListener.groupId())) {            groupId = resolveExpressionAsString(kafkaListener.groupId(), "groupId");        }        if (groupId == null && kafkaListener.idIsGroup() && StringUtils.hasText(kafkaListener.id())) {            groupId = id;        }        return groupId;    }    private TopicPartitionInitialOffset[] resolveTopicPartitions(LybGeekKafkaListener kafkaListener) {        TopicPartition[] topicPartitions = kafkaListener.topicPartitions();        List<TopicPartitionInitialOffset> result = new ArrayList<>();        if (topicPartitions.length > 0) {            for (TopicPartition topicPartition : topicPartitions) {                result.addAll(resolveTopicPartitionsList(topicPartition));            }        }        return result.toArray(new TopicPartitionInitialOffset[0]);    }    private String[] resolveTopics(LybGeekKafkaListener kafkaListener) {        String[] topics = kafkaListener.topics();        List<String> result = new ArrayList<>();        if (topics.length > 0) {            for (String topic1 : topics) {                Object topic = resolveExpression(topic1);                resolveAsString(topic, result);            }        }        return result.toArray(new String[0]);    }    private Pattern resolvePattern(LybGeekKafkaListener kafkaListener) {        Pattern pattern = null;        String text = kafkaListener.topicPattern();        if (StringUtils.hasText(text)) {            Object resolved = resolveExpression(text);            if (resolved instanceof Pattern) {                pattern = (Pattern) resolved;            }            else if (resolved instanceof String) {                pattern = Pattern.compile((String) resolved);            }            else if (resolved != null) {                throw new IllegalStateException(                        "topicPattern must resolve to a Pattern or String, not " + resolved.getClass());            }        }        return pattern;    }    private List<TopicPartitionInitialOffset> resolveTopicPartitionsList(TopicPartition topicPartition) {        Object topic = resolveExpression(topicPartition.topic());        Assert.state(topic instanceof String,                "topic in @TopicPartition must resolve to a String, not " + topic.getClass());        Assert.state(StringUtils.hasText((String) topic), "topic in @TopicPartition must not be empty");        String[] partitions = topicPartition.partitions();        PartitionOffset[] partitionOffsets = topicPartition.partitionOffsets();        Assert.state(partitions.length > 0 || partitionOffsets.length > 0,                "At least one 'partition' or 'partitionOffset' required in @TopicPartition for topic '" + topic + "'");        List<TopicPartitionInitialOffset> result = new ArrayList<>();        for (String partition : partitions) {            resolvePartitionAsInteger((String) topic, resolveExpression(partition), result);        }        for (PartitionOffset partitionOffset : partitionOffsets) {            TopicPartitionInitialOffset topicPartitionOffset =                    new TopicPartitionInitialOffset((String) topic,                            resolvePartition(topic, partitionOffset),                            resolveInitialOffset(topic, partitionOffset),                            isRelative(topic, partitionOffset));            if (!result.contains(topicPartitionOffset)) {                result.add(topicPartitionOffset);            }            else {                throw new IllegalArgumentException(                        String.format("@TopicPartition can't have the same partition configuration twice: [%s]",                                topicPartitionOffset));            }        }        return result;    }    private Integer resolvePartition(Object topic, PartitionOffset partitionOffset) {        Object partitionValue = resolveExpression(partitionOffset.partition());        Integer partition;        if (partitionValue instanceof String) {            Assert.state(StringUtils.hasText((String) partitionValue),                    "partition in @PartitionOffset for topic '" + topic + "' cannot be empty");            partition = Integer.valueOf((String) partitionValue);        }        else if (partitionValue instanceof Integer) {            partition = (Integer) partitionValue;        }        else {            throw new IllegalArgumentException(String.format(                    "@PartitionOffset for topic '%s' can't resolve '%s' as an Integer or String, resolved to '%s'",                    topic, partitionOffset.partition(), partitionValue.getClass()));        }        return partition;    }    private Long resolveInitialOffset(Object topic, PartitionOffset partitionOffset) {        Object initialOffsetValue = resolveExpression(partitionOffset.initialOffset());        Long initialOffset;        if (initialOffsetValue instanceof String) {            Assert.state(StringUtils.hasText((String) initialOffsetValue),                    "'initialOffset' in @PartitionOffset for topic '" + topic + "' cannot be empty");            initialOffset = Long.valueOf((String) initialOffsetValue);        }        else if (initialOffsetValue instanceof Long) {            initialOffset = (Long) initialOffsetValue;        }        else {            throw new IllegalArgumentException(String.format(                    "@PartitionOffset for topic '%s' can't resolve '%s' as a Long or String, resolved to '%s'",                    topic, partitionOffset.initialOffset(), initialOffsetValue.getClass()));        }        return initialOffset;    }    private boolean isRelative(Object topic, PartitionOffset partitionOffset) {        Object relativeToCurrentValue = resolveExpression(partitionOffset.relativeToCurrent());        Boolean relativeToCurrent;        if (relativeToCurrentValue instanceof String) {            relativeToCurrent = Boolean.valueOf((String) relativeToCurrentValue);        }        else if (relativeToCurrentValue instanceof Boolean) {            relativeToCurrent = (Boolean) relativeToCurrentValue;        }        else {            throw new IllegalArgumentException(String.format(                    "@PartitionOffset for topic '%s' can't resolve '%s' as a Boolean or String, resolved to '%s'",                    topic, partitionOffset.relativeToCurrent(), relativeToCurrentValue.getClass()));        }        return relativeToCurrent;    }    @SuppressWarnings("unchecked")    private void resolveAsString(Object resolvedValue, List<String> result) {        if (resolvedValue instanceof String[]) {            for (Object object : (String[]) resolvedValue) {                resolveAsString(object, result);            }        }        else if (resolvedValue instanceof String) {            result.add((String) resolvedValue);        }        else if (resolvedValue instanceof Iterable) {            for (Object object : (Iterable<Object>) resolvedValue) {                resolveAsString(object, result);            }        }        else {            throw new IllegalArgumentException(String.format(                    "@LybGeekKafkaListener can't resolve '%s' as a String", resolvedValue));        }    }    @SuppressWarnings("unchecked")    private void resolvePartitionAsInteger(String topic, Object resolvedValue,            List<TopicPartitionInitialOffset> result) {        if (resolvedValue instanceof String[]) {            for (Object object : (String[]) resolvedValue) {                resolvePartitionAsInteger(topic, object, result);            }        }        else if (resolvedValue instanceof String) {            Assert.state(StringUtils.hasText((String) resolvedValue),                    "partition in @TopicPartition for topic '" + topic + "' cannot be empty");            result.add(new TopicPartitionInitialOffset(topic, Integer.valueOf((String) resolvedValue)));        }        else if (resolvedValue instanceof Integer[]) {            for (Integer partition : (Integer[]) resolvedValue) {                result.add(new TopicPartitionInitialOffset(topic, partition));            }        }        else if (resolvedValue instanceof Integer) {            result.add(new TopicPartitionInitialOffset(topic, (Integer) resolvedValue));        }        else if (resolvedValue instanceof Iterable) {            for (Object object : (Iterable<Object>) resolvedValue) {                resolvePartitionAsInteger(topic, object, result);            }        }        else {            throw new IllegalArgumentException(String.format(                    "@LybGeekKafkaListener for topic '%s' can't resolve '%s' as an Integer or String", topic, resolvedValue));        }    }    private String resolveExpressionAsString(String value, String attribute) {        Object resolved = resolveExpression(value);        if (resolved instanceof String) {            return (String) resolved;        }        else if (resolved != null) {            throw new IllegalStateException("The [" + attribute + "] must resolve to a String. "                    + "Resolved to [" + resolved.getClass() + "] for [" + value + "]");        }        return null;    }    private Integer resolveExpressionAsInteger(String value, String attribute) {        Object resolved = resolveExpression(value);        Integer result = null;        if (resolved instanceof String) {            result = Integer.parseInt((String) resolved);        }        else if (resolved instanceof Number) {            result = ((Number) resolved).intValue();        }        else if (resolved != null) {            throw new IllegalStateException(                    "The [" + attribute + "] must resolve to an Number or a String that can be parsed as an Integer. "                            + "Resolved to [" + resolved.getClass() + "] for [" + value + "]");        }        return result;    }    private Boolean resolveExpressionAsBoolean(String value, String attribute) {        Object resolved = resolveExpression(value);        Boolean result = null;        if (resolved instanceof Boolean) {            result = (Boolean) resolved;        }        else if (resolved instanceof String) {            result = Boolean.parseBoolean((String) resolved);        }        else if (resolved != null) {            throw new IllegalStateException(                    "The [" + attribute + "] must resolve to a Boolean or a String that can be parsed as a Boolean. "                            + "Resolved to [" + resolved.getClass() + "] for [" + value + "]");        }        return result;    }    private Object resolveExpression(String value) {        return this.resolver.evaluate(resolve(value), this.expressionContext);    }    /**     * Resolve the specified value if possible.     * @param value the value to resolve     * @return the resolved value     * @see ConfigurableBeanFactory#resolveEmbeddedValue     */    private String resolve(String value) {        if (this.beanFactory != null && this.beanFactory instanceof ConfigurableBeanFactory) {            return ((ConfigurableBeanFactory) this.beanFactory).resolveEmbeddedValue(value);        }        return value;    }    private void addFormatters(FormatterRegistry registry) {        for (Converter<?, ?> converter : getBeansOfType(Converter.class)) {            registry.addConverter(converter);        }        for (GenericConverter converter : getBeansOfType(GenericConverter.class)) {            registry.addConverter(converter);        }        for (Formatter<?> formatter : getBeansOfType(Formatter.class)) {            registry.addFormatter(formatter);        }    }    private <T> Collection<T> getBeansOfType(Class<T> type) {        if (LybGeekKafkaListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ListableBeanFactory) {            return ((ListableBeanFactory) LybGeekKafkaListenerAnnotationBeanPostProcessor.this.beanFactory)                    .getBeansOfType(type)                    .values();        }        else {            return Collections.emptySet();        }    }    /**     * An {@link MessageHandlerMethodFactory} adapter that offers a configurable underlying     * instance to use. Useful if the factory to use is determined once the endpoints     * have been registered but not created yet.     * @see KafkaListenerEndpointRegistrar#setMessageHandlerMethodFactory     */    private class KafkaHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory {        private final DefaultFormattingConversionService defaultFormattingConversionService =                new DefaultFormattingConversionService();        private MessageHandlerMethodFactory messageHandlerMethodFactory;        public void setMessageHandlerMethodFactory(MessageHandlerMethodFactory kafkaHandlerMethodFactory1) {            this.messageHandlerMethodFactory = kafkaHandlerMethodFactory1;        }        @Override        public InvocableHandlerMethod createInvocableHandlerMethod(Object bean, Method method) {            return getMessageHandlerMethodFactory().createInvocableHandlerMethod(bean, method);        }        private MessageHandlerMethodFactory getMessageHandlerMethodFactory() {            if (this.messageHandlerMethodFactory == null) {                this.messageHandlerMethodFactory = createDefaultMessageHandlerMethodFactory();            }            return this.messageHandlerMethodFactory;        }        private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {            DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();            Validator validator = LybGeekKafkaListenerAnnotationBeanPostProcessor.this.registrar.getValidator();            if (validator != null) {                defaultFactory.setValidator(validator);            }            defaultFactory.setBeanFactory(LybGeekKafkaListenerAnnotationBeanPostProcessor.this.beanFactory);            ConfigurableBeanFactory cbf =                    LybGeekKafkaListenerAnnotationBeanPostProcessor.this.beanFactory instanceof ConfigurableBeanFactory ?                            (ConfigurableBeanFactory) LybGeekKafkaListenerAnnotationBeanPostProcessor.this.beanFactory :                            null;            this.defaultFormattingConversionService.addConverter(                    new BytesToStringConverter(LybGeekKafkaListenerAnnotationBeanPostProcessor.this.charset));            defaultFactory.setConversionService(this.defaultFormattingConversionService);            List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>();            // Annotation-based argument resolution            argumentResolvers.add(new HeaderMethodArgumentResolver(this.defaultFormattingConversionService, cbf));            argumentResolvers.add(new HeadersMethodArgumentResolver());            // Type-based argument resolution            final GenericMessageConverter messageConverter =                    new GenericMessageConverter(this.defaultFormattingConversionService);            argumentResolvers.add(new MessageMethodArgumentResolver(messageConverter));            argumentResolvers.add(new PayloadArgumentResolver(messageConverter, validator) {                @Override                public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {                    Object resolved = super.resolveArgument(parameter, message);                    /*                     * Replace KafkaNull list elements with null.                     */                    if (resolved instanceof List) {                        List<?> list = ((List<?>) resolved);                        for (int i = 0; i < list.size(); i++) {                            if (list.get(i) instanceof KafkaNull) {                                list.set(i, null);                            }                        }                    }                    return resolved;                }                @Override                protected boolean isEmptyPayload(Object payload) {                    return payload == null || payload instanceof KafkaNull;                }            });            defaultFactory.setArgumentResolvers(argumentResolvers);            defaultFactory.afterPropertiesSet();            return defaultFactory;        }    }    private static class BytesToStringConverter implements Converter<byte[], String> {        private final Charset charset;        BytesToStringConverter(Charset charset) {            this.charset = charset;        }        @Override        public String convert(byte[] source) {            return new String(source, this.charset);        }    }    private static class ListenerScope implements Scope {        private final Map<String, Object> listeners = new HashMap<>();        ListenerScope() {            super();        }        public void addListener(String key, Object bean) {            this.listeners.put(key, bean);        }        public void removeListener(String key) {            this.listeners.remove(key);        }        @Override        public Object get(String name, ObjectFactory<?> objectFactory) {            return this.listeners.get(name);        }        @Override        public Object remove(String name) {            return null;        }        @Override        public void registerDestructionCallback(String name, Runnable callback) {        }        @Override        public Object resolveContextualObject(String key) {            return this.listeners.get(key);        }        @Override        public String getConversationId() {            return null;        }    }}

业务侧如何使用

示例

@LybGeekKafkaListener(id = "createUser",topics = Constant.USER_TOPIC)public class UserComsumer extends BaseComusmeListener {    @Autowired    private UserService userService;    @Override    public boolean isRepeateConsume(KafkaComsumePayLoad kafkaComsumePayLoad) {        User user = JSON.parseObject(kafkaComsumePayLoad.getData(),User.class);        return userService.isExistUserByUsername(user.getUsername());    }    @Override    public boolean doBiz(KafkaComsumePayLoad kafkaComsumerPayLoad) {        User user = JSON.parseObject(kafkaComsumerPayLoad.getData(),User.class);        return userService.save(user);    }}

总结

有时候我们在宣导一些事情时，往往会发现即使我们已经说了N遍了，事情仍然会出现纰漏。这时候我们可以考虑把我们想宣导的东西工具化，通过工具来规范。比如有些业务，可能一些开发没考虑全面，我们就可以基于业务，把一些核心的场景抽象成方法，然后开发人员基于这些抽象方法，做具体实现。

demo链接

https://github.com/lyb-geek/springboot-learning/tree/master/springboot-kafka-template

原文：https://juejin.cn/post/7101110793864740877