个人中心
gpt4 book ai didi

63、Flink深入：Flink中通用RedisUtil工具类

转载 作者：大佬之路 更新时间：2024-01-07 17:06:53 25 4
gpt4 key购买 nike

1. 开发目的

在使用SpringBoot后端开发中，我们如果需要对Redis进行增删查改，可以很方便的使用RedisTemplate或者StringRedisTemplate等对象进行操作。但是在大数据中，如果想要对Redis进行操作，就没有那么方便，特别当flink新一代流式计算框架兴起后，没有直接读取和写入Redis的连接源，不管是开始的时候从Redis中获取数据，还是在中间需要读取维度数据，或者最后将数据写入到Redis，都不方便。此时一个较为方便的工具类就能很方便的使用，能达到节省开发时间、减小开发难度等目的。

2. 导入依赖


    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <scala.binary.version>2.11</scala.binary.version>
        <scala.version>2.11.8</scala.version>
        <flink.binary.version>1.10</flink.binary.version>
        <flink.version>1.10.0</flink.version>
        <log4j.version>1.2.17</log4j.version>
        <slf4j.version>1.7.21</slf4j.version>
        <mysql.version>8.0.21</mysql.version>
        <fastjson.version>2.0.20</fastjson.version>
        <avro.version>1.11.0</avro.version>
        <huaweicloud.dws.jdbc.version>8.1.0</huaweicloud.dws.jdbc.version>
        <commons.beanutils.version>1.9.4</commons.beanutils.version>
        <guava.version>29.0-jre</guava.version>
        <okhttp.version>3.6.0</okhttp.version>
        <springboot.version>2.3.3.RELEASE</springboot.version>
        <hikari.cp.version>2.6.1</hikari.cp.version>
        <avro.version>1.10.0</avro.version>
        <jedis.version>4.2.0</jedis.version>
        <commons.lang.version>3.10</commons.lang.version>
        <huaweicloud.obs.version>3.21.4</huaweicloud.obs.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>

        <!-- flink相关jar包 -->
        <!--flink流的核心包-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--flink中的Table相关包-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- flink连接kafka-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--flink的rocksdb包-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--MySQL驱动包 mysql8版本-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>${mysql.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- jdbc连接池包（使用JDBCTemplate） -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
            <version>${springboot.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
            <version>${hikari.cp.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--redis-->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>${jedis.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- 华为云GaussDB的连接JDBC的Jar包 -->
        <dependency>
            <groupId>com.huaweicloud.dws</groupId>
            <artifactId>huaweicloud-dws-jdbc</artifactId>
            <version>${huaweicloud.dws.jdbc.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- 日志打印的jar包 -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- json解析包，fastjson包 -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>${fastjson.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- avro压缩包 -->
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>${avro.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--commons-beanutils 是 Apache 开源组织提供的用于操作 JAVA BEAN 的工具包。使用 commons-beanutils，我们可以很方便的对 bean 对象的属性进行操作-->
        <dependency>
            <groupId>commons-beanutils</groupId>
            <artifactId>commons-beanutils</artifactId>
            <version>${commons.beanutils.version}</version>
            <scope>provided</scope>
        </dependency>

        <!--Guava 工程包含了若干被 Google 的 Java 项目广泛依赖的核心库,方便开发-->
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>${guava.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- 共有的lang包 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>${commons.lang.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- http包 -->
        <dependency>
            <groupId>com.squareup.okhttp3</groupId>
            <artifactId>okhttp</artifactId>
            <version>${okhttp.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- 华为云连接OBS包 -->
        <dependency>
            <groupId>com.huaweicloud</groupId>
            <artifactId>esdk-obs-java</artifactId>
            <version>${huaweicloud.obs.version}</version>
            <scope>provided</scope>
        </dependency>

    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.4</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                    <exclude>org.apache.hadoop:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the META-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                        <exclude>*.properties</exclude>
                                        <exclude>*.xml</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers combine.children="append">
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer">
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

3. 代码

注意：下述代码中使用了自定义的ModelUtil工具类，该工具类的具体介绍可以参考博主的另一篇博文：Flink（60）：Flink中通用ModelUtil工具类

3.1. 使用SpringBoot中的RedisTemplate实现

注意：下述方法为静态工具类，在使用时，直接使用RedisMlUtil调用方法即可；而且因为是静态工具类，所以这个类的库固定了，只能访问这个一个库的数据，如果需要其他的Redis集群可以再创建类似工具类（一般一个公司的Redis集群不会很多，所以一般用来读取数据的工具类，使用静态的即可）。并且其中的连接池的大小、超时时间等参数已进行固定（该参数可以根据公司集群来调节，博主使用的集群用如下配置即可）。另外还有，该类是静态，所以在每台机器上只会创建一个对象，这样连接池等大小就需要配置的大一点。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.jedis.JedisClientConfiguration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import redis.clients.jedis.JedisPoolConfig;

import java.time.Duration;

/**
 * @date: 2022/12/29
 *  @Author ddkk.com  弟弟快看，程序员编程资料站
 * @desc: ml-redis集群工具类
 */
public class RedisMlUtil {

    static Logger logger = LoggerFactory.getLogger(RedisMlUtil.class);

    /**
     * StringRedisTemplate对象
     */
    private static StringRedisTemplate stringRedisTemplate;

    /**
     * 获取StringRedisTemplate对象
     *
     * @return StringRedisTemplate对象
     */
    public static StringRedisTemplate getStringRedisTemplate() {

        if (stringRedisTemplate == null) {
            synchronized (RedisMlUtil.class) {
                if (stringRedisTemplate == null) {

                    // 创建 Jedis连接池 配置对象
                    JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
                    jedisPoolConfig.setMaxTotal(50);
                    jedisPoolConfig.setMaxIdle(50);
                    jedisPoolConfig.setMinIdle(5);

                    // 通过 Jedis连接池配置 创建 jedis客户端 配置对象
                    JedisClientConfiguration.JedisClientConfigurationBuilder jedisClientConfigurationBuilder = JedisClientConfiguration.builder();
                    jedisClientConfigurationBuilder.connectTimeout(Duration.ofMillis(3000));
                    jedisClientConfigurationBuilder.usePooling().poolConfig(jedisPoolConfig);
                    JedisClientConfiguration jedisClientConfiguration = jedisClientConfigurationBuilder.build();

                    // 创建 redis 配置对象
                    RedisStandaloneConfiguration redisConfiguration = new RedisStandaloneConfiguration();
                    redisConfiguration.setHostName(ModelUtil.getConfigValue("redis.ml.hostname"));
                    redisConfiguration.setPort(Integer.parseInt(ModelUtil.getConfigValue("redis.ml.port")));
                    redisConfiguration.setPassword(RedisPassword.of(ModelUtil.getConfigValue("redis.ml.password")));
                    redisConfiguration.setDatabase(Integer.parseInt(ModelUtil.getConfigValue("redis.ml.database")));

                    // 通过 jedis客户端配置对象 和 redis配置对象 创建Jedis连接工厂
                    JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(redisConfiguration, jedisClientConfiguration);

                    // 通过 Jedis连接工厂 创建 redisTemplate
                    stringRedisTemplate = new StringRedisTemplate(jedisConnectionFactory);

                    logger.info(
                            "##### 创建ml-redis集群客户端StringRedisTemplate对象成功，其中最大连接数为：{}，空闲连接数为：{}，连接地址为：{}",
                            50,
                            5,
                            ModelUtil.getConfigValue("redis.ml.hostname")
                    );

                }
            }
        }

        return stringRedisTemplate;

    }

    /**
     * 通过传入的key获取对应的value值
     *
     * @param key 键
     * @return 值
     */
    public static String getValue(String key) {
        return RedisMlUtil.getStringRedisTemplate().opsForValue().get(key);
    }

}

3.2. 使用Jedis实现

注意：下述方法为静态工具类，在使用时，直接使用RedisMlUtil调用方法即可；而且因为是静态工具类，所以这个类的库固定了，只能访问这个一个库的数据，如果需要其他的Redis集群可以再创建类似工具类（一般一个公司的Redis集群不会很多，所以一般用来读取数据的工具类，使用静态的即可）。并且其中的连接池的大小、超时时间等参数已进行固定（该参数可以根据公司集群来调节，博主使用的集群用如下配置即可）。另外还有，该类是静态，所以在每台机器上只会创建一个对象，这样连接池等大小就需要配置的大一点。


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
 * @date: 2022/12/29
 *  @Author ddkk.com  弟弟快看，程序员编程资料站
 * @desc: ml-redis集群工具类
 */
public class RedisMlUtil {

    static Logger logger = LoggerFactory.getLogger(RedisMlUtil.class);

    /**
     * JedisPool对象
     */
    private static JedisPool jedisPool;

    /**
     * 获取JedisPool对象
     *
     * @return JedisPool对象
     */
    public static JedisPool getJedisPool() {

        if (jedisPool == null) {
            synchronized (RedisMlUtil.class) {
                if (jedisPool == null) {
                    JedisPoolConfig poolConfig = new JedisPoolConfig();
                    poolConfig.setMaxTotal(50);
                    poolConfig.setMaxIdle(50);
                    poolConfig.setMinIdle(5);
                    jedisPool = new JedisPool(
                            poolConfig,
                            ModelUtil.getConfigValue("redis.ml.hostname"),
                            Integer.parseInt(ModelUtil.getConfigValue("redis.ml.port")),
                            3000,
                            ModelUtil.getConfigValue("redis.ml.password")
                    );
                    logger.info(
                            "根据传入的参数创建redis连接池jedisPool对象成功，使用的host为：{}，使用的port为：{}，最大连接数为{}，最大空闲连接数为{}，最小空闲连接数为：{}，连接超时时间为(毫秒)：{}",
                            ModelUtil.getConfigValue("redis.ml.hostname"),
                            Integer.parseInt(ModelUtil.getConfigValue("redis.ml.port")),
                            50,
                            50,
                            5,
                            3000
                    );
                }
            }
        }

        return jedisPool;

    }

    /**
     * 获取对应的 Jedis
     * 注意：该jedis对象是从连接池中返回，使用完之后需要关闭
     *
     * @param index redis对应的索引
     * @return Jedis
     */
    public static Jedis getJedis(int index) {
        Jedis jedis = getJedisPool().getResource();
        jedis.select(index);
        return jedis;
    }

    /**
     * 通过传入的key获取对应的value值
     *
     * @param key 键
     * @return 值
     */
    public static String getValue(int index, String key) {
        Jedis jedis = getJedis(index);
        String value = jedis.get(key);
        jedis.close();
        return value;
    }

}

3.3. 通用工具类

注意：该工具类没有使用静态，所以在每次使用的时候，需要在open方法中创建该工具类对象，然后在process方法中使用该工具类中的方法即可。此工具类即可以用于维度等数据读取，又能用于将结果数据写入，并且每次创建工具类的时候，可以指定不同的Redis集群和不同的参数。同样，因为每次使用时都创建该类的对象，这样Flink程序中每一个并发，所以在使用时需要注意连接池等参数不要配置的太大。


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

/**
 * @date: 2022/12/6
 *  @Author ddkk.com  弟弟快看，程序员编程资料站
 * @desc: RedisUtil
 */
public class RedisUtil {

    static Logger logger = LoggerFactory.getLogger(RedisUtil.class);

    /**
     * jedis连接池
     */
    private JedisPool jedisPool;

    /**
     * 通过传入的参数创建RedisUtil对象
     *
     * @param host     redis的host
     * @param port     redis的端口
     * @param password redis的password
     */
    public RedisUtil(String host, int port, String password) {
        initJedisPool(host, port, password, 2, 2, 1, 3000);
    }

    /**
     * 通过传入的参数创建RedisUtil对象
     *
     * @param host     redis的host
     * @param port     redis的端口
     * @param password redis的password
     * @param maxTotal 连接池中最大连接数
     * @param maxIdle  连接池中最大空闲连接数
     * @param minIdle  连接池中最小空闲连接数
     * @param timeout  连接Redis超时时间
     */
    public RedisUtil(String host, int port, String password, int maxTotal, int maxIdle, int minIdle, int timeout) {
        initJedisPool(host, port, password, maxTotal, maxIdle, minIdle, timeout);
    }

    /**
     * 初始化Jedis对象
     *
     * @param host     redis的host
     * @param port     redis的端口
     * @param password redis的password
     * @param maxTotal 连接池中最大连接数
     * @param maxIdle  连接池中最大空闲连接数
     * @param minIdle  连接池中最小空闲连接数
     * @param timeout  连接Redis超时时间
     */
    public void initJedisPool(String host, int port, String password, int maxTotal, int maxIdle, int minIdle, int timeout) {

        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(maxTotal);
        poolConfig.setMaxIdle(maxIdle);
        poolConfig.setMinIdle(minIdle);
        jedisPool = new JedisPool(
                poolConfig,
                host,
                port,
                timeout,
                password
        );

        logger.info(
                "根据传入的参数创建redis连接池jedisPool对象成功，使用的host为：{}，使用的port为：{}，最大连接数为{}，最大空闲连接数为{}，最小空闲连接数为：{}，连接超时时间为(毫秒)：{}",
                host,
                port,
                maxTotal,
                maxIdle,
                minIdle,
                timeout
        );

    }

    /**
     * 获取对应的 Jedis
     * 注意：该jedis对象是从连接池中返回，使用完之后需要关闭
     *
     * @param index redis对应的索引
     * @return Jedis
     */
    public Jedis getJedis(int index) {
        Jedis jedis = jedisPool.getResource();
        jedis.select(index);
        return jedis;
    }

    /**
     * 根据传入的数据库索引和key获取对应的值
     * 注意：如果该key不存在，就返回 'nil'，如果存储在key的值不是字符串，则返回错误
     *
     * @param index redis的索引
     * @param key   redis的key
     * @return value
     */
    public String getValue(int index, String key) {
        Jedis jedis = getJedis(index);
        String value = jedis.get(key);
        jedis.close();
        return value;
    }

}

4. 如何使用

在Flink作业中如何使用上述工具类，可以参考博主的另一篇MySQL工具类文章：Flink（62）：Flink中通用MySQLUtil工具类，使用基本相似，这里就不再过多描述了。

25 4 0
文章推荐： 09、Flink深入：Flink的各种创建执行环境的方法
文章推荐： 08、Flink深入：Flink的API说明和pom文件汇总
文章推荐： 62、Flink深入：Flink中通用MySQLUtil工具类
文章推荐： 60、Flink深入：Flink中通用ModelUtil工具类
大佬之路
个人简介

我是一名优秀的程序员,十分优秀！

滴滴打车优惠券免费领取
滴滴打车优惠券
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com