kafka消费者API之自定义存储offset 到mysql中

约定不等于承诺〃 2022-12-03 08:50 128阅读 0赞

pom文件

<dependencies>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>0.11.0.0</version>
            </dependency>
            
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.47</version>
            </dependency>
        </dependencies>

1.  定义producer
    
        package com.cw.kafka.consumer.mysql;
        
        import org.apache.kafka.clients.producer.*;
        import org.apache.kafka.common.serialization.StringSerializer;
        
        import java.math.BigInteger;
        import java.security.MessageDigest;
        import java.security.NoSuchAlgorithmException;
        import java.util.Properties;
        import java.util.Random;
        
        /**
         * @author 陈小哥cw
         * @date 2020/6/19 19:22
         */
        public class KafkaProducerTest {
            static Properties properties = null;
            static KafkaProducer<String, String> producer = null;
        
            static {
                properties = new Properties();
                // kafka集群，broker-list
                properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "cm1:9092,cm2:9092,cm3:9092");
                properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
                properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
                // 重试次数
                properties.put(ProducerConfig.ACKS_CONFIG, "all");
                // 批次大小
                properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
                // 等待时间
                properties.put(ProducerConfig.LINGER_MS_CONFIG, 100);
                // RecordAccumulator缓冲区大小
                properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        
                producer = new KafkaProducer<String, String>(properties);
            }
        
            public static void main(String[] args) throws NoSuchAlgorithmException {
                for (int i = 0; i < 100; i++) {
                    System.out.println("第" + (i + 1) + "条消息开始发送");
                    sendData();
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                producer.close();
            }
        
            public static String generateHash(String input) throws NoSuchAlgorithmException {
                MessageDigest digest = MessageDigest.getInstance("MD5");
                int random = new Random().nextInt(1000);
                digest.update((input + random).getBytes());
                byte[] bytes = digest.digest();
                BigInteger bi = new BigInteger(1, bytes);
                String string = bi.toString(16);
                return string.substring(0, 3) + input + random;
            }
        
            public static void sendData() throws NoSuchAlgorithmException {
                String topic = "mysql_store_offset";
        
                producer.send(new ProducerRecord<String, String>(
                                topic,
                                generateHash(topic),
                                new Random().nextInt(1000) + "\t金锁家庭财产综合险(家顺险)\t1\t金锁家庭财产综合险(家顺险)\t213\t自住型家财险\t10\t家财保险\t44\t人保财险\t23:50.0"),
                        new Callback() {
                            @Override
                            public void onCompletion(RecordMetadata metadata, Exception exception) {
                                if (exception != null) {
                                    System.out.println("|----------------------------\n|topic\tpartition\toffset\n|" + metadata.topic() + "\t" + metadata.partition() + "\t" + metadata.offset() + "\n|----------------------------");
                                } else {
                                    exception.printStackTrace();
                                }
                            }
                        });
            }
        }
2.  创建消费者
    
        package com.cw.kafka.consumer.mysql;
        
        import org.apache.kafka.clients.consumer.*;
        import org.apache.kafka.common.TopicPartition;
        import org.apache.kafka.common.serialization.StringDeserializer;
        
        import java.text.SimpleDateFormat;
        import java.util.*;
        
        /**
         * @author 陈小哥cw
         * @date 2020/6/19 20:12
         */
        public class KafkaConsumerTest {
            private static Properties properties = null;
            private static String group = "mysql_offset";
            private static String topic = "mysql_store_offset";
            private static KafkaConsumer<String, String> consumer;
        
            static {
                properties = new Properties();
                // kafka集群，broker-list
                properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "cm1:9092,cm2:9092,cm3:9092");
        
                properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                // 消费者组，只要group.id相同，就属于同一个消费者组
                properties.put(ConsumerConfig.GROUP_ID_CONFIG, group);
                // 关闭自动提交offset
                properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        
                // 1.创建一个消费者
                consumer = new KafkaConsumer<>(properties);
            }
        
            public static void main(String[] args) {
        
                consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
        
                    // rebalance之前将记录进行保存
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                        for (TopicPartition partition : partitions) {
                            // 获取分区
                            int sub_topic_partition_id = partition.partition();
                            // 对应分区的偏移量
                            long sub_topic_partition_offset = consumer.position(partition);
                            String date = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss").format(
                                    new Date(
                                            new Long(
                                                    System.currentTimeMillis()
                                            )
                                    )
                            );
        
                            DBUtils.update("replace into offset values(?,?,?,?,?)",
                                    new Offset(
                                            group,
                                            topic,
                                            sub_topic_partition_id,
                                            sub_topic_partition_offset,
                                            date
                                    )
                            );
                        }
                    }
        
                    // rebalance之后读取之前的消费记录，继续消费
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                        for (TopicPartition partition : partitions) {
                            int sub_topic_partition_id = partition.partition();
                            long offset = DBUtils.queryOffset(
                                    "select sub_topic_partition_offset from offset where consumer_group=? and sub_topic=? and sub_topic_partition_id=?",
                                    group,
                                    topic,
                                    sub_topic_partition_id
                            );
                            System.out.println("partition = " + partition + "offset = " + offset);
                            // 定位到最近提交的offset位置继续消费
                            consumer.seek(partition, offset);
        
                        }
                    }
                });
        
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(100);
                    List<Offset> offsets = new ArrayList<>();
                    for (ConsumerRecord<String, String> record : records) {
                        String date = new SimpleDateFormat("yyyy年MM月dd日 HH:mm:ss").format(
                                new Date(
                                        new Long(
                                                System.currentTimeMillis()
                                        )
                                )
                        );
                        offsets.add(new Offset(group, topic, record.partition(), record.offset(), date));
        
                        System.out.println("|---------------------------------------------------------------\n" +
                                "|group\ttopic\tpartition\toffset\ttimestamp\n" +
                                "|" + group + "\t" + topic + "\t" + record.partition() + "\t" + record.offset() + "\t" + record.timestamp() + "\n" +
                                "|---------------------------------------------------------------"
                        );
                    }
                    for (Offset offset : offsets) {
                        DBUtils.update("replace into offset values(?,?,?,?,?)", offset);
                    }
                    offsets.clear();
                }
            }
        }
