记一次Flink通过Kafka写入MySQL的过程

一、前言
总体思路:source -->transform -->sink ,即从source获取相应的数据来源,然后进行数据转换,将数据从比较乱的格式,转换成我们需要的格式,转换处理后,然后进行sink功能,也就是将数据写入的相应的数据库DB中或者写入Hive的HDFS文件存储。
思路:
在这里插入图片描述

pom部分放到最后面。

二、方案及代码实现

2.1 Source部分
Source部分构建一个web对象用于保存数据等操作,代码如下:

package com.lzl.flink;import java.util.Date;/*** @author lzl* @create 2024-01-18 12:19* @name pojo*/
public class Web {private String uuid;private String ip;private String area;private String web;private String operate;private Date createDate;public String getArea() {return area;}public String getIp() {return ip;}public String getOperate() {return operate;}public String getUuid() {return uuid;}public String getWeb() {return web;}public Date getCreateDate() {return createDate;}public void setArea(String area) {this.area = area;}public void setIp(String ip) {this.ip = ip;}public void setOperate(String operate) {this.operate = operate;}public void setUuid(String uuid) {this.uuid = uuid;}public void setWeb(String web) {this.web = web;}public void setCreateDate(Date createDate) {this.createDate = createDate;}
}

将生成的数据转化为JSON格式,测试如下:

