像Flink一样使用Redis

冷不防 2023-10-13 12:03 14阅读 0赞

> 打开思路，像Flink一样使用Redis。

微信搜索关注《Java学研大本营》，加入读者群，分享更多精彩

Apache Flink和 Redis 是两个强大的工具，可以一起使用来构建可以处理大量数据的实时数据处理管道。Flink 为处理数据流提供了一个高度可扩展和容错的平台，而 Redis 提供了一个高性能的内存数据库，可用于存储和查询数据。在本文中，将探讨如何使用 Flink 来使用异步函数调用 Redis，并展示如何使用它以非阻塞方式将数据推送到 Redis。

### Redis的故事 ###

![13bec8ec2afbe9dfea980011158e653e.png][]

“Redis：不仅仅是一个缓存

Redis 是一种功能强大的 NoSQL 内存数据结构存储，已成为开发人员的首选工具。虽然它通常被认为只是一个缓存，但 Redis 远不止于此。它可以作为数据库、消息代理和缓存三者合一。

Redis 的优势之一是它的多功能性。它支持各种数据类型，包括字符串、列表、集合、有序集合、哈希、流、HyperLogLogs 和位图。Redis 还提供地理空间索引和半径查询，使其成为基于位置的应用程序的宝贵工具。

Redis 的功能超出了它的数据模型。它具有内置的复制、Lua 脚本和事务，并且可以使用 Redis Cluster 自动分区数据。此外，Redis 通过 Redis Sentinel 提供高可用性。

注意：在本文中，将更多地关注Redis集群模式

![d3fbfc4c36b0c489695fba7d40a3b77a.png][]

Redis 集群使用带哈希槽的算法分片来确定哪个分片拥有给定的键并简化添加新实例的过程。同时，它使用 Gossiping 来确定集群的健康状况，如果主节点没有响应，可以提升辅助节点以保持集群健康。必须有奇数个主节点和两个副本才能进行稳健设置，以避免脑裂现象（集群无法决定提升谁并最终做出分裂决定）

为了与 Redis 集群对话，将使用lettuce和 Redis Async Java 客户端。

### Flink 的故事 ###

![e53151f302b96e9d07ef5903e16883f9.png][]

Apache Flink 是一个开源、统一的流处理和批处理框架，旨在处理实时、高吞吐量和容错数据处理。它建立在 Apache Gelly 框架之上，旨在支持有界和无界流上的复杂事件处理和有状态计算，它的快速之处在于其利用内存中性能和异步检查本地状态。

### 故事的主人公 ###

![99b2cfb8db696b4745b58cdff8ca49e8.png][]

与数据库的异步交互是流处理应用程序的游戏规则改变者。通过这种方法，单个函数实例可以同时处理多个请求，从而允许并发响应并显着提高吞吐量。通过将等待时间与其他请求和响应重叠，处理管道变得更加高效。

> 我们将以电商数据为例，计算24小时滑动窗口中每个品类的销售额，滑动时间为30秒，下沉到Redis，以便更快地查找下游服务。

充足的数据集

Category, TimeStamp
    Electronics,1679832334
    Furniture,1679832336
    Fashion,1679832378
    Food,16798323536

Flink Kafka 消费者类

