org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

发送消息

@Autowired
private KafkaTemplate kafkaTemplate;

public void myAction(){
    // 定义 data
    // 任务推送到 Kafka
    kafkaTemplate.send(“myJob", data.toString());
}

接收消息

@SpringBootApplication
@ServletComponentScan
public class Application {
    
    @KafkaListener(topics = “myJob”)
    @SendTo(“myJob-delay")
    public String onMessage(ConsumerRecord<?, ?> cr, Acknowledgment ack) {
        String json = (String) cr.value();
        JSONObject data = JSON.parseObject(json);

        if (/* 需要延迟处理 */){
            // 提交
            ack.acknowledge();
            // 发送到 @SendTo
           data.put("until", System.currentTimeMillis() + msToDelay);
           return data.toString();
        }

        // 正常处理
        // do real work

        // 提交
        ack.acknowledge();
        return null;
    }

    @KafkaListener(topics = “myJob-delay")
    @SendTo(“myJob")
    public String delayMessage(ConsumerRecord<?, ?> cr, Acknowledgment ack){
        
        String json = (String) cr.value();
        JSONObject data = JSON.parseObject(json);
        Long until = data.getLong("until");
        // 阻塞直到 until
        while (System.currentTimeMillis() < until){
           Thread.sleep( Math.max(0, until - System.currentTimeMillis()) );
        }

        // 提交
        ack.acknowledge();
        // 转移到 @SendTo
        return json;

    }
}

代码很简单，不用解释也能看明白。稍微提一下几个重要的地方。

@KafkaListener 的方法参数里有 Acknowledgment ack，这是AckMode.MANUAL 模式下必须要添加的参数。

ack.acknowledge() 用来标记一条消息已经消费完成，即将从消息队列里移除。执行之前消息会一直保留在队列中，即时宕机重启后也能恢复。

@SendTo 用来在队列（topic）间转移消息，只要 return 非 null 的数据。以上代码中，当需要延迟处理时，消息从 myJob 转移到 myJob-delay；而当条件满足时，消息又从 myJob-delay 转移到了 myJob。

自从 spring-kafka 2.2.4 版本之后，可以在方法上定义 max.poll.interval.ms ，更加灵活了。例如

@KafkaListener(topics = "myTopic", groupId = "group", properties = { 
    "max.poll.interval.ms:60000”, 
    ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=100”}
)

    以上是延迟消息处理的简单实现，适合延时要求不那么高的场合。朋友们想一下，假如延时比较复杂，执行的次序也不一定跟消息到达的次序一致，系统又该怎样设计呢？

假如这篇文章对你有所帮助, 请关注我公众号, 发现更多有用的文章