public static void webDataProducer() throws Exception{//构建web对象,在ip为10.117后面加两个随机数int randomInt1 = RandomUtils.nextInt(1,255);int randomInt2 = RandomUtils.nextInt(1,999);int randomInt3 = RandomUtils.nextInt(1,99999);List<String> areas = Arrays.asList("深圳", "广州", "上海", "北京", "武汉", "合肥", "杭州", "南京");List<String> webs = Arrays.asList("www.taobao.com","www.baidu.com","www.jd.com","www.weibo.com","www.qq.com","www.weixin.com","www.360.com","www.lzl.com","www.xiaomi.com");List<String> operates = Arrays.asList("register","view","login","buy","click","comment","jump","care","collect");Web web = new Web();  //实例化一个web对象,并向对象中放入数据web.setUuid("uid_" + randomInt3);web.setIp("10.110." + randomInt1 +"." + randomInt2);web.setArea(getRandomElement(areas));web.setWeb(getRandomElement(webs));web.setOperate(getRandomElement(operates));web.setCreateDate(new Date());// 转换成JSON格式String webJson = JSON.toJSONString(web);System.out.println(webJson); //打印出来看看效果}//构建一个从列表里面任意筛选一个元素的函数方法public static <T> T getRandomElement(List<T> list) {Collections.shuffle(list);return list.get(0);}public static void main(String[] args) {while (true) {try {// 每三秒写一条数据TimeUnit.SECONDS.sleep(3);webDataProducer();} catch (Exception e) {e.printStackTrace();}}}

执行测试结果如下:
在这里插入图片描述
至此Source部分结束~~~~!!!!!!

2.2 Transform_1部分

2.2.1 写入kafka方法函数:

package com.lzl.flink;import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;import java.util.*;
import java.util.concurrent.TimeUnit;/*** @author lzl* @create 2024-01-18 12:18* @name KafkaWriter*/
public class KafkaWriter {//kafka集群列表public static final String BROKER_LIST = "cdh39:9092,cdh40:9092,cdh41:9092";//kafka的topicpublic static final String TOPIC_WEB = "web";//kafka序列化的方式,采用字符串的形式public static final String KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";//value的序列化方式public static final String VALUE_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";public static void writeToKafka() throws Exception {Properties props = new Properties(); //实例化一个Propertiesprops.put("bootstrap.servers", BROKER_LIST);props.put("key.serializer", KEY_SERIALIZER);props.put("value.serializer", VALUE_SERIALIZER);// 构建Kafka生产者KafkaProducer<String, String> producer = new KafkaProducer<>(props);// 将web生成的数据发送给kafka的记录String webDataJson = webDataProducer();ProducerRecord<String,String> record = new ProducerRecord<String,String>(TOPIC_WEB,null,null,webDataJson);// 发送到缓存producer.send(record);System.out.println("向kafka发送数据:" + webDataJson);producer.flush();}public static void main(String[] args) {while (true) {try {// 每三秒写一条数据TimeUnit.SECONDS.sleep(3);writeToKafka();} catch (Exception e) {e.printStackTrace();}}}

2.2.2 建立 web的topic:
在这里插入图片描述
启动程序测试:
在这里插入图片描述
2.2.3 消费kafka看看是否有数据?

[root@cdh39 kafka]# bin/kafka-console-consumer.sh --bootstrap-server cdh39:9092 --from-beginning --topic web
{"area":"合肥","createDate":1705571020461,"ip":"10.110.104.676","operate":"comment","uuid":"uid_29661","web":"www.qq.com"}
{"area":"北京","createDate":1705571024048,"ip":"10.110.49.479","operate":"jump","uuid":"uid_77119","web":"www.weibo.com"}
{"area":"合肥","createDate":1705571027106,"ip":"10.110.232.960","operate":"click","uuid":"uid_99704","web":"www.taobao.com"}
{"area":"上海","createDate":1705571030140,"ip":"10.110.12.252","operate":"buy","uuid":"uid_99850","web":"www.jd.com"}
{"area":"合肥","createDate":1705571033228,"ip":"10.110.75.328","operate":"care","uuid":"uid_33135","web":"www.qq.com"}
{"area":"上海","createDate":1705571036267,"ip":"10.110.4.862","operate":"collect","uuid":"uid_37279","web":"www.taobao.com"}
{"area":"北京","createDate":1705571039361,"ip":"10.110.139.814","operate":"register","uuid":"uid_33016","web":"www.baidu.com"}
{"area":"武汉","createDate":1705571042422,"ip":"10.110.159.143","operate":"collect","uuid":"uid_26315","web":"www.lzl.com"}
{"area":"南京","createDate":1705571045495,"ip":"10.110.81.685","operate":"login","uuid":"uid_38712","web":"www.baidu.com"}
{"area":"南京","createDate":1705571048545,"ip":"10.110.228.267","operate":"comment","uuid":"uid_23297","web":"www.weibo.com"}
{"area":"武汉","createDate":1705571051623,"ip":"10.110.102.247","operate":"collect","uuid":"uid_77340","web":"www.lzl.com"}
{"area":"武汉","createDate":1705571054687,"ip":"10.110.184.832","operate":"comment","uuid":"uid_35230","web":"www.360.com"}
{"area":"武汉","createDate":1705571057760,"ip":"10.110.90.361","operate":"buy","uuid":"uid_52082","web":"www.lzl.com"}
{"area":"北京","createDate":1705571060825,"ip":"10.110.37.707","operate":"buy","uuid":"uid_45343","web":"www.weixin.com"}
{"area":"上海","createDate":1705571063909,"ip":"10.110.178.901","operate":"care","uuid":"uid_51015","web":"www.baidu.com"}
{"area":"杭州","createDate":1705571066945,"ip":"10.110.153.758","operate":"collect","uuid":"uid_46772","web":"www.xiaomi.com"}
{"area":"合肥","createDate":1705571069980,"ip":"10.110.177.755","operate":"comment","uuid":"uid_78442","web":"www.taobao.com"}
{"area":"广州","createDate":1705571073020,"ip":"10.110.151.427","operate":"register","uuid":"uid_92174","web":"www.weixin.com"}
{"area":"上海","createDate":1705571076072,"ip":"10.110.217.622","operate":"jump","uuid":"uid_86059","web":"www.xiaomi.com"}

至此,Transform_1部分结束~~~!!!!

2.3 Sink部分
创建一个MySQLSink,继承RichSinkFunction类。重载里边的open、invoke 、close方法,在执行数据sink之前先执行open方法,然后开始调用invoke方法,调用完之后最后执行close方法关闭资源。即在open里面创建数据库连接,然后调用invoke执行具体的数据库写入程序,完毕之后调用close关闭和释放资源。这里要继承flink的RichSinkFunction接口。代码如下:

package com.lzl.flink;import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.logging.Logger;/*** @author lzl* @create 2024-01-22 15:30* @name MySqlToPojoSink*/
public class MySqlToPojoSink extends RichSinkFunction<Web> {private static final Logger log = Logger.getLogger(MySqlToPojoSink.class.getName());private static final long serialVersionUID = 1L;private Connection connection = null;private PreparedStatement ps = null;private String tableName = "web";@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);log.info("获取数据库连接");// 通过Druid获取数据库连接,准备写入数据库connection = DbUtils.getConnection();// 插入数据库的语句   因为我们封装的pojo的类型为PojoType<com.lzl.flink.Web, fields = [area: String, createDate: Date, ip: String, operate: String, uuid: String, web: String]>String insertQuery = "INSERT INTO " +   tableName + "(time,ip,uid,area,web,operate) VALUES (?,?,?,?,?,?)" ;// 执行插入语句ps = connection.prepareStatement(insertQuery);}// 重新关闭方法。   关闭并释放资源@Overridepublic void close() throws Exception {super.close();if(connection != null) {connection.close();}if (ps != null ) {ps.close();}}// 重写invoke方法@Overridepublic void invoke(Web value,Context context) throws Exception {//组装数据,执行插入操作ps.setTimestamp(1, new Timestamp(value.getCreateDate().getTime()));ps.setString(2,value.getIp());ps.setString(3, value.getUuid());ps.setString(4, value.getArea());ps.setString(5, value.getWeb());ps.setString(6, value.getOperate());ps.addBatch();// 一次性写入int[] count = ps.executeBatch();System.out.println("成功写入MySQL数量:" + count.length);}
}

特别说明:从kafka读取到的内容是String,里面包含JSON格式。本文是先将它封装成Pojo对象,然后在Sink这里解析它的Value。(开始是尝试通过apply算子将它转换为List,但是失败了(时间有限,后续再弄),最后是通过map算子)

至此,Sink部分结束~!

2.4 Transform_2部分。消费kafka 数据,添加Sink。

package com.lzl.flink;import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
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.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/*** @author lzl* @create 2024-01-19 8:49* @name DataSourceFromKafka*/
public class DataSourceFromKafka {public static void transformFromKafka() throws Exception {// 构建流执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//kafka 配置Properties prop = new Properties();prop.put("bootstrap.servers", KafkaWriter.BROKER_LIST);prop.put("zookeeper.connect", "cdh39:2181");prop.put("group.id", KafkaWriter.TOPIC_WEB);prop.put("key.serializer", KafkaWriter.KEY_SERIALIZER);prop.put("value.serializer", KafkaWriter.VALUE_SERIALIZER);prop.put("auto.offset.reset", "earliest");// 建立流数据源DataStreamSource<String> dataStreamSource = env.addSource(new FlinkKafkaConsumer<String>(KafkaWriter.TOPIC_WEB,new SimpleStringSchema(),prop)).setParallelism(1); // 单线程打印,控制台不乱序,不影响结果SingleOutputStreamOperator<Web> webStream = env.addSource(new FlinkKafkaConsumer<>("web",new SimpleStringSchema(),prop)).setParallelism(1).map(string-> JSON.parseObject(string,Web.class));webStream.addSink(new MySqlToPojoSink());env.execute();}public static void main(String[] args) throws Exception {while (true) {try {// 每1毫秒写一条数据TimeUnit.MILLISECONDS.sleep(1);transformFromKafka();} catch (Exception e) {e.printStackTrace();}}}
}

如果要设置空值报错异常,或者排除空值可以:

package com.lzl.flink;import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import java.util.Properties;
import java.util.concurrent.TimeUnit;public class DataSourceFromKafka {public static void transformFromKafka() throws Exception {// 构建流执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);//checkpoint设置//每隔10s进行启动一个检查点【设置checkpoint的周期】env.enableCheckpointing(10000);//设置模式为:exactly_one,仅一次语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//确保检查点之间有1s的时间间隔【checkpoint最小间隔】env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);//检查点必须在10s之内完成,或者被丢弃【checkpoint超时时间】env.getCheckpointConfig().setCheckpointTimeout(10000);//同一时间只允许进行一次检查点env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);//表示一旦Flink程序被cancel后,会保留checkpoint数据,以便根据实际需要恢复到指定的checkpointenv.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//设置statebackend,将检查点保存在hdfs上面,默认保存在内存中。先保存到resources目录下env.setStateBackend(new FsStateBackend("D:java//Flink1.17//src//main//resources"));//         kafka 配置Properties prop = new Properties();prop.put("bootstrap.servers", KafkaWriter.BROKER_LIST);prop.put("zookeeper.connect", "cdh39:2181");prop.put("group.id", KafkaWriter.TOPIC_WEB);prop.put("key.serializer", KafkaWriter.KEY_SERIALIZER);prop.put("value.serializer", KafkaWriter.VALUE_SERIALIZER);prop.put("auto.offset.reset", "earliest")DataStreamSource<String> webStream = env.addSource(new FlinkKafkaConsumer<>("web",new SimpleStringSchema(),prop)).setParallelism(1);//使用process算子 排除空值DataStream<Web> processData = webStream.process(new ProcessFunction<String, Web>() {@Overridepublic void processElement(String s, Context context, Collector<Web> collector) throws Exception {try {Web webs = JSON.parseObject(s, Web.class);if (webs != null) {collector.collect(webs);}} catch (Exception e) {System.out.println("有空值数据");}}});processData.addSink(new MySqlToPojoSink());env.execute();}public static void main(String[] args) throws Exception {while (true) {try {// 每1毫秒写一条数据TimeUnit.MILLISECONDS.sleep(1);transformFromKafka();} catch (Exception e) {e.printStackTrace();}}}
}

至此,Transfrom结束~!

2.5 DB部分(这部分可以先做,或者放到前面,因为需要测试)
本次的DB演示采用常规的MySQL数据库。采用Druid工具连接。
思路:创建一个数据库连接的工具,用于连接数据库。使用Druid工具,然后放入具体的Driver,Url,数据库用户名和密码,初始化连接数,最大活动连接数,最小空闲连接数也就是数据库连接池,创建好之后返回需要的连接。

package com.lzl.flink;import com.alibaba.druid.pool.DruidDataSource;
import java.sql.Connection;
/*** @author lzl* @create 2024-01-18 17:58* @name DbUtils*/
public class DbUtils {private static DruidDataSource dataSource;public static Connection getConnection() throws Exception {dataSource = new DruidDataSource();dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");dataSource.setUrl("jdbc:mysql://cdh129:3306/flink?useSSL=true");dataSource.setUsername("root");dataSource.setPassword("xxb@5196");// 设置初始化连接数,最大连接数,最小闲置数dataSource.setInitialSize(10);dataSource.setMaxActive(50);dataSource.setMinIdle(5);// 返回连接return dataSource.getConnection();}
}

数据库建表语句:

CREATE TABLE `web_traffic_analysis` (`time` varchar(64) DEFAULT NULL COMMENT '时间',`ip` varchar(32) DEFAULT NULL COMMENT 'ip地址',`uid` varchar(32) DEFAULT NULL COMMENT 'uuid',`area` varchar(32) DEFAULT NULL COMMENT '地区',`web` varchar(64) DEFAULT NULL COMMENT '网址',`operate` varchar(32) DEFAULT NULL COMMENT '操作'
) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC COMMENT='网页流量分析表'

三、启动程序
开始本来是想将上面所有的功能都写成函数方法,然后单独开一个Main()主函数的入口,然后在主函数下面调用那些方法(生产数据、消费数据方法)。思路是借鉴python的:if name == ‘main’: 下调用很多的方法 。但实际执行过程,是先生成数据,然后将数据写入kafka,然后再消费数据,过程执行非常慢,这个方案被pass了。后来又想到多线程方案,一个线程跑生产数据和写入数据,一个线程跑消费数据和写入下游数据库。这个方法是测试成功了,但是跑了一会儿就出现数据的积压和内存oom了,因为我设定的是1毫秒生产一条数据,写入kafka也需要一定的时间,加上电脑内存不足,有点卡,这个方案也被pass了。最后的方案是将生产数据打包放到集群去跑,本地电脑开启消费kafka数据写入MySQL数据库。结果如下:
生产数据:
在这里插入图片描述
消费和写入数据库数据:
在这里插入图片描述
数据库数据:
在这里插入图片描述
至此结束,后面有其他想法再补充~!

多线程部分代码:

package com.example.study;import com.lzl.flink.DataSourceFromKafka;
import com.lzl.flink.KafkaWriter;public class WebApplication {public static void main(String[] args) throws Exception {// 创建线程1Thread threadOne = new Thread(new Runnable() {@Overridepublic void run() {while (true){try{KafkaWriter kafkaWriter = new KafkaWriter();kafkaWriter.webDataProducer();kafkaWriter.writeToKafka();System.out.println("线程一在跑~!");} catch (Exception e) {e.printStackTrace();}}}});// 创建线程2Thread threadTwo = new Thread(new Runnable() {@Overridepublic void run() {while (true) {DataSourceFromKafka dataSourceFromKafka = new DataSourceFromKafka();try {dataSourceFromKafka.transformFromKafka();System.out.println("线程二在跑~!");} catch (Exception e) {e.printStackTrace();}}}});//启动线程threadOne.start();threadTwo.start();Thread.sleep(5);}
}

结果:
在这里插入图片描述

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

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

相关文章

Linux|centos7操作系统|rtl8188gu芯片的网卡驱动使用DKMS安装管理

前言&#xff1a; DKMS是一个外置模块管理工具&#xff0c;可以自动build&#xff0c;安装各类驱动 DKMS (Dynamic Kernel Module Support) 是一种框架&#xff0c;用于在 Linux 操作系统中构建和维护内核模块。内核模块是 Linux 内核的一部分&#xff0c;它们提供了许多功能&…

(C++)简单计算器

文章目录 一、实验目的、内容二、实验程序设计及结构1.需求分析变量函数 2.设计结构或流程图 三、设计过程四、测试分析第一组第二组实验中出现的bug及解决方案 五、设计的特点和结果 一、实验目的、内容 输入是一个带有括号的四则运算表达式&#xff0c;输出是计算得出的正确…

详细分析Java中的list.foreach()和list.stream().foreach()

目录 前言1. 基本知识2. 差异之处2.1 执行顺序2.2 串行并行2.3 复杂数据处理2.4 CRUD集合2.5 迭代器 3. 总结4. 彩蛋 前言 典故来源于项目中使用了两种方式的foreach&#xff0c;后面尝试体验下有何区别&#xff01; 先看代码示例&#xff1a; 使用List的forEach&#xff1a…

手写Vue3源码

Vue3核心源码 B站视频地址&#xff1a;https://www.bilibili.com/video/BV1nW4y147Pd?p2&vd_source36bacfbaa95ea7a433650dab3f7fa0ae Monorepo介绍 Monorepo 是管理项目代码的一种方式&#xff0c;只在一个仓库中管理多个模块/包 一个仓库可以维护多个模块&#xff0c;…

打 jar 包运行 在windows 平台控制台和日志 乱码解决

--拒絕鷄巴囉嗦&#xff0c;直接解決問題 我们在Windows下运行jar包时&#xff0c;常常会出现乱码&#xff0c;主要分为dos窗口输出的日志中出现乱码和程序返回数据出现乱码。 dos窗口输出的日志中出现乱码 执行如下命令&#xff0c;将控制台输出编码改为UTF8&#xff1a; ch…

基于springboot+vue的足球青训俱乐部管理系统(前后端分离)

博主主页&#xff1a;猫头鹰源码 博主简介&#xff1a;Java领域优质创作者、CSDN博客专家、公司架构师、全网粉丝5万、专注Java技术领域和毕业设计项目实战 主要内容&#xff1a;毕业设计(Javaweb项目|小程序等)、简历模板、学习资料、面试题库、技术咨询 文末联系获取 研究背景…

Tkinter + removebg实现Ai抠图

我们先对removebg进行一个说明&#xff1a; "removebg" 是一个在线服务&#xff0c;它允许用户自动去除图片的背景。用户只需上传图片&#xff0c;系统会通过算法自动识别并去除背景&#xff0c;生成透明背景&#xff08;PNG格式&#xff09;或者用户指定的新背景图…

Qt6入门教程 10:菜单栏、工具栏和状态栏

目录 一.菜单栏 1.Qt Designer 1.1添加菜单和菜单项 1.2添加二级菜单 1.3给菜单和菜单项添加图标 1.4给菜单项添加功能 2.纯手写 二.工具栏 1.Qt Designer 1.1添加工具栏按钮 1.2工具栏的几个重要属性 2.纯手写 三.状态栏 1.Qt Designer 2.纯手写 用Qt Creator新…

excel学习1

直接ctrl cctrl v会报错位移选择粘贴时用123那个数字粘贴而不是ctrl V 只要结果不要公式 上面复制的为数值这里是复制的公式他们两个不一样 这个方法太麻烦了直接用格式刷&#xff0c;选择一个区域一个单元格&#xff0c;不要选择多个一刷就出来了 第一个计算后向下拖就行了&…

Overleaf(LaTeX文档在线编写平台)使用学习记录

一、LaTeX简概[1] LaTeX&#xff0c;是一种基于TEX的排版系统&#xff0c;是一种可以处理排版和渲染的标记语言。由美国计算机科学家莱斯利兰伯特在20世纪80年代初期开发&#xff0c;利用这种格式系统的处理&#xff0c;即使用户没有排版和程序设计的知识也可以充分发挥由TEX所…

SpringBoot整合ElasticSearch实现基础的CRUD操作

本文来说下SpringBoot整合ES实现CRUD操作 文章目录 概述spring-boot-starter-data-elasticsearch项目搭建ES简单的crud操作保存数据修改数据查看数据删除数据 本文小结 概述 SpringBoot支持两种技术和es交互。一种的jest&#xff0c;还有一种就是SpringData-ElasticSearch。根据…

centos7安装Redis7.2.4

文章目录 下载Redis解压Redis安装gcc依赖&#xff08;Redis是C语言编写的&#xff0c;编译需要&#xff09;编译安装src目录下二进制文件安装到/usr/local/bin修改redis.conf文件启动redis服务外部连接测试 参考&#xff1a; 在centos中安装redis-5.0.7 Memory overcommit must…

docker 部署及命令

一、容器概述 1、为什么要用到容器&#xff1f; ①容器可以屏蔽底层操作系统的差异性&#xff0c;让业务应用不管在哪里都是使用容器的环境运行&#xff0c;从而保证开发测试环境与生产环境的一致性 ②容器部署起来非常便捷和迅速&#xff0c;缩短开发测试部署的周期时间 2…

神经网络的学习(Neural Networks: Learning)

1.代价函数 案例&#xff1a;假设神经网络的训练样本有&#x1d45a;个&#xff0c;每个包含一组输入&#x1d465;和一组输出信号&#x1d466;&#xff0c;&#x1d43f;表示神经网络层数&#xff0c;&#x1d446;&#x1d43c;表示每层的 neuron 个数(&#x1d446;&#…

【Java-框架-SpringSecurity】单点登录(认证和授权)- 随笔

项目文件&#xff1b; 【1】 预览 文件 文件01 名称 pom.xml&#xff1b; 内容 &#xff08;01&#xff09;总的 <?xml version"1.0" encoding"UTF-8"?> <project xmlns"http://maven.apache.org/POM/4.0.0" xmlns:xsi"http:…

18.鸿蒙HarmonyOS App(JAVA)日期选择器-时间选择器

18.鸿蒙HarmonyOS App(JAVA)日期选择器-时间选择器 点击button按钮触发事件显示月份与获取的时间 Button button3 (Button) findComponentById(ResourceTable.Id_button3);button3.setClickedListener(new Component.ClickedListener() {Overridepublic void onClick(Compon…

【趣味CSS3.0】粘性定位属性Position:sticky是不是真的没用了?

&#x1f680; 个人主页 极客小俊 ✍&#x1f3fb; 作者简介&#xff1a;web开发者、设计师、技术分享博主 &#x1f40b; 希望大家多多支持一下, 我们一起学习和进步&#xff01;&#x1f604; &#x1f3c5; 如果文章对你有帮助的话&#xff0c;欢迎评论 &#x1f4ac;点赞&a…

Webpack5 基本使用 - 2

常用 loader loader 是辅助打包工具。webpack 默认只能打包 js 文件&#xff0c;打包其它模块就需要配置 loader 来告诉 webpack 该怎么去打包其它文件。loader 可以将文件从不同的语言转换为 JavaScript。一类文件如果需要多个 loader 处理&#xff0c;loader 的执行顺序是从…

SQL注入 ---> Day1 !

终于&#xff0c;我的课程开始讲SQL注入了&#xff0c;对就是那个常年在OWASP上有名的SQL注入 今天是真的冷啊&#xff0c;幸苦的小编还在写csdn 1.SQL注入原理 SQl注入其实就是将恶意的代码向服务器提交&#xff0c;但是后端又不过滤而引发的漏洞&#xff0…

2、Line Charts折线图

可视化时间趋势 现在你已经熟悉了编码环境,是时候学习如何制作自己的图表了! 在本教程中,您将学习足够的Python来创建专业外观的折线图。然后,在接下来的练习中,您将使用您的最新技能处理真实世界的数据集。 本课程数据集夸克网盘下载链接:https://pan.quark.cn/s/a235ac…