package Aysnc_kafka_redis;
    
    import AsyncIO.RedisSink;
    import akka.japi.tuple.Tuple3;
    import deserializer.Ecommdeserialize;
    import model.Ecomm;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    import org.apache.flink.connector.kafka.source.KafkaSource;
    import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
    import org.apache.flink.streaming.api.datastream.AsyncDataStream;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    import java.util.concurrent.TimeUnit;
    
    public class FlinkAsyncRedis {
    
        public static void main(String[] args) throws Exception {
    
    
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            Ecommdeserialize jsonde = new Ecommdeserialize(); 
    
            KafkaSource<Ecomm> source = KafkaSource.<Ecomm>builder()
                    .setTopics("{dummytopic}")
                    .setBootstrapServers("{dummybootstrap}")
                    .setGroupId("test_flink")
                    .setStartingOffsets(OffsetsInitializer.earliest())
                    .setValueOnlyDeserializer(jsonde)
                    .build();
    
    
            DataStream<Ecomm> orderData = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
    
    
            orderData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Ecomm>(Time.seconds(10)) {
                @Override
                public long extractTimestamp(Ecomm element) {
                    return element.getEventTimestamp(); // extract watermark column from stream
                }
            });
    
            SingleOutputStreamOperator<Tuple3<String, Long, Long>> aggregatedData = orderData.keyBy(Ecomm::getCategory)
                    .window(SlidingEventTimeWindows.of(Time.hours(24),Time.seconds(30)))
                    .apply((WindowFunction<Ecomm, Tuple3<String, Long, Long>, String, TimeWindow>) (key, window, input, out) -> {
                        long count = 0;
                        for (Ecomm event : input) {
                            count++; // increment the count for each event in the window
                        }
                        out.collect(new Tuple3<>(key, window.getEnd(), count)); // output the category, window end time, and count
                    });
    
    
            // calling async I/0 operator to sink data to redis in UnOrdered way
            SingleOutputStreamOperator<String> sinkResults = AsyncDataStream.unorderedWait(aggregatedData,new RedisSink(
                    "{redisClusterUrl}"),
                    1000, // the timeout defines how long an asynchronous operation take before it is finally considered failed
                    TimeUnit.MILLISECONDS,
                     100); //capacity This parameter defines how many asynchronous requests may be in progress at the same time.
    
            sinkResults.print(); // print out the redis set response stored in the future for every key
    
            env.execute("RedisAsyncSink"); // you will be able to see your job running on cluster by this name
    
    
        }
    
    }

Redis 设置键异步 I/0 运算符

package AsyncIO;
    
    import akka.japi.tuple.Tuple3;
    import io.lettuce.core.RedisFuture;
    import io.lettuce.core.cluster.RedisClusterClient;
    import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
    import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
    import lombok.AllArgsConstructor;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.functions.async.ResultFuture;
    import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
    import scala.collection.immutable.List;
    
    import java.util.ArrayList;
    import java.util.Collections;
    
    @AllArgsConstructor
    public class RedisSink extends RichAsyncFunction<Tuple3<String, Long, Long>, String> {
    
        String redisUrl;
    
        public RedisSink(String redisUrl){
            this.redisUrl=redisUrl;
        }
    
        private transient RedisClusterClient client = null;
        private transient StatefulRedisClusterConnection<String, String> clusterConnection = null;
        private transient RedisAdvancedClusterAsyncCommands<String, String> asyncCall = null;
    
    
        // method executes any operator-specific initialization 
        @Override
        public void open(Configuration parameters) {
            if (client == null ) {
                client = RedisClusterClient.create(redisUrl);
            }
            if (clusterConnection == null) {
                clusterConnection = client.connect();
            }
            if (asyncCall == null) {
                asyncCall  = clusterConnection.async();
            }
        }
    
        // core logic to set key in redis using async connection and return result of the call via ResultFuture
        @Override
        public void asyncInvoke(Tuple3<String, Long, Long> stream, ResultFuture<String> resultFuture) {
    
            String productKey = stream.t1();
            System.out.println("RedisKey:" + productKey); //for logging
            String count = stream.t3().toString();
            System.out.println("Redisvalue:" +  count); //for logging
            RedisFuture<String> setResult = asyncCall.set(productKey,count);
    
            setResult.whenComplete((result, throwable) -> {if(throwable!=null){
                System.out.println("Callback from redis failed:" + throwable);
                resultFuture.complete(new ArrayList<>());
            }
            else{
                resultFuture.complete(new ArrayList(Collections.singleton(result)));
            }});
        }
        
         // method closes what was opened during initialization to free any resources 
        //  held by the operator (e.g. open network connections, io streams)
        @Override
        public void close() throws Exception {
            client.close();
        }
    
    }

### 用例： ###

*  数据科学模型可以使用流式传输到 Redis 的数据来查找和生成更多在销售季节经常销售的类别的产品。
 *  它可用于在网页上展示图表和数字作为销售统计数据，以在用户中产生积极购买的动力。

### 要点： ###

*  Flink 为处理数据流提供了一个高度可扩展和容错的平台，而 Redis 提供了一个高性能的内存数据库，可用于存储和查询数据。
 *  异步编程可用于通过允许对外部系统（如 Redis）进行非阻塞调用来提高数据处理管道的性能。
 *  两者的结合可能有助于带来实时数据决策文化。

### 推荐书单 ###

#### 《项目驱动零起点学Java》 ####

