Canal监听MySQL增量数据
+Canal,主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费服务。
+分享主要围绕如何使用Canal快速搭建Mysql增量数据订阅服务，针对Mysql表的增删改查服务进行监听。
...大约 3 分钟
404
页面不存在
我们是怎么来到这儿的？
Canal,主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费服务。
+分享主要围绕如何使用Canal快速搭建Mysql增量数据订阅服务，针对Mysql表的增删改查服务进行监听。
线程代码何时适合加锁？
+++参考：什么是线程池
+
线程池其实是一种池化的技术实现，池化技术的核心思想就是实现资源的复用，避免资源的重复创建和销毁带来的性能开销。线程池可以管理一堆线程，让线程执行完任务之后不进行销毁，而是继续去处理其它线程已经提交的任务。
swagger官网: https://swagger.io/
+Swagger 是一个规范和完整的框架，用于生成、描述、调用和可视化 RESTful 风格的 Web 服务。
<dependency>
+ <groupId>io.springfox</groupId>
+ <artifactId>springfox-boot-starter</artifactId>
+ <version>3.0.0</version>
+</dependency>
+Java Virtual Machine(JAVA虚拟机，JVM)内存结构, 包括 程序计数器，虚拟机栈，本地方法栈，方法区，堆几部分组成。JVM的使用是为保证Java程序跨平台运行，在操作系统层次之上构建的虚拟机，以便实现一次编译，到处运行。
+介绍ES的分布式架构原理，包括ES存储结构和ES分布式框架设计
+本篇文章分享Maven的作用和自动化构建项目的常见方法，如clean,install,package等命令
本篇文章分享Maven的作用和自动化构建项目的常见方法，如clean,install,package等命令
Maven是Apache软件基金会唯一维护的一款自动化构建工具，专注于服务Java平台的项目构建和依赖管理。
Maven是基于项目对象模型（POM），可以通过一小段描述信息来管理项目的构建、报告和文档的软件项目管理工具。
target），为重新编译做好准备target目录。package包含了compile命令的过程。mvn -version/-v 显示版本信息
+
+mvn clean 清空生成的文件
+
+mvn compile 编译
+
+mvn test 编译并测试
+
+mvn package 生成target目录，编译、测试代码，生成测试报告，生成jar/war文件
+
+mvn site 生成项目相关信息的网站
+
+mvn clean compile 表示先运行清理之后运行编译，会将代码编译到target文件夹中
+
+mvn clean package 运行清理和打包
+
+mvn clean install 运行清理和安装，会将打好的包安装到本地仓库中，以便其他的项目可以调用
+
+mvn clean deploy 运行清理和发布
+mvn clean package依次执行了clean、resources、compile、testResources、testCompile、test、jar(打包)等７个阶段。
mvn clean install依次执行了clean、resources、compile、testResources、testCompile、test、jar(打包)、install等8个阶段。
mvn clean deploy依次执行了clean、resources、compile、testResources、testCompile、test、jar(打包)、install、deploy等９个阶段。
提示
-DskipTests跟-Dmaven.test.skip=true的区别
不执行测试用例，但编译测试用例类生成相应的class文件至target/test-classes下。
不执行测试用例，也不编译测试用例类。
一般建议使用第二种，直接忽略测试的编译，如下：
mvn clean package -Dmaven.test.skip=true
+swagger官网: https://swagger.io/
\\nSwagger 是一个规范和完整的框架，用于生成、描述、调用和可视化 RESTful 风格的 Web 服务。
<dependency>\\n <groupId>io.springfox</groupId>\\n <artifactId>springfox-boot-starter</artifactId>\\n <version>3.0.0</version>\\n</dependency>\\n<dependency>
+ <groupId>io.springfox</groupId>
+ <artifactId>springfox-boot-starter</artifactId>
+ <version>3.0.0</version>
+</dependency>
+springfox:
+ documentation:
+ swagger-ui:
+ enabled: true
+import io.swagger.annotations.ApiOperation;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpMethod;
+import springfox.documentation.builders.*;
+import springfox.documentation.oas.annotations.EnableOpenApi;
+import springfox.documentation.schema.ScalarType;
+import springfox.documentation.service.*;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spring.web.plugins.Docket;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author: SunHB
+ * @createTime: 2023/09/24 上午11:00
+ * @description:
+ */
+@Configuration
+@EnableOpenApi
+public class Swagger3Config {
+
+
+ @Bean
+ public Docket createRestApi() {
+ //返回文档摘要信息
+ return new Docket(DocumentationType.OAS_30)
+ .apiInfo(apiInfo())
+ .select()
+ .apis(RequestHandlerSelectors.withMethodAnnotation(ApiOperation.class))
+ //.apis(RequestHandlerSelectors.basePackage("com.ytkj.controller"))
+ .paths(PathSelectors.any())
+ .build()
+ .globalRequestParameters(getGlobalRequestParameters())
+ .globalResponses(HttpMethod.GET, getGlobalResponseMessage())
+ .globalResponses(HttpMethod.POST, getGlobalResponseMessage());
+ }
+
+ /**
+ * 生成接口信息，包括标题、联系人等
+ */
+ private ApiInfo apiInfo() {
+ return new ApiInfoBuilder()
+ .title("yantu测试接口文档")
+ .description("如有疑问，可联系孙鸿博")
+ .version("1.0")
+ .build();
+ }
+
+ /**
+ * 封装全局通用参数
+ */
+ private List<RequestParameter> getGlobalRequestParameters() {
+ List<RequestParameter> parameters = new ArrayList<>();
+ return parameters;
+ }
+
+ /**
+ * 封装通用响应信息
+ */
+ private List<Response> getGlobalResponseMessage() {
+ List<Response> responseList = new ArrayList<>();
+ responseList.add(new ResponseBuilder().code("404").description("未找到资源").build());
+ return responseList;
+ }
+
+
+}
+@Api：用在类上，说明该类的作用。
@ApiOperation：注解来给API增加方法说明。
@ApiImplicitParams : 用在方法上包含一组参数说明。
@ApiImplicitParam：用来注解来给方法入参增加说明。
@ApiResponses：用于表示一组响应
@RestController
+@Slf4j
+@RequestMapping("/test")
+@Api(tags = "测试接口管理")
+public class TestController
+@ApiOperation(value = "主控测试接口")
+@PostMapping("/chat")
+@CrossOrigin
+public Result<Map> testChat
+@ApiOperation(value = "文档模块测试接口")
+@PostMapping("v1/docQA")
+@CrossOrigin
+public Result docQA
+\\n\\n参考：什么是线程池
\\n
线程池其实是一种池化的技术实现，池化技术的核心思想就是实现资源的复用，避免资源的重复创建和销毁带来的性能开销。线程池可以管理一堆线程，让线程执行完任务之后不进行销毁，而是继续去处理其它线程已经提交的任务。
","autoDesc":true}');export{e as data}; diff --git a/assets/ThreadPoolExecutor.html-ee29ff56.js b/assets/ThreadPoolExecutor.html-ee29ff56.js new file mode 100644 index 0000000..8614176 --- /dev/null +++ b/assets/ThreadPoolExecutor.html-ee29ff56.js @@ -0,0 +1,37 @@ +import{_ as e}from"./plugin-vue_export-helper-c27b6911.js";import{r as t,o as p,c as o,a as n,b as s,d as c,f as l}from"./app-cbb8ea4f.js";const i={},u=n("h1",{id:"threadpoolexecutor线程池",tabindex:"-1"},[n("a",{class:"header-anchor",href:"#threadpoolexecutor线程池","aria-hidden":"true"},"#"),s(" ThreadPoolExecutor线程池")],-1),r={href:"https://javabetter.cn/thread/pool.html#%E4%B8%80%E3%80%81%E4%BB%80%E4%B9%88%E6%98%AF%E7%BA%BF%E7%A8%8B%E6%B1%A0",target:"_blank",rel:"noopener noreferrer"},k=l(`线程池其实是一种池化的技术实现，池化技术的核心思想就是实现资源的复用，避免资源的重复创建和销毁带来的性能开销。线程池可以管理一堆线程，让线程执行完任务之后不进行销毁，而是继续去处理其它线程已经提交的任务。
// 一共分为三步
+public void execute(Runnable command) {
+ // 首先检查提交的任务是否为null，是的话则抛出NullPointerException。
+ if (command == null)
+ throw new NullPointerException();
+ // 获取线程池的状态，包括线程池状态（？），工作线程数量，初始数量为负数
+ // get获取线程的value
+ int c = ctl.get();
+
+ // 1. 检查当前运行的工作线程数是否少于核心线程数（corePoolSize）
+ if (workerCountOf(c) < corePoolSize) {
+ // 少于核心线程数量则添加worker
+ if (addWorker(command, true))
+ return;
+ c = ctl.get();
+ }
+
+ // 2. 尝试将任务添加到任务队列中
+ if (isRunning(c) && workQueue.offer(command)) {
+ int recheck = ctl.get();
+ // 双重检查线程池的状态
+ if (! isRunning(recheck) && remove(command)) // 如果线程池已经停止，从队列中移除任务， && 左面为true才会执行remove函数移除任务
+ reject(command);
+ // 如果线程池正在运行，但是工作线程数为0，尝试添加一个新的工作线程
+ else if (workerCountOf(recheck) == 0)
+ addWorker(null, false);
+ }
+ // 3. 如果任务队列满了，尝试添加一个新的非核心工作线程来执行任务
+ else if (!addWorker(command, false))
+ reject(command);
+ }
+private static final int CAPACITY = (1 << COUNT_BITS) - 1;
+private static int workerCountOf(int c) { return c & CAPACITY; } //与运算获取线程数
+private static final int SHUTDOWN = 0 << COUNT_BITS;
+private static boolean isRunning(int c) { return c < SHUTDOWN; } //小于0，即为线程池正在运行
+workQueue.offer(command) // 能够添加返回True,不能添加返回False
+介绍ES的分布式架构原理，包括ES存储结构和ES分布式框架设计
+`,r:{minutes:2.34,words:701},y:"a",t:"ES分布式的结构分享",i:"article"},[":md"]],["v-09041878","/posts/java/",{y:"p",t:"JAVA学习",i:"java"},["/posts/java/README.md"]],["v-15460eb0","/posts/java/ThreadPoolExecutor.html",{d:16955136e5,l:"2023年9月24日",c:["JAVA学习"],g:["多线程","JUC"],e:`++参考：什么是线程池
+
线程池其实是一种池化的技术实现，池化技术的核心思想就是实现资源的复用，避免资源的重复创建和销毁带来的性能开销。线程池可以管理一堆线程，让线程执行完任务之后不进行销毁，而是继续去处理其它线程已经提交的任务。
`,r:{minutes:1.77,words:530},y:"a",t:"ThreadPoolExecutor线程池",i:"editor"},[":md"]],["v-1832e065","/posts/java/intro.html",{y:"p",t:"JAVA介绍页",i:"java"},[":md"]],["v-0c020534","/posts/java/javalock.html",{d:16996608e5,l:"2023年11月11日",c:["JAVA学习"],g:["JAVA锁","JAVA异常"],e:`线程代码何时适合加锁？
+`,r:{minutes:2.42,words:726},y:"a",t:"JAVA何时加锁",i:"editor"},[":md"]],["v-59a01cfa","/posts/java/javap.html",{d:16935264e5,l:"2023年9月1日",c:["JAVA学习"],g:["常量池","JVM异常"],e:`Java Virtual Machine(JAVA虚拟机，JVM)内存结构, 包括 程序计数器，虚拟机栈，本地方法栈，方法区，堆几部分组成。JVM的使用是为保证Java程序跨平台运行，在操作系统层次之上构建的虚拟机，以便实现一次编译，到处运行。
+`,r:{minutes:2.75,words:825},y:"a",t:"JVM内存结构",i:"editor"},[":md"]],["v-37c966b9","/posts/project/Maven.html",{d:16930944e5,l:"2023年8月27日",c:["项目分享"],g:["maven","springboot"],e:`本篇文章分享Maven的作用和自动化构建项目的常见方法，如clean,install,package等命令
swagger官网: https://swagger.io/
+Swagger 是一个规范和完整的框架，用于生成、描述、调用和可视化 RESTful 风格的 Web 服务。
<dependency>
+ <groupId>io.springfox</groupId>
+ <artifactId>springfox-boot-starter</artifactId>
+ <version>3.0.0</version>
+</dependency>
+Canal,主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费服务。
+分享主要围绕如何使用Canal快速搭建Mysql增量数据订阅服务，针对Mysql表的增删改查服务进行监听。
1.1 数据同步
1.2 数据展示
1.3 低版本迁移高版本
选择canal.deployer-1.1.7.tar.gz下载
提示
canal的各个组件的用途各不相同，下面分别介绍下：
canal-deploy：用于监听MySQL的binlog，是一个伪装的MySQL从库，只负责从MySQL主库接收数据，不做处理。
canal-adapter：canal的客户端，从canal-deploy中获取数据，然后同步数据到目标数据源。
canal-admin：为canal提供整体配置管理、节点运维等面向运维的功能，提供相对友好的WebUI操作界面，方便更多用户快速和安全的操作
[mysqld]
+log-bin=mysql-bin # 开启 binlog
+binlog-format=ROW # 选择 ROW 模式
+server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
+show variables like \`log_bin\`;# 显示ON 即为打开binlog模式
+canal.instance.dbUsername,canal.instance.dbPassword## mysql serverId , v1.0.26+ will autoGen
+## v1.0.26版本后会自动生成slaveId，所以可以不用配置
+# canal.instance.mysql.slaveId=0
+
+# 数据库地址
+canal.instance.master.address=127.0.0.1:3306
+# binlog日志名称
+canal.instance.master.journal.name=mysql-bin.000001
+# mysql主库链接时起始的binlog偏移量
+canal.instance.master.position=154
+# mysql主库链接时起始的binlog的时间戳
+canal.instance.master.timestamp=
+canal.instance.master.gtid=
+
+# username/password
+# 在MySQL服务器授权的账号密码
+canal.instance.dbUsername=xxx
+canal.instance.dbPassword=xxx
+# 字符集
+canal.instance.connectionCharset = UTF-8
+# enable druid Decrypt database password
+canal.instance.enableDruid=false
+
+# table regex .*\\\\..*表示监听所有表 也可以写具体的表名，用，隔开
+canal.instance.filter.regex=.*\\\\..*
+# mysql 数据解析表的黑名单，多个表用，隔开
+canal.instance.filter.black.regex=
+bin 目录下 start.sh (Linux系统)或者start.bat(Windows系统)./start.sh
+jps 命令可以查看是否正常启动
正常启动显示xxxx CanalLauncher
<dependency>
+ <groupId>com.alibaba.otter</groupId>
+ <artifactId>canal.client</artifactId>
+ <version>1.1.0</version>
+ </dependency>
+public static void main(String[] args) throws InvalidProtocolBufferException {
+ //1.获取 canal 连接对象
+ CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("localhost", 11111), "example", "", "");
+ while (true) {
+ //2.获取连接
+ canalConnector.connect();
+ //3.指定要监控的数据库
+ canalConnector.subscribe("canal_demo.*");
+ //4.获取 Message
+ // 一个Message 对应一个Sql语句
+ Message message = canalConnector.get(100);
+ List<CanalEntry.Entry> entries = message.getEntries();
+ if (entries.size() <= 0) {
+ System.out.println("没有数据，休息一会");
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ } else {
+ for (CanalEntry.Entry entry : entries) {
+ // 获取表名
+ String tableName = entry.getHeader().getTableName();
+ // Entry 类型
+ CanalEntry.EntryType entryType = entry.getEntryType();
+ // 判断 entryType 是否为 ROWDATA
+ if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
+ // 序列化数据
+ ByteString storeValue = entry.getStoreValue();
+ // 反序列化
+ CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
+ // 获取事件类型
+ CanalEntry.EventType eventType = rowChange.getEventType();
+ // 获取具体的数据
+ List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
+ // 遍历并打印数据
+ for (CanalEntry.RowData rowData : rowDatasList) {
+ List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
+ Map<String, Object> bMap = new HashMap<String,Object>();
+ for (CanalEntry.Column column : beforeColumnsList) {
+ bMap.put(column.getName(), column.getValue());
+ }
+ Map<String, Object> afMap = new HashMap<String,Object>();
+ List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
+ for (CanalEntry.Column column : afterColumnsList) {
+ afMap.put(column.getName(), column.getValue());
+ }
+ System.out.println("表名:" + tableName + ",操作类型:" + eventType);
+ System.out.println("改前:" + bMap );
+ System.out.println("改后:" + afMap );
+ }
+ }
+ }
+ }
+ }
+ }
+
+UPDATE USER set name = "ssskkk" where name = "zhangsan";
+修改数据后的日志打印
没有数据，休息一会
+没有数据，休息一会
+没有数据，休息一会
+表名:USER,操作类型:UPDATE
+改前:{name=zhangsan, id=4, age=18}
+改后:{name=ssskkk, id=4, age=18}
+表名:USER,操作类型:UPDATE
+改前:{name=zhangsan, id=5, age=188}
+改后:{name=ssskkk, id=5, age=188}
+Canal,主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费服务。
\\n分享主要围绕如何使用Canal快速搭建Mysql增量数据订阅服务，针对Mysql表的增删改查服务进行监听。