flink写入redis（续）

柔光的暖阳◎ 2023-06-09 10:29 11阅读 0赞

参照源码：[https://github.com/apache/bahir-flink/tree/master/flink-connector-redis][https_github.com_apache_bahir-flink_tree_master_flink-connector-redis]

先分析，后修改。

flink-connector-redis代码目录：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3d19jaGFuZ2Vy_size_16_color_FFFFFF_t_70][]

着重看RedisSink类：

**1. 包含如下几个私有对象： **

private String additionalKey;
    private Integer additionalTTL;
    private RedisMapper<IN> redisSinkMapper;
    private RedisCommand redisCommand;
    private FlinkJedisConfigBase flinkJedisConfigBase;
    private RedisCommandsContainer redisCommandsContainer;

additionalKey：hset中的hashField 或 zadd中的score

additionalTTL：ttl

redisSinkMapper：redis输出映射接口，提供抽象方法，由用户实现

public interface RedisMapper<T> extends Function, Serializable {
    
        /**
         * Returns descriptor which defines data type.
         *
         * @return data type descriptor
         */
        RedisCommandDescription getCommandDescription();
    
        /**
         * Extracts key from data.
         *
         * @param data source data
         * @return key
         */
        String getKeyFromData(T data);
    
        /**
         * Extracts value from data.
         *
         * @param data source data
         * @return value
         */
        String getValueFromData(T data);
    
        /**
         * Extracts the additional key from data as an {@link Optional<String>}.
         * The default implementation returns an empty Optional.
         *
         * @param data
         * @return Optional
         */
        default Optional<String> getAdditionalKey(T data) {
            return Optional.empty();
        }
    
        /**
         * Extracts the additional time to live (TTL) for data as an {@link Optional<Integer>}.
         * The default implementation returns an empty Optional.
         *
         * @param data
         * @return Optional
         */
        default Optional<Integer> getAdditionalTTL(T data) {
            return Optional.empty();
        }
    }

redisCommand：redis命令枚举类

public enum RedisCommand {
    
        /**
         * Insert the specified value at the head of the list stored at key.
         * If key does not exist, it is created as empty list before performing the push operations.
         */
        LPUSH(RedisDataType.LIST),
    
        /**
         * Insert the specified value at the tail of the list stored at key.
         * If key does not exist, it is created as empty list before performing the push operation.
         */
        RPUSH(RedisDataType.LIST),
    
        /**
         * Add the specified member to the set stored at key.
         * Specified member that is already a member of this set is ignored.
         */
        SADD(RedisDataType.SET),
    
        ...
    }

flinkJedisConfigBase：redis配置信息抽象类，提供三种实现类

*  FlinkJedisPoolConfig：**Single **
 *  FlinkJedisClusterConfig：**Cluster**
 *  FlinkJedisSentinelConfig：**Sentinel**

### **重点！** ###

**RedisCommandsContainer：redis命令容器接口，提供open()、close()及hset()等抽象方法**

/**
     * The container for all available Redis commands.
     */
    public interface RedisCommandsContainer extends Serializable {
    
        /**
         * Open the Jedis container.
         *
         * @throws Exception if the instance can not be opened properly
         */
        void open() throws Exception;
    
        void hset(String key, String hashField, String value, Integer ttl);
        void rpush(String listName, String value);
        void lpush(String listName, String value);
        void sadd(String setName, String value);
        void publish(String channelName, String message);
        void set(String key, String value);
        void setex(String key, String value, Integer ttl);
        void pfadd(String key, String element);
        void zadd(String key, String score, String element);
        void zrem(String key, String element);
    
        /**
         * Close the Jedis container.
         *
         * @throws IOException if the instance can not be closed properly
         */
        void close() throws IOException;
    }

提供两个实现类：

public class RedisContainer implements RedisCommandsContainer, Closeable {}

public class RedisClusterContainer implements RedisCommandsContainer, Closeable {}

和一个构造类：

public class RedisCommandsContainerBuilder {}

