SpringBoot整合kafka(实现producer和consumer)

男娘i 2023-07-01 15:55 47阅读 0赞

转载自 [SpringBoot整合kafka(实现producer和consumer)][SpringBoot_kafka_producer_consumer]

在Windows环境下安装运行Kafka：[https://www.jianshu.com/p/d64798e81f3b][https_www.jianshu.com_p_d64798e81f3b]

本文代码使用的是Spring Boot 2.1.1.RELEASE 版本

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.1.RELEASE</version>
        <relativePath/> 
    </parent>

### 一、 pom.xml文件，引入依赖 ###

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
        </dependency>
    
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化

**1、在application-dev.properties配置生产者**

#============== kafka ===================
    # 指定kafka server的地址，集群配多个，中间，逗号隔开
    spring.kafka.bootstrap-servers=127.0.0.1:9092
    
    #=============== provider  =======================
    # 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，
    # 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
    spring.kafka.producer.retries=0
    # 每次批量发送消息的数量,produce积累到一定数据，一次发送
    spring.kafka.producer.batch-size=16384
    # produce积累数据一次发送，缓存大小达到buffer.memory就发送数据
    spring.kafka.producer.buffer-memory=33554432
    
    #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
    #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
    #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
    #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
    #可以设置的值为：all, -1, 0, 1
    spring.kafka.producer.acks=1
    
    # 指定消息key和消息体的编解码方式
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

*  bootstrap.servers:kafka server的地址
 *  acks:写入kafka时，leader负责一个该partion读写，当写入partition时，需要将记录同步到repli节点，all是全部同步节点都返回成功，leader才返回ack。
 *  retris:写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败，当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。
 *  batch.size:produce积累到一定数据，一次发送。

buffer.memory: produce积累数据一次发送，缓存大小达到buffer.memory就发送数据。

*  linger.ms :当设置了缓冲区，消息就不会即时发送，如果消息总不够条数、或者消息不够buffer大小就不发送了吗？当消息超过linger时间，也会发送。
 *  key/value serializer：序列化类。

**2、生产者向kafka发送消息**

@RestController
    public class KafkaController {
    
        @Autowired
        private KafkaTemplate<String,Object> kafkaTemplate;
    
        @GetMapping("/message/send")
        public boolean send(@RequestParam String message){
            kafkaTemplate.send("testTopic",message);
            return true;
        }
    
    }

**3、在application-dev.properties配置消费者**

#=============== consumer  =======================
    # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名
    spring.kafka.consumer.group-id=testGroup
    # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
    spring.kafka.consumer.auto-offset-reset=earliest
    # enable.auto.commit:true --> 设置自动提交offset
    spring.kafka.consumer.enable-auto-commit=true
    #如果'enable.auto.commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。
    spring.kafka.consumer.auto-commit-interval=100
    
    # 指定消息key和消息体的编解码方式
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

*  Producer是一个接口，声明了同步send和异步send两个重要方法。
 *  ProducerRecord 消息实体类，每条消息由（topic,key,value,timestamp)四元组封装。一条消息key可以为空和timestamp可以设置当前时间为默认值。

**4、消费者监听topic=testTopic的消息**

@Component
    public class ConsumerListener {
    
        @KafkaListener(topics = "testTopic")
        public void onMessage(String message){
            //insertIntoDb(buffer);//这里为插入数据库代码
            System.out.println(message);
        }
    
    }

到此，采用Kafka提供的StringSerializer和StringDeserializer进行序列化和反序列化，因为此种序列化方式无法序列化实体类，顾，下面为自定义序列化和反序列化器进行实体类的消息传递

采用自定义序列化和反序列化器进行实体类的序列化和反序列化

和内置的StringSerializer字符串序列化一样，如果要自定义序列化方式，需要实现接口**Serializer**。假设每个字段按照下图所示的方式自定义序列化：

![format_png][]

**1、创建User实体类**

public class User implements Serializable {
    
        private Long id;
    
        private String name;
    
        private Integer age;
    
        /**
         * transient 关键字修饰的字段不会被序列化
         */
        private transient String desc;
    
        public Long getId() {
            return id;
        }
    