《项目驱动零起点学Java》共分 13 章，围绕 6 个项目和 258 个代码示例，分别介绍了走进Java 的世界、变量与数据类型、运算符、流程控制、方法、数组、面向对象、异常、常用类、集合、I/O流、多线程、网络编程相关内容。《项目驱动零起点学Java》总结了马士兵老师从事Java培训十余年来经受了市场检验的教研成果，通过6 个项目以及每章的示例和习题，可以帮助读者快速掌握Java 编程的语法以及算法实现。扫描每章提供的二维码可观看相应章节内容的视频讲解。

《项目驱动零起点学Java》贯穿6个完整项目，经过作者多年教学经验提炼而得，项目从小到大、从短到长，可以让读者在练习项目的过程中，快速掌握一系列知识点。

马士兵，马士兵教育创始人，毕业于清华大学，著名IT讲师，所讲课程广受欢迎，学生遍布全球大厂，擅长用简单的语言讲授复杂的问题，擅长项目驱动知识的综合学习。马士兵教育获得在线教育“名课堂”奖、“最受欢迎机构”奖。

赵珊珊，从事多年一线开发，曾为国税、地税税务系统工作。拥有7年一线教学经验，多年线上、线下教育的积累沉淀，培养学员数万名，讲解细致，脉络清晰。

[《项目驱动零起点学Java》(马士兵，赵珊珊)【摘要 书评 试读】- 京东图书京东JD.COM图书频道为您提供《项目驱动零起点学Java》在线选购，本书作者：，出版社：清华大学出版社。买图书，到京东。网购图书，享受最低优惠折扣!![icon-default.png?t=N5K3][icon-default.png_t_N5K3]https://item.jd.com/13607758.html][Java_ _ _- _JD.COM_Java_icon-default.png_t_N5K3_icon-default.png_t_N5K3_https_item.jd.com_13607758.html]

![53f24038334137fcaa04c320693e7fe7.png][]

> > > > ### 精彩回顾 ###
> > 
> > [数据架构演进史（上）][Link 1]
> > 
> > [数据架构演进史（下）][Link 2]
> > 
> > [如何选择适合的后端框架][Link 3]

> > 微信搜索关注《Java学研大本营》
> > 
> > 访问【IT今日热榜】，发现每日技术热点

[13bec8ec2afbe9dfea980011158e653e.png]: https://img-blog.csdnimg.cn/img_convert/13bec8ec2afbe9dfea980011158e653e.png
[d3fbfc4c36b0c489695fba7d40a3b77a.png]: https://img-blog.csdnimg.cn/img_convert/d3fbfc4c36b0c489695fba7d40a3b77a.png
[e53151f302b96e9d07ef5903e16883f9.png]: https://img-blog.csdnimg.cn/img_convert/e53151f302b96e9d07ef5903e16883f9.png
[99b2cfb8db696b4745b58cdff8ca49e8.png]: https://img-blog.csdnimg.cn/img_convert/99b2cfb8db696b4745b58cdff8ca49e8.png
[icon-default.png_t_N5K3]: https://csdnimg.cn/release/blog_editor_html/release2.3.2/ckeditor/plugins/CsdnLink/icons/icon-default.png?t=N5K3
[Java_ _ _- _JD.COM_Java_icon-default.png_t_N5K3_icon-default.png_t_N5K3_https_item.jd.com_13607758.html]: https://item.jd.com/13607758.html
[53f24038334137fcaa04c320693e7fe7.png]: https://img-blog.csdnimg.cn/img_convert/53f24038334137fcaa04c320693e7fe7.png
[Link 1]: https://mp.weixin.qq.com/s?__biz=MzkzMDM3MjI3Nw==&mid=2247486741&idx=1&sn=517ba7d754622f7eb0faea16a6a6e354&chksm=c27a02cff50d8bd900503680cedcde21ae7525987f6d7dccbcec8da14282446a0964a7fd1d97&scene=21#wechat_redirect
[Link 2]: https://mp.weixin.qq.com/s?__biz=MzkzMDM3MjI3Nw==&mid=2247486755&idx=1&sn=d6d655a966870e3ced25f1da1d309fc1&chksm=c27a02f9f50d8bef677c96f1b609717e572913833dc3039facbb4f27edadcf1421dac5857eb5&scene=21#wechat_redirect
[Link 3]: https://mp.weixin.qq.com/s?__biz=MzkzMDM3MjI3Nw==&mid=2247485911&idx=1&sn=33ccbc9f8c9f98de24e05cdae26767f7&chksm=c27a060df50d8f1bb9d16fc293b5bd94faabacf4763d20f60a4962e709f1bd60efae4dd5a4ec&scene=21#wechat_redirect