5.2.SpringBoot整合Kafka（开整）

1、准备工作

pom配置:

    <dependency>
        <groupid>org.springframework.kafka</groupid>
        <artifactid>spring-kafka</artifactid>
        <version>2.8.8</version>
    </dependency>

application-dev.yml配置 (配置在spring下)：

 kafka:
    bootstrap-servers:
      - localhost:9092
    template:
      default-topic: demo.topic
    producer:
      retries: 3 # 重试次数，设置大于0的值，则客户端会将发送失败的记录重新发送
      batch-size: 16384 #批量处理大小，16K
      buffer-memory: 33554432 #缓冲存储大，32M
      acks: 1
      # 指定消息key和消息体的编解码方式
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      # 是否自动提交(手动提交要关闭，不然会报错)
      group-id: spring_customer
      enable-auto-commit: false
      # 消费偏移配置
      # none：如果没有为消费者找到先前的offset的值,即没有自动维护偏移量,也没有手动维护偏移量,则抛出异常
      # earliest：在各分区下有提交的offset时：从offset处开始消费；在各分区下无提交的offset时：从头开始消费
      # latest：在各分区下有提交的offset时：从offset处开始消费；在各分区下无提交的offset时：从最新的数据开始消费
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    # 监听
    listener:
      # record：当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
      # batch：当每一批poll()的数据被ListenerConsumer处理之后提交
      # time：当每一批poll()的数据被ListenerConsumer处理之后，距离上次提交时间大于TIME时提交
      # count：当每一批poll()的数据被ListenerConsumer处理之后，被处理record数量大于等于COUNT时提交
      # count_time：TIME或COUNT中有一个条件满足时提交
      # manual：当每一批poll()的数据被ListenerConsumer处理之后, 手动调用Acknowledgment.acknowledge()后提交
      # manual_immediate：手动调用Acknowledgment.acknowledge()后立即提交，一般推荐使用这种
      ack-mode: manual_immediate
      # 在侦听器容器中运行的线程数。
      concurrency: 5
 # 其它属性配置
 properties:
 # 设置发送消息的大小
    max.request.size: 10240000

2、创建生产者和消费者

生产者：

/**
 * cf
 * 生产者
 */
@RestController
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<string, object> kafkaTemplate;

    /**
     * &#x53D1;&#x9001;&#x6D88;&#x606F;
     * @param message
     */
    @PostMapping("/kafka")
    public void sendMessage1(@RequestBody String message) {
        kafkaTemplate.send("topic_test01", "cfKey",message);
    }

}
</string,>

消费者：

/**
 * cf
 * 消费者
 */
@Component
@Slf4j
public class KafkaConsumer {
    /**
     * kafka的监听器 消费消息
     * @param record
     * @param item
     */
    @KafkaListener(topics = "topic_test01", groupId = "spring_customer")
    public void topicListener(ConsumerRecord<string, string> record, Acknowledgment item) {
        log.info("&#x6211;&#x5F00;&#x59CB;&#x6D88;&#x8D39;&#xFF1A;{}==,{}==,{};",record.topic(),record.partition(),record.value());
        //&#x624B;&#x52A8;&#x63D0;&#x4EA4;
        item.acknowledge();
    }
}
</string,>

在 Kafka 中消息通过服务器推送给各个消费者，而 Kafka 的消费者在消费消息时，需要提供一个监听器（Listener）对某个 Topic 实现监听，从而获取消息，这也是 Kafka 消费消息的唯一方式。

@KafkaListener 注解实现监听器，就是我们用到的。

源码：

@Target({ ElementType.TYPE, ElementType.METHOD, ElementType.ANNOTATION_TYPE })
@Retention(RetentionPolicy.RUNTIME)
@MessageMapping
@Documented
@Repeatable(KafkaListeners.class)
public @interface KafkaListener {
    String id() default "";
    String containerFactory() default "";
    //消息 Topic
    String[] topics() default {};
    //Topic 的模式匹配表达式
    String topicPattern() default "";
    //Topic 分区
    TopicPartition[] topicPartitions() default {};
    String containerGroup() default "";
    String errorHandler() default "";
    //消息分组 Id
    String groupId() default "";
    boolean idIsGroup() default true;
    String clientIdPrefix() default "";
    String beanRef() default "__listener";
}

在使用 @KafkaListener 时，最核心的操作是设置 Topic，而 Kafka 还提供了一个模式匹配表达式可以对目标 Topic 实现灵活设置。
在这里，强调下 groupId 这个属性，这就涉及 Kafka 中另一个核心概念： 消费者分组（Consumer Group）。

设计消费者组的目的是应对集群环境下的多服务实例问题。

显然，如果采用发布-订阅模式会导致一个服务的不同实例可能会消费到同一条消息。

为了解决这个问题，Kafka 中提供了消费者组的概念。 一旦我们使用了消费组，一条消息只能被同一个组中的某一个服务实例所消费。

3.测试一下

成功：

