在实际开发中，我们常常需要使用转发功能实现业务解耦。比如：应用 A 从 topic1 获取到消息，经过处理后转发到 topic2。应用 B 监听 topic2 获取消息再次进行处理。 十一、消费者6：消息转发 （1）Spring-Kafka 只需要通过一个 @SendTo 注解即可以实现消息的转发......

消息过滤器可以让消息在抵达监听容器前被拦截，过滤器根据系统业务逻辑去筛选出需要的数据交由 KafkaListener 处理，不需要的消息则会过滤掉。 十、消费者5：消息过滤器 1，配置消息过滤器 配置消息过滤器十分简单，只需要为监听容器工厂配置一个......

通常来说 KafkaListener 要做的事只是监听 Topic 中的数据并消费，如果在 KafkaListener 中还需要对异常进行 try catch 捕获并处理的话，则会显得代码块非常臃肿不利于维护。 好在 spring-kafka 为我们提供了专门的异常处理器（ConsumerAwareListenerErrorHandler......

由于 Kafka 的写性能非常高，因此项目经常会碰到 Kafka 消息队列拥堵的情况。遇到这种情况，我们可以通过并发消费、批量消费的方法进行解决。 八、消费者3：并发、批量消费 1，批量消费 （1）首先我们在项目 application.properties 文件中添加如下配置......

之前的样例中消费者这边都直接获取消息内容并使用，如果我们还想要获取分区信息、消息头等其他内容的话，有如下两种方式。 七、消费者2：获取消息头和消息体 1，使用 ConsumerRecord 类方式 （1）使用 ConsumerRecord 类接收有一定的好处......

六、消费者1：指定 topic、partition、offset 1，使用 topics 指定 topic （1）监听器主要是使用 @KafkaListenter 注解即可，而通过 topics 参数设置监听的 topic（可监听多个，用逗号隔开）： 其他参数介绍：id（消费者 ID）、 groupId（消费组 ID......

Kafka 同数据库一样支持事务，当发生异常或者出现特定逻辑判断的时候可以进行回滚，确保消息监听器不会接收到一些错误的或者不需要的消息。Kafka 使用事务有两种方式，下面分别进行介绍。 五、生产者3：使用事务 1，使用 executeInTransaction 方法......

当我们发送消息到 Kafka 后，有时我们需要确认消息是否发送成功，如果消息发送失败，就要重新发送或者执行对应的业务逻辑。下面分别演示如何在异步或者同步发送消息时，获取发送结果。 四、生产者2：消息回调、同步异步发送消息 1，获取异步发送消息的结果......

三、生产者1：指定 topic、partition、key 等 1，send() 方法 （1）在之前的文章中我们都是通过 KafkaTemplate 的 send() 方法指定一个 topic 发送消息，其实 send() 方法还支持其他参数，具体如下： 参数说明： topic：这里填写的是 Topic 的名字 partition......

在之前的文章样例中，我们发送消息的时候并没有事先创建相应的 Topic。这是因为 KafkaTemplate 在发送的时候就已经帮我们完成了创建的操作。 但这样也会存在一些问题，比如这种情况创建出来的 Topic 的 Partition（分区)）数永远只有 1 个，也不会有副本......