3.  其他类
    
    数据库工具类
    
        package com.cw.kafka.consumer.mysql;
        
        import java.io.IOException;
        import java.sql.*;
        import java.util.Properties;
        
        /**
         * JDBC操作工具类, 提供注册驱动, 连接, 发送器, 动态绑定参数, 关闭资源等方法
         * jdbc连接参数的提取, 使用Properties进行优化(软编码)
         *
         * @author 陈小哥cw
         * @date 2020/6/19 19:41
         */
        public class DBUtils {
        
            private static String driver;
            private static String url;
            private static String user;
            private static String password;
        
            static {
                // 借助静态代码块保证配置文件只读取一次就行
                // 创建Properties对象
                Properties prop = new Properties();
                try {
                    // 加载配置文件, 调用load()方法
                    // 类加载器加载资源时, 去固定的类路径下查找资源, 因此, 资源文件必须放到src目录才行
                    prop.load(DBUtils.class.getClassLoader().getResourceAsStream("db.properties"));
                    // 从配置文件中获取数据为成员变量赋值
                    driver = prop.getProperty("db.driver").trim();
                    url = prop.getProperty("db.url").trim();
                    user = prop.getProperty("db.user").trim();
                    password = prop.getProperty("db.password").trim();
                    // 加载驱动
                    Class.forName(driver);
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
            }
        
            /**
             * 动态绑定参数
             *
             * @param pstmt
             * @param params
             */
            public static void bindParam(PreparedStatement pstmt, Object... params) {
                try {
                    for (int i = 0; i < params.length; i++) {
                        pstmt.setObject(i + 1, params[i]);
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        
        
            /**
             * 预处理发送器
             *
             * @param conn
             * @param sql
             * @return
             */
            public static PreparedStatement getPstmt(Connection conn, String sql) {
                PreparedStatement pstmt = null;
                try {
                    pstmt = conn.prepareStatement(sql);
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                return pstmt;
            }
        
            /**
             * 获取发送器的方法
             *
             * @param conn
             * @return
             */
            public static Statement getStmt(Connection conn) {
                Statement stmt = null;
                try {
                    stmt = conn.createStatement();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                return stmt;
            }
        
            /**
             * 获取数据库连接的方法
             *
             * @return
             */
            public static Connection getConn() {
                Connection conn = null;
                try {
                    conn = DriverManager.getConnection(url, user, password);
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                return conn;
            }
        
            /**
             * 获取特定消费者组，主题，分区下的偏移量
             *
             * @return offset
             */
            public static long queryOffset(String sql, Object... params) {
                Connection conn = getConn();
                long offset = 0;
                PreparedStatement preparedStatement = getPstmt(conn, sql);
                bindParam(preparedStatement, params);
        
                ResultSet resultSet = null;
                try {
                    resultSet = preparedStatement.executeQuery();
                    while (resultSet.next()) {
                        offset = resultSet.getLong("sub_topic_partition_offset");
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                } finally {
                    close(resultSet, preparedStatement, conn);
                }
        
                return offset;
            }
        
            /**
             * 根据特定消费者组，主题，分区，更新偏移量
             *
             * @param offset
             */
            public static void update(String sql, Offset offset) {
                Connection conn = getConn();
                PreparedStatement preparedStatement = getPstmt(conn, sql);
        
                bindParam(preparedStatement,
                        offset.getConsumer_group(),
                        offset.getSub_topic(),
                        offset.getSub_topic_partition_id(),
                        offset.getSub_topic_partition_offset(),
                        offset.getTimestamp()
                );
        
                try {
                    preparedStatement.executeUpdate();
                } catch (SQLException e) {
                    e.printStackTrace();
                } finally {
                    close(null, preparedStatement, conn);
                }
        
            }
        
        
            /**
             * 统一关闭资源
             *
             * @param rs
             * @param stmt
             * @param conn
             */
            public static void close(ResultSet rs, Statement stmt, Connection conn) {
                try {
                    if (rs != null) {
                        rs.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                try {
                    if (stmt != null) {
                        stmt.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                try {
                    if (conn != null) {
                        conn.close();
                    }
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        
            public static void main(String[] args) {
        
            }
        }
    
    offset实体类
    
        package com.cw.kafka.consumer.mysql;
        
        /**
         * @author 陈小哥cw
         * @date 2020/6/19 20:00
         */
        public class Offset {
            private String consumer_group;
            private String sub_topic;
            private Integer sub_topic_partition_id;
            private Long sub_topic_partition_offset;
            private String timestamp;
        
            public Offset() {
            }
        
            public Offset(String consumer_group, String sub_topic, Integer sub_topic_partition_id, Long sub_topic_partition_offset, String timestamp) {
                this.consumer_group = consumer_group;
                this.sub_topic = sub_topic;
                this.sub_topic_partition_id = sub_topic_partition_id;
                this.sub_topic_partition_offset = sub_topic_partition_offset;
                this.timestamp = timestamp;
            }
        
            public String getConsumer_group() {
                return consumer_group;
            }
        
            public void setConsumer_group(String consumer_group) {
                this.consumer_group = consumer_group;
            }
        
            public String getSub_topic() {
                return sub_topic;
            }
        
            public void setSub_topic(String sub_topic) {
                this.sub_topic = sub_topic;
            }
        
            public Integer getSub_topic_partition_id() {
                return sub_topic_partition_id;
            }
        
            public void setSub_topic_partition_id(Integer sub_topic_partition_id) {
                this.sub_topic_partition_id = sub_topic_partition_id;
            }
        
            public Long getSub_topic_partition_offset() {
                return sub_topic_partition_offset;
            }
        
            public void setSub_topic_partition_offset(Long sub_topic_partition_offset) {
                this.sub_topic_partition_offset = sub_topic_partition_offset;
            }
        
            public String getTimestamp() {
                return timestamp;
            }
        
            public void setTimestamp(String timestamp) {
                this.timestamp = timestamp;
            }
        
            @Override
            public String toString() {
                return "Offset{" +
                        "consumer_group='" + consumer_group + '\'' +
                        ", sub_topic='" + sub_topic + '\'' +
                        ", sub_topic_partition_id=" + sub_topic_partition_id +
                        ", sub_topic_partition_offset=" + sub_topic_partition_offset +
                        ", timestamp='" + timestamp + '\'' +
                        '}';
            }
        }
    
     *  配置文件
    
        # mysql连接相关信息
        db.driver=com.mysql.jdbc.Driver
        db.user=root
        db.password=123456
        db.url=jdbc:mysql://192.168.139.101:3306/mydb?useUnicode=true&characterEncoding=UTF8&useSSL=false
4.  数据库建表语句
    
        CREATE TABLE `offset` (
          `consumer_group` varchar(255) NOT NULL DEFAULT '',
          `sub_topic` varchar(255) NOT NULL DEFAULT '',
          `sub_topic_partition_id` int(11) NOT NULL DEFAULT '0',
          `sub_topic_partition_offset` bigint(20) NOT NULL,
          `timestamp` varchar(255) CHARACTER SET utf8 NOT NULL,
          PRIMARY KEY (`consumer_group`,`sub_topic`,`sub_topic_partition_id`)
        ) ENGINE=InnoDB DEFAULT CHARSET=latin1;
    
    mysql数据截图
    
    使用replace的话，必须有相应的主键作为限制，不然起不到我们想要的目的
    
    replace：根据三个主键，先去查询是否存在三个主键对应值得存在，不存在的话直接insert，存在的话就覆盖

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70]  
根据需求，如果需要加入consumer\_id的话，那就同样可以设置为4号主键，动态的数据不能设置为主键，动手尝试一下就知道其中的奥妙了

==============回收的分区=============
    ==============重新得到的分区==========
    partition = first-2
    partition = first-1
    partition = first-0

*  此时在不关闭已开启的程序的情况下，再启动一次程序

第一次运行的程序结果

==============回收的分区=============
    partition = first-2
    partition = first-1
    partition = first-0
    ==============重新得到的分区==========
    partition = first-2

第二次运行的程序结果

==============回收的分区=============
    ==============重新得到的分区==========
    partition = first-1
    partition = first-0

这是因为两次运行的程序的消费者组id都是test，为同一个消费者组，当第二次运行程序时，对原来的分区进行回收，进行了分区的rebalance重新分配(range分配)。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI2ODM4MzE1_size_16_color_FFFFFF_t_70]: /images/20221123/45ca2ea87ecd4b00a4bf8fb81464763f.png