        public void setId(Long id) {
            this.id = id;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public Integer getAge() {
            return age;
        }
    
        public void setAge(Integer age) {
            this.age = age;
        }
    
        public String getDesc() {
            return desc;
        }
    
        public void setDesc(String desc) {
            this.desc = desc;
        }
    
        @Override
        public String toString() {
            return "User{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", age=" + age +
                    ", desc='" + desc + '\'' +
                    '}';
        }
    }

**2、创建User序列化器**

public class UserSerializable implements Serializer<User> {
        @Override
        public void configure(Map<String, ?> map, boolean b) {
    
        }
    
        @Override
        public byte[] serialize(String topic, User user) {
            System.out.println("topic : " + topic + ", user : " + user);
            byte[] dataArray = null;
            ByteArrayOutputStream outputStream = null;
            ObjectOutputStream objectOutputStream = null;
            try {
                outputStream = new ByteArrayOutputStream();
                objectOutputStream = new ObjectOutputStream(outputStream);
                objectOutputStream.writeObject(user);
                dataArray = outputStream.toByteArray();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }finally {
                if(outputStream != null){
                    try {
                        outputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if(objectOutputStream != null){
                    try {
                        objectOutputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            return dataArray;
        }
    
        @Override
        public void close() {
    
        }
    }

**3、创建User反序列化器**

public class UserDeserializer implements Deserializer<User> {
        @Override
        public void configure(Map<String, ?> map, boolean b) {
    
        }
    
        @Override
        public User deserialize(String topic, byte[] bytes) {
            User user = null;
            ByteArrayInputStream inputStream = null;
            ObjectInputStream objectInputStream = null;
            try {
                inputStream = new ByteArrayInputStream(bytes);
                objectInputStream = new ObjectInputStream(inputStream);
                user = (User)objectInputStream.readObject();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }finally {
                if(inputStream != null){
                    try {
                        inputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if(objectInputStream != null){
                    try {
                        objectInputStream.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
            return user;
        }
    
        @Override
        public void close() {
    
        }
    }

**4、修改application-dev.properties配置**

**A、修改生产者配置的value-serializer**

# 指定生产者消息key和消息体的编解码方式
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=com.yibo.springbootkafkademo.Serializable.UserSerializable

**B、修改消费者配置的value-deserializer**

# 指定消费者消息key和消息体的编解码方式
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=com.yibo.springbootkafkademo.Serializable.UserDeserializer

**5、生产者向kafka发送消息**

@RestController
    public class KafkaController {
    
        @Autowired
        private KafkaTemplate<String,Object> kafkaTemplate;
    
        @PostMapping("/user/save")
        public boolean saveUser(@RequestBody User user){
            kafkaTemplate.send("userTopic",user);
            return true;
        }
    }

**6、消费者监听topic=userTopic的消息**

@Component
    public class ConsumerListener {
    
        @KafkaListener(topics = "userTopic")
        public void onMessage(User user){
            //insertIntoDb(buffer);//这里为插入数据库代码
            System.out.println(user);
        }
    }

### 总结 ###

可以看到，自定义Serializer和Deserializer非常痛苦，还有很多类型不支持，非常脆弱。复杂类型的支持更是一件痛苦的事情，不同版本之间的兼容性问题更是一个极大的挑战。由于Serializer和Deserializer影响到上下游系统，导致牵一发而动全身。自定义序列化&反序列化实现不是能力的体现，而是逗比的体现。所以强烈不建议自定义实现序列化&反序列化，推荐直接使用StringSerializer和StringDeserializer，然后使用json作为标准的数据传输格式。站在巨人的肩膀上，事半功倍。

[SpringBoot_kafka_producer_consumer]: https://www.jianshu.com/p/5da86afed228
[https_www.jianshu.com_p_d64798e81f3b]: https://www.jianshu.com/p/d64798e81f3b
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy82OTE4OTk1LTcxZTUxZjRhYWUxMWUyNzYucG5nP2ltYWdlTW9ncjIvYXV0by1vcmllbnQvc3RyaXB8aW1hZ2VWaWV3Mi8yL3cvODYwL2Zvcm1hdC93ZWJw?x-oss-process=image/format,png