聊聊spring kafka的retry-白红宇

聊聊spring kafka的retry

阅读量：6231 次

发布时间：2019-06-21

本文共 7958 字，大约阅读时间需要 26 分钟。

序

本文主要聊一下spring for kafka的retry

AbstractRetryingMessageListenerAdapter

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.java

主要有两个实现类RetryingAcknowledgingMessageListenerAdapter以及RetryingMessageListenerAdapter

RetryingAcknowledgingMessageListenerAdapter

public class RetryingAcknowledgingMessageListenerAdapter
    
             extends AbstractRetryingMessageListenerAdapter
     
      >        implements AcknowledgingMessageListener
      
        {    private final AcknowledgingMessageListener
       
         delegate;    /**     * Construct an instance with the provided template and delegate. The exception will     * be thrown to the container after retries are exhausted.     * @param messageListener the listener delegate.     * @param retryTemplate the template.     */    public RetryingAcknowledgingMessageListenerAdapter(AcknowledgingMessageListener
        
          messageListener,            RetryTemplate retryTemplate) {        this(messageListener, retryTemplate, null);    }    /**     * Construct an instance with the provided template, callback and delegate.     * @param messageListener the listener delegate.     * @param retryTemplate the template.     * @param recoveryCallback the recovery callback; if null, the exception will be     * thrown to the container after retries are exhausted.     */    public RetryingAcknowledgingMessageListenerAdapter(AcknowledgingMessageListener
         
           messageListener, RetryTemplate retryTemplate, RecoveryCallback
           recoveryCallback) { super(messageListener, retryTemplate, recoveryCallback); Assert.notNull(messageListener, "'messageListener' cannot be null"); this.delegate = messageListener; } @SuppressWarnings("unchecked") @Override public void onMessage(final ConsumerRecord
          
            record, final Acknowledgment acknowledgment) { getRetryTemplate().execute(new RetryCallback
           
            () { @Override public Void doWithRetry(RetryContext context) throws KafkaException { context.setAttribute(CONTEXT_RECORD, record); context.setAttribute(CONTEXT_ACKNOWLEDGMENT, acknowledgment); RetryingAcknowledgingMessageListenerAdapter.this.delegate.onMessage(record, acknowledgment); return null; } }, (RecoveryCallback

RetryingMessageListenerAdapter

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java

public class RetryingMessageListenerAdapter
    
             extends AbstractRetryingMessageListenerAdapter
     
      >        implements MessageListener
      
        {    /**     * Construct an instance with the provided template and delegate. The exception will     * be thrown to the container after retries are exhausted.     * @param messageListener the delegate listener.     * @param retryTemplate the template.     */    public RetryingMessageListenerAdapter(MessageListener
       
         messageListener, RetryTemplate retryTemplate) {        this(messageListener, retryTemplate, null);    }    /**     * Construct an instance with the provided template, callback and delegate.     * @param messageListener the delegate listener.     * @param retryTemplate the template.     * @param recoveryCallback the recovery callback; if null, the exception will be     * thrown to the container after retries are exhausted.     */    public RetryingMessageListenerAdapter(MessageListener
        
          messageListener, RetryTemplate retryTemplate,            RecoveryCallback
          recoveryCallback) {        super(messageListener, retryTemplate, recoveryCallback);        Assert.notNull(messageListener, "'messageListener' cannot be null");    }    @SuppressWarnings("unchecked")    @Override    public void onMessage(final ConsumerRecord
         
           record) { getRetryTemplate().execute(new RetryCallback
          
           () { @Override public Void doWithRetry(RetryContext context) throws KafkaException { context.setAttribute(CONTEXT_RECORD, record); RetryingMessageListenerAdapter.this.delegate.onMessage(record); return null; } }, (RecoveryCallback

具体是哪种listener呢

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

private void setupMessageListener(MessageListenerContainer container, MessageConverter messageConverter) {        Object messageListener = createMessageListener(container, messageConverter);        Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener");        if (this.retryTemplate != null) {            if (messageListener instanceof AcknowledgingMessageListener) {                messageListener = new RetryingAcknowledgingMessageListenerAdapter<>(                        (AcknowledgingMessageListener
    
     ) messageListener, this.retryTemplate,                        this.recoveryCallback);            }            else {                messageListener = new RetryingMessageListenerAdapter<>((MessageListener
     
      ) messageListener,                        this.retryTemplate, (RecoveryCallback

如果retryTemplate不为null的话，会先判断是不是AcknowledgingMessageListener的子类，如果是则创建RetryingAcknowledgingMessageListenerAdapter，如果不是则创建RetryingMessageListenerAdapter

spring-kafka-1.2.3.RELEASE-sources.jar!/org/springframework/kafka/config/MethodKafkaListenerEndpoint.java

protected MessagingMessageListenerAdapter
    
      createMessageListenerInstance(MessageConverter messageConverter) {        if (isBatchListener()) {            BatchMessagingMessageListenerAdapter
     
       messageListener = new BatchMessagingMessageListenerAdapter
      
       (                    this.bean, this.method);            if (messageConverter instanceof BatchMessageConverter) {                messageListener.setBatchMessageConverter((BatchMessageConverter) messageConverter);            }            return messageListener;        }        else {            RecordMessagingMessageListenerAdapter
       
         messageListener =                    new RecordMessagingMessageListenerAdapter
        
         (this.bean, this.method);            if (messageConverter instanceof RecordMessageConverter) {                messageListener.setMessageConverter((RecordMessageConverter) messageConverter);            }            return messageListener;        }    }

其中RecordMessagingMessageListenerAdapter实现了AcknowledgingMessageListener接口

public class RecordMessagingMessageListenerAdapter
    
      extends MessagingMessageListenerAdapter
     
              implements MessageListener
      
       , AcknowledgingMessageListener
       
         {        //......}