### **重中之重！** ###

RedisContainer 中包含 private transient （私有、**临时**）的JedisPool、JedisSentinel Pool

public class RedisContainer implements RedisCommandsContainer, Closeable {
    
        private transient JedisPool jedisPool;
        private transient JedisSentinelPool jedisSentinelPool;
    
        ...
    }

RedisClusterContainer  中包含 private transient （私有、**临时**）的JedisCluster

public class RedisClusterContainer implements RedisCommandsContainer, Closeable {
    
        private transient JedisCluster jedisCluster;
    
        ...
    }

关于 transient：

*  一旦变量被transient修饰，变量将不再是对象持久化的一部分，该变量内容在序列化后无法获得访问。
 *  transient关键字只能修饰变量，而不能修饰方法和类。注意，本地变量是不能被transient关键字修饰的。变量如果是用户自定义类变量，则该类需要实现Serializable接口。
 *  被transient关键字修饰的变量不再能被序列化，一个静态变量不管是否被transient修饰，均不能被序列化。

解决了Jedis序列化报错的问题：

org.apache.flink.api.common.InvalidProgramException: The implementation of the RichFilterFunction is not serializable. The object probably contains or references non serializable fields.
    
    ...
    
    Caused by: java.io.NotSerializableException: redis.clients.jedis.JedisSentinelPool

**2.构造方法：**

public RedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RedisMapper<IN> redisSinkMapper) {
            Objects.requireNonNull(flinkJedisConfigBase, "Redis connection pool config should not be null");
            Objects.requireNonNull(redisSinkMapper, "Redis Mapper can not be null");
            Objects.requireNonNull(redisSinkMapper.getCommandDescription(), "Redis Mapper data type description can not be null");
    
            this.flinkJedisConfigBase = flinkJedisConfigBase;
    
            this.redisSinkMapper = redisSinkMapper;
            RedisCommandDescription redisCommandDescription = redisSinkMapper.getCommandDescription();
    
            this.redisCommand = redisCommandDescription.getCommand();
            this.additionalTTL = redisCommandDescription.getAdditionalTTL();
            this.additionalKey = redisCommandDescription.getAdditionalKey();
        }

**3.invoker()方法：**

public void invoke(IN input, Context context) throws Exception {
            String key = redisSinkMapper.getKeyFromData(input);
            String value = redisSinkMapper.getValueFromData(input);
    
            Optional<String> optAdditionalKey = redisSinkMapper.getAdditionalKey(input);
            Optional<Integer> optAdditionalTTL = redisSinkMapper.getAdditionalTTL(input);
    
            switch (redisCommand) {
                case RPUSH:
                    this.redisCommandsContainer.rpush(key, value);
                    break;
                case LPUSH:
                    this.redisCommandsContainer.lpush(key, value);
                    break;
    
                ...
    
                default:
                    throw new IllegalArgumentException("Cannot process such data type: " + redisCommand);
            }
        }

**4.open()和close()方法：**

@Override
        public void open(Configuration parameters) throws Exception {
            try {
                this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
                this.redisCommandsContainer.open();
            } catch (Exception e) {
                LOG.error("Redis has not been properly initialized: ", e);
                throw e;
            }
        }
    
        @Override
        public void close() throws IOException {
            if (redisCommandsContainer != null) {
                redisCommandsContainer.close();
            }
        }

**以上，**

*  **修改RedisMapper提供对应的Redis映射**
 *  **修改RedisCommandsContainer提供需要的Redis命令**
 *  **修改FlinkJedisConfigBase提供需要的Redis配置**

**根据自身需求，写一版自己的RedisSink，问题不大。**

[https_github.com_apache_bahir-flink_tree_master_flink-connector-redis]: https://github.com/apache/bahir-flink/tree/master/flink-connector-redis
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d3d19jaGFuZ2Vy_size_16_color_FFFFFF_t_70]: /images/20230531/ae7c15c9959f47e89973fd68eb27e00e.png