1. 复杂操作1 （ConcurrentKafkaListenerContainerFactory）应用

kafka过滤器

1.我们在使用kafka的时候，会遇到某一些消息过滤的情况，所以我们需要配置过滤器，来清洗kafka中的数据。
2.有时候我们需要配置并发消费，所以要Specify the container concurrency.指定容器并发性

配置如下：

/**
 * cf
 * 消费者消息过滤器
 */
@Configuration
public class CustomKafkaFilter {
    @Autowired
    private ConsumerFactory consumerFactory;
    @Bean("concurrentKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(){
        ConcurrentKafkaListenerContainerFactory factory =
                new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        //指定容器并发性（配置文件配置也可）
        factory.setConcurrency(3);
        //开启批量消费 （配置文件配置也可）
        factory.setBatchListener(true);
        // 被过滤的消息将被丢弃
        factory.setAckDiscarded(true);
        // 设置记录筛选策略
        factory.setRecordFilterStrategy(new RecordFilterStrategy() {
            @Override
            public boolean filter(ConsumerRecord consumerRecord) {
                String msg = consumerRecord.value().toString();
                 //这里写筛选规则
                return true;// 返回true消息将会被丢弃
            }
        });
        return factory;
    }
}

注意： @Bean(“concurrentKafkaListenerContainerFactory”)一定要指定名称，否则spring会认为没有这个bean自动创建一个，会出现重复执行问题。

1. 复杂操作2 封装 kafka生产者

工具类KafkaAnalyzeProducer：

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * kafka生产者
 */
@Slf4j
@Component
public class KafkaAnalyzeProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 发送数据到kafka
     *
     * @param topic        topic名称
     * @param message      发送信息字符串
     * @param sendCallBack 发送回调
     */
    public void send(String topic, String message, SendCallBack sendCallBack) {

        ListenableFuture listenableFuture = kafkaTemplate.send(topic, message);
        //发送成功后回调
        SuccessCallback successCallback = new SuccessCallback() {
            @SneakyThrows
            @Override
            public void onSuccess(Object result) {
                sendCallBack.sendSuccessCallBack(topic, message);
            }
        };
        //发送失败回调
        FailureCallback failureCallback = new FailureCallback() {
            @SneakyThrows
            @Override
            public void onFailure(Throwable ex) {
                sendCallBack.sendFailCallBack(topic, message, ex);
            }
        };

        listenableFuture.addCallback(successCallback, failureCallback);
    }

    /**
     * producer 同步方式发送数据
     *
     * @param topic   topic名称
     * @param message producer发送的数据
     */
    public void sendAsynchronize(String topic, String message) throws InterruptedException, ExecutionException, TimeoutException {
        kafkaTemplate.send(topic, message).get(10, TimeUnit.SECONDS);
    }

    /**
     * producer 异步方式发送数据
     *
     * @param topic   topic名称
     * @param message producer发送的数据
     */
    public void sendSynchronize(String topic, String message) {
        kafkaTemplate.send(topic, message).addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable throwable) {
                log.error("----事件kafka记录解析完成放入topic：{}，发送失败{}", topic, message, throwable);
            }

            @Override
            public void onSuccess(Object o) {
                log.info("----事件kafka记录解析完成放入topic：{}，发送成功:{}", topic, message);
            }
        });
    }

    public void sendMessage(String topic, String message) {
        this.kafkaTemplate.send(topic, message);
    }
}

SendCallBack:

import java.text.ParseException;

public interface SendCallBack {

    /**
     * 生产成功回调
     * @param topic topic
     * @param msg 信息字符串
     */
    void sendSuccessCallBack(String topic,String msg) throws ParseException;

    /**
     * 生产失败回调
     * @param topic topic
     * @param msg 信息字符串
     * @param ex 异常
     */
    void sendFailCallBack(String topic,String msg,Throwable ex) throws ParseException;

}

使用方式：

    @Autowired
    private KafkaAnalyzeProducer kafkaAnalyzeProducer;

// 下面的放方法里面就好了
kafkaAnalyzeProducer.send(topic,JSON.toJSONString(itemMap,SerializerFeature.WriteMapNullValue),new SendCallBack() {
        @Override
        public void sendSuccessCallBack (String topic, String msg){
            JSONObject jsonObject = JSONObject.parseObject(msg);

            log.info("----sendSuccessCallBack事件kafka记录解析完成放入topic：{}，发送成功:{}", topic, jsonObject.getOrDefault("id", "").toString());
        }

        @Override
        public void sendFailCallBack (String topic, String msg, Throwable ex){
            JSONObject jsonObject = JSONObject.parseObject(msg);

            log.error("----sendFailCallBack事件kafka记录解析完成放入topic：{}，发送失败{}", topic, jsonObject.getOrDefault("id", "").toString(), ex);
        }
    });

ok基本上就分享这些就够用了。

Original: https://www.cnblogs.com/daohangtaiqian/p/16537782.html
Author: 道行太浅
Title: 5.2.SpringBoot整合Kafka（开整）