Flink-03 Flink Java 3分钟上手 Stream 给 Flink-02 DataStreamSource Socket写一个测试的工具!

代码仓库

会同步代码到 GitHub
https://github.com/turbo-duck/flink-demo

在这里插入图片描述

当前章节

继续上一节的内容:https://blog.csdn.net/w776341482/article/details/139875037

上一节中,我们需要使用 nc 或者 telnet 等工具来模拟 Socket 流。这节我们写一个 ServerSocket 来模拟这些 操作,让流自动的写入不用我们手动去操作了。

POM.xml

与上一节一致,不需要修改

编写代码

还是和上一节一样的 Socket 流,这里略去其他的代码

DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");

FlinkServer

继承 Thread 启动一个线程来进行Flink的服务

package icu.wzk.demo03;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;public class FlinkServer extends Thread {@Overridepublic void run() {String ip = "0.0.0.0";int port = 9999;StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {@Overridepublic void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {String[] splits = s.split("\\s");for (String word : splits) {collector.collect(Tuple2.of(word, 1L));}}});SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator.keyBy(new KeySelector<Tuple2<String, Long>, Object>() {@Overridepublic Object getKey(Tuple2<String, Long> stringLongTuple2) throws Exception {return stringLongTuple2.f0;}}).window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))).sum(1);System.out.println("wait word print()");word.print();try {streamExecutionEnvironment.execute("stream!");} catch (Exception e) {throw new RuntimeException(e);}}
}

NumRandom

使用 ServerSocket 实现一个持续的流输出

package icu.wzk.demo03;import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;public class RandomNumClient extends Thread {@Overridepublic void run() {String ip = "0.0.0.0";int port = 9999;try {ServerSocket serverSocket = new ServerSocket();InetSocketAddress address = new InetSocketAddress(ip, port);serverSocket.bind(address);Socket socket = serverSocket.accept();OutputStream output = socket.getOutputStream();PrintWriter writer = new PrintWriter(output, true);Random random = new Random();for (int i = 0; i < 500; i ++) {int randomNumber = random.nextInt(10) + 1;writer.println(randomNumber);System.out.println("ServerSocket Send To Flink: " + randomNumber);Thread.sleep(200);}} catch (Exception e) {throw new RuntimeException(e);}}}

StartApp

将上述的两个类组装起来

请添加图片描述

package icu.wzk.demo03;public class StartApp {public static void main(String[] args) throws Exception {RandomNumClient randomNumClient = new RandomNumClient();FlinkServer flinkServer = new FlinkServer();flinkServer.start();randomNumClient.start();}}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/diannao/32644.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

36.Http协议的设计与解析

Http协议比Redis协议复杂的多,如果程序员自己去实现,工作量大。 Netty已经把Http协议的编解码器实现好了,只需要简单的配置就可以使用。 做一个http的服务端需要HttpServerCodec。 看它继承的父类: 结合了两个类: HttpRequestDecoder(入站处理器extends Channelnbound…

【网络安全常用术语解读 :什么是0day、1day、nday漏洞】

脆弱性攻击的时间窗被称作脆弱性窗口。通常情况下&#xff0c;一个安全漏洞的时间越久&#xff0c;攻击者就会有更多的机会去攻击它。 2. 0day 漏洞 0天漏洞&#xff0c;也被称作"零日漏洞"&#xff0c;是指尚未由供应商公布的缺陷&#xff0c;表示攻击者已知晓该缺…

力扣练习题 (2024.6.23)

//我们定义&#xff0c;在以下情况时&#xff0c;单词的大写用法是正确的&#xff1a; //全部字母都是大写&#xff0c;比如 "USA" 。 //单词中所有字母都不是大写&#xff0c;比如 "leetcode" 。 //如果单词不只含有一个字母&#xff0c;只有首字母大写&a…

Java线程基础知识汇总

进程与线程 什么是进程&#xff1f; 进程是指在计算机中运行的程序的实例。它是操作系统进行资源分配和调度的基本单位。一个进程可以包含多个线程&#xff0c;每个线程都共享该进程的资源&#xff0c;如内存、文件和打开的网络连接等。每个进程都有自己的地址空间&#xff0…

FlowUs2024重磅革新预告:RAG技术赋能『问问AI』,笔记变现新纪元等你开启!

&#x1f389; 在FlowUs的广阔天地间&#xff0c;知识的边界被无限拓展&#xff0c;引领着一场场创新与收获的庆典&#xff01;&#x1f680; 随着一年间不断的精进与革新&#xff0c;FlowUs与众多用户并肩前行&#xff0c;在追求极致体验的道路上迈出坚实步伐。步入2024年&am…

WordPress项目教程:自动采集并发布,让你轻松实现网站内容更新

随着互联网的发展&#xff0c;越来越多的人开始关注自己的个人网站&#xff0c;通过网站展示自己的才华、分享知识、推广产品等。然而&#xff0c;个人网站的运营并非易事&#xff0c;尤其是内容更新方面。为了解决这个问题&#xff0c;今天我们将为大家推荐一款WordPress插件主…

minio直接通过地址访问无需服务器转发

背景 做网站有些图片、视频、js等资源&#xff0c;没有什么变化&#xff0c;想在网站上直接使用&#xff0c;前端拿到地址可直接访问获得&#xff0c;而不是通过后台转一道再获得&#xff0c;折腾了半天访问不到&#xff0c;从网上找资料挨个试&#xff0c;也没完全解决&#…

云计算期末综合测试题

云计算综合测试题 单选题填空题判断题简答题 单选题 这里选择题&#xff0c;直接以填空题展示&#xff0c;并给出解析 Bigtable是&#xff08;Google&#xff09;开发的分布式存储系统 解析&#xff1a;分布式结构化数据表Bigtable是Google基于GFS和Chubby开发的分布式存储系统…

君子小人的格局、境界

子曰&#xff1a;君子怀德&#xff0c;小人怀土&#xff1b;君子怀刑&#xff0c;小人怀惠。 直译&#xff1a;君子怀念道德&#xff0c;小人怀念乡土&#xff1b;君子关心法度&#xff0c;小人关心恩惠。 这里的君子与小人只是体现格局、境界的不同&#xff1b; 君子怀的是德…

DVWA 靶场 SQL Injection (Blind) 通关解析

前言 DVWA代表Damn Vulnerable Web Application&#xff0c;是一个用于学习和练习Web应用程序漏洞的开源漏洞应用程序。它被设计成一个易于安装和配置的漏洞应用程序&#xff0c;旨在帮助安全专业人员和爱好者了解和熟悉不同类型的Web应用程序漏洞。 DVWA提供了一系列的漏洞场…

sklearn之各类朴素贝叶斯原理

sklearn之贝叶斯原理 前言1 高斯朴素贝叶斯1.1 对连续变量的处理1.2 高斯朴素贝叶斯算法原理 2 多项式朴素贝叶斯2.1 二项分布和多项分布2.2 详细原理2.3 如何判断是否符合多项式贝叶斯 3 伯努利朴素贝叶斯4 类别贝叶斯4 补充朴素贝叶斯4.1 核心原理4.2 算法流程 前言 如果想看…

docker nacos2.3.2安装填坑

#nacos2.3.2安装# git clone https://github.com/nacos-group/nacos-docker.git cd nacos-docker #安装mysql8单机版 docker-compose -f example/standalone-mysql-8.yaml up #https://hub.docker.com 抽风# 因为网络原因&#xff0c;nacos/nacos-server 默认安装的最后版本…

Python封装cryptography对称加密方法

安装依赖 pip install cryptography实现方法 from cryptography.fernet import Fernetclass SymmetricEncryption(object):对称加密算法def __init__(self, secret_key=None

学习笔记——交通安全分析08

目录 前言 当天学习笔记整理 4信控交叉口交通安全分析 结束语 前言 #随着上一轮SPSS学习完成之后&#xff0c;本人又开始了新教材《交通安全分析》的学习 #整理过程不易&#xff0c;喜欢UP就点个免费的关注趴 #本期内容接上一期07笔记 当天学习笔记整理 4信控交叉口交…

天马学航——智慧教务系统(移动端)开发日志四

天马学航——智慧教务系统(移动端)开发日志四 日志摘要&#xff1a;优化了教师端界面的UI&#xff0c;更新了教师端添加课程&#xff0c;提交成绩等功能&#xff0c;修复了一些已知的BUG 1、教师添加课程设计 教师在此界面添加课程&#xff0c;并将数据提交后端进行审核 界…

代码随想录算法训练营第四十七天|LeetCode123 买卖股票的最佳时机Ⅲ

题1&#xff1a; 指路&#xff1a;123. 买卖股票的最佳时机 III - 力扣&#xff08;LeetCode&#xff09; 思路与代码&#xff1a; 买卖股票专题中三者不同的是Ⅰ为只买卖一次&#xff0c;Ⅱ可多次买卖&#xff0c;Ⅲ最多可买卖两次。那么我们将买买卖行为分为五个状态部分(…

LeetCode 每日一题 2024/6/17-2024/6/23

记录了初步解题思路 以及本地实现代码&#xff1b;并不一定为最优 也希望大家能一起探讨 一起进步 目录 6/17 522. 最长特殊序列 II6/18 2288. 价格减免6/19 2713. 矩阵中严格递增的单元格数6/20 2748. 美丽下标对的数目6/21 LCP 61. 气温变化趋势6/22 2663. 字典序最小的美丽字…

WAAP的特性、功能以及优势

随着互联网技术的快速发展&#xff0c;Web应用程序和API已经成为企业日常运营中不可或缺的部分。然而&#xff0c;与此同时&#xff0c;网络攻击手段也愈发复杂和隐蔽&#xff0c;给企业的数据安全带来了严峻的挑战。为了应对这一挑战&#xff0c;WAAP&#xff08;WebApplicati…

fastapi+vue3+primeflex前后端分离开发项目第一个程序

安装axios axios是用来请求后端接口的。 https://www.axios-http.cn/docs/intro pnpm 是一个前端的包管理工具&#xff0c;当我们需要给前端项目添加新的依赖的时候&#xff0c;就可以使用pnpm install 命令进行安装。 pnpm install axios安装 primeflex primeflex是一个cs…

知乎客户端跨平台-Hybrid-调试实战

在开发上述功能的过程中&#xff0c;发现了 flipper 这个工具 flipper 提供了一个桌面客户端&#xff0c;然后这个桌面客户端提供了一个和手机客户端通信的机制&#xff0c;免去了 socket 服务的开销&#xff0c;依靠这个通信机制&#xff0c;我们可以把上述的功能复制过来 基…