要做什么因为项目需要异地对达梦数据库进行实时数据同步，需要采用一个免费的方案，所以采用 达梦8 Flink CDC方案。在我的场景中，需要将20个子节点的数据同步到一个中心节点，且子节点与中心节点的网络并不稳定。因此有如下设计：子节点1 (DM) ──► Flink CDC Job1 ──► Kafka Topic (node1) 子节点2 (DM) ──► Flink CDC Job2 ──► Kafka Topic (node2) ... 子节点20 (DM) ──► Flink CDC Job20 ──► Kafka Topic (node20) Kafka ──► Flink Merge Job (1 个) ──► 顶级节点 (DM)简单来说，流程如下：1.每个子节点都有自己的 达梦8 数据库，在达梦数据库中开启归档模式以支持CDC；2.每个子节点单独配置一个Flink CDC Job，从达梦8 数据库中读取数据库变化记录；将读取到的记录转成 SQL 语句，随后写入 Kafka中；3.顶级节点从Kafka中订阅记录并通过 JDBC 写入自己的数据库中；怎么做按照上面说的流程，分为三个大的步骤：达梦8开启归档模式使用SYSDBA用户开启归档模式-- 1. 切换数据库到 MOUNT 状态 ALTER DATABASE MOUNT; -- 2. 开启归档模式 ALTER DATABASE ARCHIVELOG; -- 3. 添加一个本地归档配置（目录、文件大小、空间限制请按需修改， DEST指定的目录需要先创建好，并且这个目录数据库用户要有写入权限） ALTER DATABASE ADD ARCHIVELOG 'DEST = /dm8/arch, TYPE = LOCAL, FILE_SIZE = 1024, SPACE_LIMIT = 20480'; -- 4. 切回 OPEN 状态 ALTER DATABASE OPEN;验证是否成功-- 以 SYSDBA 执行 SELECT para_name, para_value FROM v$dm_ini WHERE para_name IN ('ARCH_INI', 'RLOG_APPEND_LOGIC');查询结果应该是这样的：LINEID PARA_NAME PARA_VALUE---------- ----------------- ----------1 RLOG_APPEND_LOGIC 22 ARCH_INI 1查出来的两个结果应该都不为 0 ，如果你查出来的值为0 ，可以通过以下 SQL 更改：-- 1. 切换数据库到 MOUNT 状态 ALTER DATABASE MOUNT; -- 2. 设置（推荐值，多数教程一致） SP_SET_PARA_VALUE(1, 'ARCH_INI', 1); SP_SET_PARA_VALUE(1, 'RLOG_APPEND_LOGIC', 2); -- 1 或 2 均可，2 更全面支持 UPDATE/DELETE -- 3. 切回 OPEN 状态 ALTER DATABASE OPEN; 特别注意：有的文章中只是用 select arch_mode from v$database; 来验证是否开启了归档模式。当 arch_mode 值为 Y 时，即代码开启了归档模式，其可以实现对历史数据的记录归档，但是如果此时上面写到的 RLOG_APPEND_LOGIC 值是 0，则 UPDATE、DELETE 等语句实际没有被记录到。这也是很多人反馈更新记录没有获取到的原因。搭建 Flink +Kafka 环境（docker）环境搭建我们采用最简单的方式进行，直接使用 docker-compose一键启动即可！要的资源不多，在本地的4G的虚拟机上也能正常启动，唯一到担心的就是你的网络能不能顺利的把镜像拉下来docker-compose.yaml注意：启动前把里面 kafka 的配置里面 IP改为自己服务器的IP，目前我写的是 192.168.20.220，只有一个地方version: "3.8" services: # ==================== Flink JobManager ==================== jobmanager: image: flink:1.20.3-scala_2.12-java17 container_name: flink-jobmanager hostname: jobmanager restart: unless-stopped ports: - "8081:8081" - "6123:6123" command: jobmanager environment: TZ: Asia/Shanghai FLINK_JOBMANAGER_RPC_ADDRESS: jobmanager FLINK_JOBMANAGER_RPC_PORT: 6123 FLINK_PROPERTIES: | jobmanager.rpc.address: jobmanager jobmanager.rpc.port: 6123 jobmanager.heap.size: 1024m taskmanager.numberOfTaskSlots: 4 parallelism.default: 4 state.backend: rocksdb state.checkpoints.dir: file:///opt/flink/checkpoints volumes: - ./flink-logs:/opt/flink/log - ./checkpoints:/opt/flink/checkpoints - ./jars:/opt/flink/usrlib - /etc/localtime:/etc/localtime:ro networks: - flink-net # ==================== Flink TaskManager ==================== taskmanager: image: flink:1.20.3-scala_2.12-java17 hostname: taskmanager restart: unless-stopped deploy: replicas: 2 depends_on: - jobmanager command: taskmanager environment: TZ: Asia/Shanghai FLINK_JOBMANAGER_RPC_ADDRESS: jobmanager FLINK_JOBMANAGER_RPC_PORT: 6123 FLINK_PROPERTIES: | jobmanager.rpc.address: jobmanager jobmanager.rpc.port: 6123 taskmanager.heap.size: 1536m taskmanager.numberOfTaskSlots: 4 taskmanager.memory.process.size: 2048m volumes: - ./flink-logs:/opt/flink/log - ./jars:/opt/flink/usrlib - /etc/localtime:/etc/localtime:ro networks: - flink-net # ==================== Kafka ==================== kafka: image: apache/kafka:3.8.0 container_name: kafka restart: unless-stopped ports: - "9092:9092" environment: TZ: Asia/Shanghai KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.20.220:9092 KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_LOG_DIRS: /tmp/kraft-combined-logs KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 networks: - flink-net # ==================== Flink SQL Client ==================== sql-client: image: flink:1.20.3-scala_2.12-java17 container_name: flink-sql-client restart: unless-stopped command: /opt/flink/bin/sql-client.sh depends_on: - jobmanager environment: TZ: Asia/Shanghai FLINK_JOBMANAGER_RPC_ADDRESS: jobmanager FLINK_JOBMANAGER_RPC_PORT: 6123 volumes: - ./jars:/opt/flink/usrlib - ./sql:/opt/flink/sql stdin_open: true tty: true networks: - flink-net networks: flink-net: driver: bridge启动成功后，访问你服务器的 8081 端口就可以看到你的 Flink 界面了。关于版本的选择，参考如下（来自参考地址【1】）Flink CDC 版本Flink版本JDK版本DM8版本2.4.11.13.、1.14.、1.15.、1.16.、1.17.*1.8.08.1.3.77及以上3.0.01.14.、1.15.、1.16.、1.17.、1.18.*1.8.0及以上8.1.3.77及以上3.3.01.19.、1.20.1.8.0及以上8.1.3.77及以上Flink Job将数据库操作记录写入KafkaFlink 可以通 Flink SQL 或 DataStream API来定义 Job ,在本文中，我们采用 DataStream API， java代码的方式实现。（需要了解Flink SQL 的可以查看文章最后的参考地址）很简单，总共3个文件：引入需要的依赖：pom.xml<?xml version="1.0" encoding="UTF-8"?> <project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>org.bg</groupId> <artifactId>DmCdc</artifactId> <version>0.0.1-SNAPSHOT</version> <name>DmCdc</name> <description>达梦8 Flink CDC</description> <properties> <java.version>17</java.version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.binary.version>2.12</scala.binary.version> <flink.version>1.20.3</flink.version> <flinkcdc.version>3.3.0</flinkcdc.version> <version.infinispan>13.0.17.Final</version.infinispan> <version.infinispan.protostream>4.4.4.Final</version.infinispan.protostream> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <!-- <scope>provided</scope>--> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner_${scala.binary.version}</artifactId> <version>${flink.version}</version> <!-- <scope>provided</scope>--> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-base</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-cdc-base</artifactId> <version>${flinkcdc.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-debezium</artifactId> <version>${flinkcdc.version}</version> </dependency> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-dm-cdc</artifactId> <version>${flinkcdc.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>3.3.0-1.20</version> </dependency> <dependency> <groupId>com.dameng</groupId> <artifactId>DmJdbcDriver18</artifactId> <version>8.1.3.140</version> </dependency> <dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.83</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-slf4j-impl</artifactId> <version>2.17.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-api</artifactId> <version>2.17.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.17.1</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-1.2-api</artifactId> <version>2.17.1</version> </dependency> <dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.4.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.5.0</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>org.bg.dm.cdc.DmCdcApplication</mainClass> </transformer> </transformers> <createDependencyReducedPom>false</createDependencyReducedPom> </configuration> </execution> </executions> </plugin> </plugins> </build> </project> 配置好日志打印文件log4j2.xml 放到 resource下<?xml version="1.0" encoding="UTF-8"?> <Configuration status="warn" monitorInterval="5"> <properties> <property name="LOG_HOME">D:/java/logs</property> <property name="FILE_NAME">${date:yyyy-MM-dd}</property> </properties> <Appenders> <Console name="Console" target="SYSTEM_OUT"> <PatternLayout pattern="%d{HH:mm:ss.SSS}[%-5level]%m%n"/> </Console> <RollingRandomAccessFile name="MyFile" fileName="D:/test.log" filePattern="${LOG_HOME}/${date:yyyy-MM}/${FILE_NAME}-%d{yyyy-MM-dd HH-mm}-%i.log"> <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/> <Policies> <TimeBasedTriggeringPolicy interval="120"/> <SizeBasedTriggeringPolicy size="10 MB"/> </Policies> <DefaultRolloverStrategy max="20"/> </RollingRandomAccessFile> </Appenders> <Loggers> <AsyncLogger name="com.bg" level="trace" additivity="false"> <AppenderRef ref="MyFile"/> <AppenderRef ref="Console"/> </AsyncLogger> <Root level="info"> <AppenderRef ref="MyFile"/> <AppenderRef ref="Console"/> </Root> <Logger name="mylog" level="trace" additivity="false"> <AppenderRef ref="MyFile" /> </Logger> </Loggers> </Configuration>编写CDC业务代码 注意： 里面包含了数据库的连接信息，改成自己的**推荐先执行这个简单的版本进行调试：DmCdcApplication.javapackage org.bg.dm.cdc; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource; import org.apache.flink.cdc.connectors.dm.source.DMSourceBuilder; import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.DeliveryGuarantee; import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema; import org.apache.flink.connector.kafka.sink.KafkaSink; import org.apache.flink.runtime.state.filesystem.FsStateBackend; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.time.Instant; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.Properties; public class DmCdcApplication1 { public static void main(String[] args) throws Exception { Properties properties = new Properties(); properties.setProperty("database.tablename.case.insensitive", "false"); properties.setProperty("log.mining.strategy", "offline_catalog"); properties.setProperty("log.mining.continuous.mine", "true"); properties.setProperty("lob.enabled", "true"); // Schema History 配置（必须正确设置） properties.setProperty("schema.history.internal", "io.debezium.storage.kafka.history.KafkaSchemaHistory"); properties.setProperty("schema.history.internal.kafka.bootstrap.servers", "192.168.20.220:9092"); properties.setProperty("schema.history.internal.kafka.topic", "dm-history-topic"); // 增加稳定性配置 properties.setProperty("snapshot.mode", "initial"); // 先用 initial 测试 properties.setProperty("snapshot.fetch.size", "512"); properties.setProperty("connect.timeout.ms", "60000"); properties.setProperty("heartbeat.interval.ms", "3000"); JdbcIncrementalSource<String> changeEventSource = new DMSourceBuilder<String>() .hostname("192.168.128.5") .port(5236) .databaseList("DMDB") .tableList("BACKGROUND_DATA.OPEN_API_APP") // 推荐这样写多表 .schemaList("BACKGROUND_DATA") .username("SYSDBA") .password("123456") .startupOptions(StartupOptions.initial()) // 先用 initial 测试 .dmProperties(properties) .includeSchemaChanges(true) .deserializer(new JsonDebeziumDeserializationSchema()) .sliceSize(20) .scanNewlyAddedTableEnabled(true) .build(); Configuration configuration = new Configuration(); // 检查点文件 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration); env.enableCheckpointing(20 * 1000); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(6 * 1000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.setStateBackend(new FsStateBackend("file:///opt/flink/dmcdc/ck")); DataStream<String> sourceStream = env.fromSource( changeEventSource, WatermarkStrategy.noWatermarks(), "DmSource" ); sourceStream.print(); env.execute(); } } 正常情况下，如果你配置的表中有数据，那你的日志中会打印出相应的数据来。且你去数据库执行更新操作会被记录并打印上面的版本能跑通之后，可以跑写入的 Kafka 的版本（从 DM8 中获取操作记录，转为 SQL 后写入 Kafka）:DmCdcApplication.javapackage org.bg.dm.cdc; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.JsonNodeFactory; import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.cdc.connectors.base.options.StartupOptions; import org.apache.flink.cdc.connectors.base.source.jdbc.JdbcIncrementalSource; import org.apache.flink.cdc.connectors.dm.source.DMSourceBuilder; import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.configuration.Configuration; import org.apache.flink.connector.base.DeliveryGuarantee; import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema; import org.apache.flink.connector.kafka.sink.KafkaSink; import org.apache.flink.runtime.state.filesystem.FsStateBackend; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.time.Instant; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.Properties; public class DmCdcApplication { public static void main(String[] args) throws Exception { Properties properties = new Properties(); properties.setProperty("database.tablename.case.insensitive", "false"); properties.setProperty("log.mining.strategy", "offline_catalog"); properties.setProperty("log.mining.continuous.mine", "true"); properties.setProperty("lob.enabled", "true"); // Schema History 配置（必须正确设置） properties.setProperty("schema.history.internal", "io.debezium.storage.kafka.history.KafkaSchemaHistory"); properties.setProperty("schema.history.internal.kafka.bootstrap.servers", "192.168.20.220:9092"); properties.setProperty("schema.history.internal.kafka.topic", "dm-history-topic"); // 增加稳定性配置 properties.setProperty("snapshot.mode", "initial"); // 先用 initial 测试 properties.setProperty("snapshot.fetch.size", "512"); properties.setProperty("connect.timeout.ms", "60000"); properties.setProperty("heartbeat.interval.ms", "3000"); JdbcIncrementalSource<String> changeEventSource = new DMSourceBuilder<String>() .hostname("192.168.128.5") .port(5236) .databaseList("DMDB") .tableList("BACKGROUND_DATA.OPEN_API_APP") .schemaList("BACKGROUND_DATA") .username("SYSDBA") .password("123456") .startupOptions(StartupOptions.initial()) // 先用 initial 测试 .dmProperties(properties) .includeSchemaChanges(true) .deserializer(new JsonDebeziumDeserializationSchema()) .sliceSize(20) .scanNewlyAddedTableEnabled(true) .build(); Configuration configuration = new Configuration(); // 检查点文件 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration); env.enableCheckpointing(20 * 1000); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(6 * 1000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); env.setStateBackend(new FsStateBackend("file:///opt/flink/dmcdc/ck")); DataStream<String> sourceStream = env.fromSource( changeEventSource, WatermarkStrategy.noWatermarks(), "DmSource" ); // 处理CDC数据并生成JSON格式的SQL语句 DataStream<String> jsonStream = sourceStream.map(new MapFunction<String, String>() { private transient ObjectMapper objectMapper; private final JsonNodeFactory nodeFactory = JsonNodeFactory.instance; @Override public String map(String value) throws Exception { if (objectMapper == null) { objectMapper = new ObjectMapper(); } JsonNode rootNode = objectMapper.readTree(value); String op = rootNode.path("op").asText(); JsonNode sourceNode = rootNode.path("source"); String schema = sourceNode.path("schema").asText(); String table = sourceNode.path("table").asText(); long tsMs = rootNode.path("ts_ms").asLong(); // 创建JSON结构 ObjectNode json = nodeFactory.objectNode(); // 1. 添加metadata部分 ObjectNode metadata = json.putObject("metadata") .put("schema", schema) .put("table", table) .put("source_timestamp", tsMs); // 处理时间戳字段 JsonNode afterNode = rootNode.path("after"); if (!afterNode.isMissingNode() && afterNode.has("TIMESTAMP")) { long timestampNs = afterNode.path("TIMESTAMP").asLong(); metadata.put("event_time", formatTimestamp(timestampNs)); } else { metadata.put("event_time", formatTimestamp(tsMs * 1000000L)); } // 2. 根据操作类型生成SQL并添加到JSON String sql = ""; switch (op) { case "r": sql = generateInsertSQL(schema + "." + table, afterNode); json.putObject("sql").put("dml", sql); break; case "c": sql = generateInsertSQL(schema + "." + table, afterNode); json.putObject("sql").put("dml", sql); break; case "u": JsonNode beforeNode = rootNode.path("before"); sql = generateUpdateSQL(schema + "." + table, beforeNode, afterNode); json.putObject("sql").put("dml", sql); break; case "d": JsonNode beforeNodeDelete = rootNode.path("before"); sql = generateDeleteSQL(schema + "." + table, beforeNodeDelete); json.putObject("sql").put("dml", sql); break; default: json.put("error", "UNKNOWN OPERATION: " + op); } // 配置ObjectMapper禁用美化打印（默认即为紧凑格式） // ObjectMapper mapper = new ObjectMapper(); // String jsonString = mapper.writeValueAsString(json); return objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(json); } private String generateInsertSQL(String tableName, JsonNode afterNode) { StringBuilder fields = new StringBuilder(); StringBuilder values = new StringBuilder(); afterNode.fieldNames().forEachRemaining(fieldName -> { if (fields.length() > 0) { fields.append(", "); values.append(", "); } fields.append(fieldName); JsonNode valueNode = afterNode.get(fieldName); if (valueNode.isNull()) { values.append("NULL"); } else if (valueNode.isTextual()) { // 处理文本类型（含转义） values.append("'").append(escapeSQL(valueNode.asText())).append("'"); } else if (valueNode.isLong() && isTimestampField(fieldName)) { // 处理时间戳转换（纳秒/毫秒级） values.append("'").append(formatTimestamp(valueNode.asLong())).append("'"); } else if (valueNode.isNumber()) { // 处理其他数字类型 values.append(valueNode.asText()); } else { // 默认处理（如布尔值等） values.append(valueNode.asText()); } }); return String.format("INSERT INTO %s (%s) VALUES (%s)", tableName, fields.toString(), values.toString()); } private String generateUpdateSQL(String tableName, JsonNode beforeNode, JsonNode afterNode) { StringBuilder setClause = new StringBuilder(); StringBuilder whereClause = new StringBuilder(); // 构建SET部分 afterNode.fieldNames().forEachRemaining(fieldName -> { if (!fieldName.equals("ID")) { // 假设ID是主键，不更新 if (setClause.length() > 0) { setClause.append(", "); } JsonNode valueNode = afterNode.get(fieldName); if (valueNode.isNull()) { setClause.append(fieldName).append(" = NULL"); } else if (valueNode.isTextual()) { setClause.append(fieldName) .append(" = '") .append(escapeSQL(valueNode.asText())) .append("'"); } else if (valueNode.isLong() && isTimestampField(fieldName)) { // 处理时间戳转换（纳秒/毫秒级） setClause.append(fieldName) .append(" = '") .append(formatTimestamp(valueNode.asLong())) .append("'"); } else { setClause.append(fieldName) .append(" = ") .append(valueNode.asText()); } } }); // 构建WHERE部分(使用所有字段作为条件以确保准确性) beforeNode.fieldNames().forEachRemaining(fieldName -> { if (whereClause.length() > 0) { whereClause.append(" AND "); } JsonNode valueNode = beforeNode.get(fieldName); if (valueNode.isNull()) { whereClause.append(fieldName).append(" IS NULL"); } else if (valueNode.isTextual()) { whereClause.append(fieldName) .append(" = '") .append(escapeSQL(valueNode.asText())) .append("'"); } else if (valueNode.isLong() && isTimestampField(fieldName)) { // 处理时间戳转换（纳秒/毫秒级） whereClause.append(fieldName) .append(" = '") .append(formatTimestamp(valueNode.asLong())) .append("'"); } else { whereClause.append(fieldName) .append(" = ") .append(valueNode.asText()); } }); return String.format("UPDATE %s SET %s WHERE %s", tableName, setClause.toString(), whereClause.toString()); } private String generateDeleteSQL(String tableName, JsonNode beforeNode) { StringBuilder whereClause = new StringBuilder(); beforeNode.fieldNames().forEachRemaining(fieldName -> { if (whereClause.length() > 0) { whereClause.append(" AND "); } JsonNode valueNode = beforeNode.get(fieldName); if (valueNode.isNull()) { whereClause.append(fieldName).append(" IS NULL"); } else if (valueNode.isTextual()) { whereClause.append(fieldName) .append(" = '") .append(escapeSQL(valueNode.asText())) .append("'"); } else if (valueNode.isLong() && isTimestampField(fieldName)) { // 处理时间戳转换（纳秒/毫秒级） whereClause.append(fieldName) .append(" = '") .append(formatTimestamp(valueNode.asLong())) .append("'"); } else { whereClause.append(fieldName) .append(" = ") .append(valueNode.asText()); } }); return String.format("DELETE FROM %s WHERE %s", tableName, whereClause.toString()); } // 判断字段是否为时间戳字段（根据命名约定或业务逻辑） private boolean isTimestampField(String fieldName) { return fieldName.toLowerCase().contains("time") || fieldName.toLowerCase().contains("date") || fieldName.toLowerCase().contains("ts"); } // 格式化时间戳（支持纳秒/毫秒/秒级） private String formatTimestamp(long timestamp) { // 判断时间戳精度（假设大于1e16为纳秒，大于1e12为微秒，其余为毫秒/秒） if (timestamp > 1e16) { timestamp /= 1_000_000; // 纳秒转毫秒 } else if (timestamp > 1e12) { timestamp /= 1_000; // 微秒转毫秒 } else if (timestamp < 1e10) { timestamp *= 1_000; // 秒转毫秒 } // 使用Java 8时间API格式化（线程安全） return Instant.ofEpochMilli(timestamp) .atZone(ZoneId.systemDefault()) .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); } // SQL特殊字符转义（防止注入） private String escapeSQL(String input) { return input.replace("'", "''"); } }); // 创建Kafka Sink KafkaSink<String> kafkaSink = KafkaSink.<String>builder() .setBootstrapServers("192.168.20.220:9092") // Kafka broker地址 .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("dm-cdc-topic") // Kafka主题名称 .setValueSerializationSchema(new SimpleStringSchema()) // 使用字符串序列化 .build() ) .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE) // 精确一次交付 .setProperty("transaction.timeout.ms", "300000") // 设置为5分钟（需小于Broker的15分钟限制） .setProperty("acks", "all") // 确保高可靠性 .setProperty("schema.history.internal.kafka.topic", "dm-history-topic") .build(); // 构建数据处理管道 jsonStream.sinkTo(kafkaSink); // 将数据发送到Kafka env.execute(); } }正常情况下，你的数据会被写入到 kafka 中，如果日志中提示错误的话，可以尝试手动在 kafka 中先建好主题：dm-cdc-topic借助 idea 中的 kafka 插件能够查看到主题中的数据。从Kafka读取记录并通过 JDBC 写入数据库此部分比较灵活，可以根据实际的需要来接收 kafka 的数据并进行处理。博主目前是采用 idea 中的 kafka 插件创建一个 consumer 来验证消费数据的。后续步骤本地调试通过后，就可以打包出 jar 文件并上传到 Flink 中进行线上测试了，或者可以通过命令行的方式运行（将打包出来的 jar 文件上传到 docker-compose.yaml 同级的 jars 目录下后执行）：docker-compose exec jobmanager flink run -d -c org.bg.dm.cdc.DmCdcApplication /opt/flink/usrlib/DmCdc-0.0.1-SNAPSHOT.jar参考地址【1】DM8 Flink CDC 连接器工具【2】FlinkCDC 达梦数据库实时同步【3】Flink 架构
Linux中使用Chrony同步时间Chrony是什么?Chrony 是网络时间协议(NTP)的一种实现。可以使用 Chrony:将系统时钟与 NTP 服务器同步，将系统时钟与参考时钟同步，如ASCII 接收器.要将系统时钟与手动时间输入同步，作为 NTPv4(RFC 5905) 服务器或对等服务器，为网络中的其他计算机提供时间服务。简而言之，Chrony是一个NTP协议的实现，能够同步其他NTP服务器的时间，也可以作为NTP服务器供其他服务器同步时间。Chrony怎么用？假设你的所有服务器都在 10.0.0.0/8 网段中,且此网段中所有服务器之间的网络策略是互通的。此时你可以将一台服务器作为Chrony服务器让其他服务器同步它的时间。例如，我在 10.0.0.0/8 网段中选择 10.10.2.101作为Chrony服务器，然后允许本网段中其他服务器来同步时间，那么配置如下：服务端：备份并修改chrony配置文件cp /etc/chrony.conf /etc/chrony.conf_bak vim /etc/chrony.conf文件内容：# 手动控制时间源（用于手工调整） manual # 修改 chrony 对外服务端口（默认 123 → 改为 11123） # 可以不修改，也可以修改，看需求 port 11123 # 允许 10 段网段访问 allow 10.0.0.0/8 # driftfile 和日志 driftfile /var/lib/chrony/drift logdir /var/log/chrony # 启用本地时间源，即使没有外部同步源也能提供服务 local stratum 10 # 外部同步源，如果有外部时间源，chrony服务器也可以通过外部时间来更新自己的时间，没有的话就注释 # 如果服务器通互联网，也可以配置互联网上的 ntp 服务器地址 #server 10.15.163.89 iburst修改完成后保存退出，重启chrony服务即可：systemctl restart chronyd客户端：备份并修改chrony配置文件cp /etc/chrony.conf /etc/chrony.conf_bak vim /etc/chrony.conf文件内容：server 10.10.2.101 port 11123 iburst driftfile /var/lib/chrony/drift makestep 1.0 3 rtcsync logdir /var/log/chrony修改完成后保存退出，重启chrony服务即可：systemctl restart chronyd配置成功后可以通过这条指令来查看当前同步状态chronyc tracking输出示例：[root@server1 ~]# chronyc tracking Reference ID : 0A055046 (10.10.2.101) Stratum : 11 Ref time (UTC) : Wed Nov 19 02:11:31 2025 System time : 0.000006415 seconds slow of NTP time Last offset : -0.000008004 seconds RMS offset : 0.000010776 seconds Frequency : 3.412 ppm fast Residual freq : -0.001 ppm Skew : 0.013 ppm Root delay : 0.000045834 seconds Root dispersion : 0.000062100 seconds Update interval : 260.9 seconds Leap status : Normal其中查看最后一行的 Leap status ，如果是Normal则证明与 Reference ID中的IP同步成功，否则代表不成功。使用下面这条指令可以手动同步一次时间chronyc makestep执行以下指令会打印出chrony正在访问的时间源的信息chronyc sources -v输出示例：[root@server1 ~]# chronyc sources -v 210 Number of sources = 1 .-- Source mode '^' = server, '=' = peer, '#' = local clock. / .- Source state '*' = current synced, '+' = combined , '-' = not combined, | / '?' = unreachable, 'x' = time may be in error, '~' = time too variable. || .- xxxx [ yyyy ] +/- zzzz || Reachability register (octal) -. | xxxx = adjusted offset, || Log2(Polling interval) --. | | yyyy = measured offset, || \ | | zzzz = estimated error. || | | \ MS Name/IP address Stratum Poll Reach LastRx Last sample =============================================================================== ^* 10.10.2.101 10 8 377 170 -5891ns[ -14us] +/- 23us其中最后列出的 IP 之前 如果显示 ^*则代表与当前时间源正常同步，如果显示^?则代表未正常同步。参考链接：银河麒麟Linux同步时间：NTPd、Chrony、systemd-timesyncd 配置与使用 使用 chrony 套件配置 NTP
docker 部署nacos 并配置达梦数据库作为数据源
背景国产化的推行，项目开始使用达梦数据库，使用Nacos作为配置中心，需要对数据进行持久化存储。此处使用docker-compose部署nacos。参考文章Nacos适配达梦数据库(不修改源码)，基于Docker-compose_nacos 达梦数据库Nacos数据库适配插件实现步骤第一步：部署一个没有使用外部数据源的nacos此处部署的是 nacos 2.3.2版本，docker-compose配置文件内容如下，运行启动后默认账号密码都是nacosversion: "3" services: nacos: image: nacos/nacos-server:v2.3.2-slim container_name: nacos environment: - PREFER_HOST_MODE=hostname - MODE=standalone - NACOS_AUTH_ENABLE=true - NACOS_AUTH_IDENTITY_KEY=serverIdentity - NACOS_AUTH_IDENTITY_VALUE=security - NACOS_AUTH_TOKEN=ZDFhMTUyMDItYjAyZi00ZTM3LTkzYzctNmQxNTIyOTUyNzQ1 volumes: - ./standalone-logs/:/home/nacos/logs ports: - 8848:8848 - 9848:9848部署完成之后，使用docker exec -it nacos /bin/bash命令进入到容器中，执行ll可以看到当前目录下的文件信息如下图nacos容器目录结构进入conf目录下可以看到nacos启动的配置文件： application.properties第二步：拷贝出nacos启动的配置文件进行修改确定容器中application.properties配置文件的路径之后，退出容器，使用docker cp命令将文件从容器中拷贝出来：docker cp nacos:/home/nacos/conf/application.properties ./conf/application.properties使用vim编辑器打开application.properties进行修改，修改后的文件内容如下：# spring server.servlet.contextPath=${SERVER_SERVLET_CONTEXTPATH:/nacos} server.contextPath=/nacos server.port=${NACOS_APPLICATION_PORT:8848} server.tomcat.accesslog.max-days=30 server.tomcat.accesslog.pattern=%h %l %u %t "%r" %s %b %D %{User-Agent}i %{Request-Source}i server.tomcat.accesslog.enabled=${TOMCAT_ACCESSLOG_ENABLED:false} server.error.include-message=ALWAYS # default current work dir server.tomcat.basedir=file:. #*************** Config Module Related Configurations ***************# ### Deprecated configuration property, it is recommended to use `spring.sql.init.platform` replaced. #spring.datasource.platform=${SPRING_DATASOURCE_PLATFORM:} #spring.sql.init.platform=${SPRING_DATASOURCE_PLATFORM:} spring.sql.init.platform=mysql nacos.cmdb.dumpTaskInterval=3600 nacos.cmdb.eventTaskInterval=10 nacos.cmdb.labelTaskInterval=300 nacos.cmdb.loadDataAtStart=false #db.num=${MYSQL_DATABASE_NUM:1} #db.url.0=jdbc:mysql://${MYSQL_SERVICE_HOST}:${MYSQL_SERVICE_PORT:3306}/${MYSQL_SERVICE_DB_NAME}?${MYSQL_SERVICE_DB_PARAM:characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useSSL=false} #db.user.0=${MYSQL_SERVICE_USER} #db.password.0=${MYSQL_SERVICE_PASSWORD} db.num=1 db.url.0=jdbc:dm://192.168.20.222:5237?schema=NACOS&characterEncoding=UTF-8&useUnicode=true&serverTimezone=Asia/Shanghai db.user.0=NACOS db.password.0=wydm@nacos ## DB connection pool settings db.pool.config.connectionTimeout=${DB_POOL_CONNECTION_TIMEOUT:30000} db.pool.config.validationTimeout=10000 db.pool.config.maximumPoolSize=20 db.pool.config.minimumIdle=2 db.pool.config.driver-class-name=dm.jdbc.driver.DmDriver ### The auth system to use, currently only 'nacos' and 'ldap' is supported: nacos.core.auth.system.type=${NACOS_AUTH_SYSTEM_TYPE:nacos} ### worked when nacos.core.auth.system.type=nacos ### The token expiration in seconds: nacos.core.auth.plugin.nacos.token.expire.seconds=${NACOS_AUTH_TOKEN_EXPIRE_SECONDS:18000} ### The default token: nacos.core.auth.plugin.nacos.token.secret.key=${NACOS_AUTH_TOKEN:} ### Turn on/off caching of auth information. By turning on this switch, the update of auth information would have a 15 seconds delay. nacos.core.auth.caching.enabled=${NACOS_AUTH_CACHE_ENABLE:false} nacos.core.auth.enable.userAgentAuthWhite=${NACOS_AUTH_USER_AGENT_AUTH_WHITE_ENABLE:false} nacos.core.auth.server.identity.key=${NACOS_AUTH_IDENTITY_KEY:} nacos.core.auth.server.identity.value=${NACOS_AUTH_IDENTITY_VALUE:} ## spring security config ### turn off security nacos.security.ignore.urls=${NACOS_SECURITY_IGNORE_URLS:/,/error,/**/*.css,/**/*.js,/**/*.html,/**/*.map,/**/*.svg,/**/*.png,/**/*.ico,/console-fe/public/**,/v1/auth/**,/v1/console/health/**,/actuator/**,/v1/console/server/**} # metrics for elastic search management.metrics.export.elastic.enabled=false management.metrics.export.influx.enabled=false nacos.naming.distro.taskDispatchThreadCount=10 nacos.naming.distro.taskDispatchPeriod=200 nacos.naming.distro.batchSyncKeyCount=1000 nacos.naming.distro.initDataRatio=0.9 nacos.naming.distro.syncRetryDelay=5000 nacos.naming.data.warmup=true nacos.console.ui.enabled=true nacos.core.param.check.enabled=true主要修改了这部分：application.properties修改部分注意：虽然我们配置的是达梦的数据库连接信息，但是数据源平台配置成了mysql，为啥呢，因为官方文档里面说的配置成dm并不生效，会提示找不到数据源dm。以下图片来源于参考文章的第二篇（nacos官方）。nacos官方配置达梦说明文档第三步：将达梦数据库驱动程序映射进入docker在docker-compose.yaml同级目录下创建一个plugins目录，把达梦数据库的驱动程序放入这个目录下，然后修改docker-compose.yaml文件， 把修改后的application.properties文件和达梦数据库的驱动映射到容器中。最终修改后的docker-compose.yaml文件内容如下：version: "3" services: nacos: image: nacos/nacos-server:v2.3.2-slim container_name: nacos environment: - PREFER_HOST_MODE=hostname - MODE=standalone - NACOS_AUTH_ENABLE=true - NACOS_AUTH_IDENTITY_KEY=serverIdentity - NACOS_AUTH_IDENTITY_VALUE=security - NACOS_AUTH_TOKEN=ZDFhMTUyMDItYjAyZi00ZTM3LTkzYzctNmQxNTIyOTUyNzQ1 volumes: - ./standalone-logs/:/home/nacos/logs - ./conf/application.properties:/home/nacos/conf/application.properties - ./plugins/:/home/nacos/plugins/ ports: - 8848:8848 - 9848:9848docker-compose.yaml文件所在目录结构如下图所示：配置完成后的docker-compose.yaml同级目录第四步：启动容器，验证启动容器，访问页面，随便创建几个用户，看数据库里用户表有没有变化，有则配置成功。总结三步走：修改application.properties配置文件；映射驱动；启动验证。过程中的难点在于配置application.properties的时候有个坑，就是官方文档写的dm不生效，改成了mysql解决。
asyncTool是个啥官方仓库这样说：解决任意的多线程并行、串行、阻塞、依赖、回调的并行框架，可以任意组合各线程的执行顺序，带全链路执行结果回调。多线程编排一站式解决方案。来自于京东主App后台怎么用1、引入Maven依赖<repository> <id>jitpack.io</id> <url>https://jitpack.io</url> </repository><dependency> <groupId>com.gitee.jd-platform-opensource</groupId> <artifactId>asyncTool</artifactId> <version>V1.4-SNAPSHOT</version> </dependency>2、自定义任务类IWorker<Input, Output>是一个泛型接口，其中：Input代表入参类型，Output代表返回值类型下面的示例中，入参类型为fastjson包中的JSONObject类型，返回值类型为String字符串类型@Slf4j public class JobWorker implements IWorker<JSONObject, String>, ICallback<JSONObject, String> { @Override public void begin() { log.info("任务开始执行"); } @Override public void result(boolean b, JSONObject params, WorkResult<String> workResult) { log.info("任务执行完毕, 成功:{}, 入参:{}, 输出结果:{}", b, params, workResult); } @Override public String action(JSONObject params, Map<String, WorkerWrapper> map) { // 这里是要执行的耗时操作 return ""; } @Override public String defaultValue() { return "自定义默认值"; } }3、任务编排执行 JobWorker w = new JobWorker(); WorkerWrapper<JSONObject, String> workerWrapper = new WorkerWrapper .Builder<JSONObject, String>() .worker(w) .callback(w) .param(param) .build(); try { Async.beginWork(60000, workerWrapper); } catch (ExecutionException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } WorkResult<String> workResult = workerWrapper.getWorkResult(); log.info("result: {}", workResult);参考文章官方仓库
需求在某台设备上，安装有Centos7操作系统，此设备本身拥有6个串口，但centos默认只识别了4个串口，但某些数据被输出到了串口6上，所以需要识别出另外两个串口信息，以能够接收使用数据。修改步骤查看当前串口信息使用如下指令可以查看当前操作系统识别到的串口信息cat /proc/tty/driver/serial在centos下默认只有四个串口，串口信息展示如下图所示：默认串口信息编辑/etc/default/grub文件在我的设备上实际有6个串口，但是默认只识别了4个，现在在配置文件中进行配置：在文件的GRUB_CMDLINE_LINUX这一行末尾添加 8250.nr_uarts=6, 这里的数字6就是你的设备实际的串口数量，例如实际是10个串口，那就改为10，修改后的文件示例如下图所示：添加配置信息后的示例重新编译grub使用如下指令重新编译grub：grub2-mkconfig -o /boot/grub2/grub.cfg 注意如果是UEFI 系统上则需要执行命令grub2-mkconfig -o /boot/efi/EFI/centos/grub.cfg如何查看是不是UEFI 你可以在启动的时候进入bios里面查看一下是不是设置的UEFI都不会的可以两条都执行一下，执行完成后重启操作系统。效果重新查看串口信息重启完成之后，再次执行查看串口信息的命令：cat /proc/tty/driver/serial可以看到输出的信息中多出了4、5两条记录，总共6条。最终效果至此配置完成。参考文章centos下 新添加串口方法
字典树定义字典树，顾名思义，是关于“字典”的一棵树。即：它是对于字典的一种存储方式（所以是一种数据结构而不是算法）。这个词典中的每个“单词”就是从根节点出发一直到某一个目标节点的路径，路径中每条边的字母连起来就是一个单词。结构实例trie1字典树的功能根据字典树的概念，我们可以发现：字典树的本质是把很多字符串拆成单个字符的形式，以树的方式存储起来。所以我们说字典树维护的是”字典“。那么根据这个最基本的性质，我们可以由此延伸出字典树的很多妙用。简单总结起来大体如下：1、维护字符串集合（即字典）。2、向字符串集合中插入字符串（即建树）。3、查询字符串集合中是否有某个字符串（即查询）。4、统计字符串在集合中出现的个数（即统计）。5、将字符串集合按字典序排序（即字典序排序）。6、求集合内两个字符串的LCP（Longest Common Prefix，最长公共前缀）（即求最长公共前缀）。我们可以发现，以上列举出的功能都是建立在“字符串集合”的基础上的。再一次强调，字典树是“字典”的树，一切功能都是“字典”的功能。这也为我们使用字典树的时候提供了一个准则：看到一大堆字符串同时出现，就往哈希和Trie树那边想一下。基于字典树实现的域名匹配有一个需求，给出一个域名名单，当输入一个新域名之后判断其是否存在域名名单中，要求支持通配符匹配代码实现根据字典树结构，使用go语言实现如下：package main import ( "fmt" "github.com/gookit/goutil/arrutil" "github.com/gookit/goutil/maputil" "strings" ) type TrieNode struct { children map[string]*TrieNode isEnd bool } type Trie struct { root *TrieNode } func initTrie() *Trie { return &Trie{root: &TrieNode{children: make(map[string]*TrieNode)}} } func (t *Trie) insertDomain(domain string) { parts := strings.Split(domain, ".") arrutil.Reverse(parts) node := t.root for _, part := range parts { if node.children[part] == nil { node.children[part] = &TrieNode{children: make(map[string]*TrieNode)} } node = node.children[part] } node.isEnd = true } func (t *Trie) searchDomain(domain string) bool { parts := strings.Split(domain, ".") arrutil.Reverse(parts) node := t.root for _, part := range parts { isNil := node.children[part] == nil if isNil && !maputil.HasKey(node.children, "*") { return false } if isNil { node = node.children["*"] } else { node = node.children[part] } } return node != nil && node.isEnd } func main() { trie := initTrie() // 示例域名 domains := []string{"wx.*.com", "example.com", "test.org"} // 插入域名到 Trie 中 for _, domain := range domains { trie.insertDomain(domain) } // 搜索示例 searchDomains := []string{"wx.qq.com", "openai.com", "test.org", "wx.qq1.com"} // 搜索域名是否存在于 Trie 中 for _, domain := range searchDomains { if trie.searchDomain(domain) { fmt.Printf("%s 存在\n", domain) } else { fmt.Printf("%s 不存在\n", domain) } } }参考文档字典树(Trie)详解 - Seaway-Fu - 博客园 (cnblogs.com)字典树 (Trie) - OI Wiki (oi-wiki.org)
本文来自转载，补充了函数参考文档地址，调整了代码格式原文地址0x00 文件操作函数ZwDeleteFile 删除指定的文件 ZwOpenFile 打开现有文件、目录、设备或卷 ZwClose 关闭对象句柄ZwQueryInformationFile 返回有关文件对象的各种信息ZwReadFile 从打开的文件读取数据ZwWriteFile 将数据写入打开的文件ZwCreateFile 创建新文件或打开现有文件调用Zw函数会进行参数检查,检查完成后再调用nt函数代码示例#include <Ntifs.h> #include <wdm.h> //文件删除 NTSTATUS FileDelete(PWSTR wFilePath) { NTSTATUS status = STATUS_SUCCESS; UNICODE_STRING uFilePath = { 0 }; RtlInitUnicodeString(&uFilePath, wFilePath); OBJECT_ATTRIBUTES objectAttributes = { 0 }; InitializeObjectAttributes(&objectAttributes, &uFilePath, OBJ_CASE_INSENSITIVE, NULL, NULL); status = ZwDeleteFile(&objectAttributes); if (status == STATUS_SUCCESS) DbgPrint("FileDelete -> DeleteFile success"); else DbgPrint("FileDelete -> DeleteFile failed"); return status; } //文件复制 NTSTATUS FileCopy(PWSTR wDestinationPath, PWSTR wSourcePath) { NTSTATUS status = STATUS_SUCCESS; //打开源文件 HANDLE hSourceFile = NULL; //句柄 - 参数1 UNICODE_STRING uSourcePath = { 0 }; //路径名 RtlInitUnicodeString(&uSourcePath, wSourcePath); OBJECT_ATTRIBUTES oaSourceFile = { 0 }; //对象属性,包含路径名 - 参数3 InitializeObjectAttributes(&oaSourceFile, &uSourcePath, OBJ_CASE_INSENSITIVE | OBJ_KERNEL_HANDLE, NULL, NULL); IO_STATUS_BLOCK isbSourceFile = { 0 }; //IO_STATUS_BLOCK -> 接收完成状态 -> 参数4 status = ZwOpenFile(&hSourceFile, GENERIC_READ, &oaSourceFile, &isbSourceFile, FILE_SHARE_READ | FILE_SHARE_WRITE, FILE_SYNCHRONOUS_IO_NONALERT); if (status != STATUS_SUCCESS) { //没有需要释放的资源 DbgPrint("FileCopy -> ZwOpenFile failed"); return status; } //获取文件大小 IO_STATUS_BLOCK isbQuery = { 0 }; //IO_STATUS_BLOCK -> 接收完成状态 FILE_STANDARD_INFORMATION FileInformation = { 0 }; status = ZwQueryInformationFile(hSourceFile, &isbQuery, &FileInformation, sizeof(FileInformation), FileStandardInformation); if (status != STATUS_SUCCESS) { //需要释放资源 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy -> ZwQueryInformationFile failed"); return status; } if (FileInformation.EndOfFile.QuadPart == 0) { //需要释放资源 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy -> ZwQueryInformationFile failed -> source file size is %Id ", FileInformation.EndOfFile.QuadPart); return STATUS_NO_SUCH_FILE; } LONGLONG sourceRemainFileSize = FileInformation.EndOfFile.QuadPart; //源文件大小 //创建目标文件 HANDLE hDestinationFile = NULL; //句柄 - 参数1 UNICODE_STRING uDestinationPath = { 0 }; //路径名 RtlInitUnicodeString(&uDestinationPath, wDestinationPath); OBJECT_ATTRIBUTES oaDestinationFile = { 0 }; //对象属性,包含路径名 - 参数3 InitializeObjectAttributes(&oaDestinationFile, &uDestinationPath, OBJ_CASE_INSENSITIVE | OBJ_KERNEL_HANDLE, NULL, NULL); IO_STATUS_BLOCK isbDestinationFile = { 0 }; //IO_STATUS_BLOCK -> 接收完成状态 -> 参数4 status = ZwCreateFile(&hDestinationFile, GENERIC_WRITE, &oaDestinationFile, &isbDestinationFile, NULL, FILE_ATTRIBUTE_NORMAL, NULL, FILE_SUPERSEDE, FILE_SYNCHRONOUS_IO_NONALERT, NULL, NULL); if (status != STATUS_SUCCESS) { //需要释放资源 ZwClose(hSourceFile); //关闭 DbgPrint("FileCopy -> ZwOpenFile failed"); return status; } //通过缓冲区复制内容 //1 申请缓冲区 PVOID buffer = ExAllocatePoolWithTag(NonPagedPool, 1024, '1agT'); if (!buffer) { //需要释放资源 ZwClose(hDestinationFile); //关闭目标文件 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy -> ExAllocatePoolWithTag failed"); return STATUS_NO_MEMORY; } IO_STATUS_BLOCK isbRead = { 0 }; //IO_STATUS_BLOCK -> 接收完成状态 -> ZwReadFile参数5 ULONG length = 0; //读取长度 ZwReadFile参数7 LARGE_INTEGER offset = { 0 }; //读取偏移 -> 从文件那个位置开始读 ZwReadFile参数8 while (sourceRemainFileSize) { if (sourceRemainFileSize > 1024) length = 1024; else length = sourceRemainFileSize; //DbgPrint("FileCopy -> while -> ZwReadFile"); //2 读取源文件数据至缓冲区 status = ZwReadFile(hSourceFile, NULL, NULL, NULL, &isbRead, buffer, length, &offset, NULL); if (status != STATUS_SUCCESS) { //需要释放资源 ExFreePool(buffer); //释放缓冲区 ZwClose(hDestinationFile); //关闭目标文件 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy -> ZwReadFile failed"); return status; } //DbgPrint("FileCopy -> while -> ZwWriteFile"); //3 将缓冲区数据写入目标文件 status = ZwWriteFile(hDestinationFile, NULL, NULL, NULL, &isbRead, buffer, length, &offset, NULL); if (status != STATUS_SUCCESS) { //需要释放资源 ExFreePool(buffer); //释放缓冲区 ZwClose(hDestinationFile); //关闭目标文件 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy -> ZwWriteFile failed"); return status; } sourceRemainFileSize -= length; offset.QuadPart += length; } //关闭资源 ExFreePool(buffer); //释放缓冲区 ZwClose(hDestinationFile); //关闭目标文件 ZwClose(hSourceFile); //关闭源文件 DbgPrint("FileCopy success"); return status; } VOID DriverUnload(PDRIVER_OBJECT DriverObject) { DbgPrint("DriverUnload"); } NTSTATUS DriverEntry(PDRIVER_OBJECT DriverObject, PUNICODE_STRING RegistryPath) { NTSTATUS status = STATUS_SUCCESS; DriverObject->DriverUnload = DriverUnload; FileCopy(L"\\??\\C:\\2.exe", L"\\??\\C:\\1.exe"); //FileDelete(L"\\??\\C:\\1.exe"); return status; }
下午五点多，一时兴起，约上了两位挚友，来了一场说走就走的郊游，刚出门就雷声大作，刚到半路就已经是瓢泼大雨了，三个人都表示无所谓，干就完了。于是乎驱车到达山下，到了山下发现这边还没开始下雨，未作停留开始爬山，其中乐趣，在于许久未见的老友依然无话不说。三个段子手聚到了一起，一路上山都是笑声不断。快到山顶时，小雨便开始淅淅沥沥的落了下来，继续上行，雨也开始愈发的大了起来，终于一气呵成登顶，一览城市的夜景，顺手拍了几张照片纪念一下，奈何座机画质有点惨不忍睹！大帅比大帅比长虫山长虫山
有些信息需要留存，又找不到合适的地方，那就把它留到这里吧顺便分享一首歌 [binary]七月以来，我的生活里发生了一些事情。扰的我甚是烦恼。我是一个厌恶孤独的人，讨厌一个人孤独的呆着的感觉，一切的问题似乎也因此而起。讲讲我的过往吧。大学的时候，成天和室友们一起从早到晚打游戏度日，似乎也算过的“充实”，后来到了假期，室友们都回家了，我留下了做兼职，每天去做家教，下班回到宿舍，一个人孤零零的，呆着特难受，一个人连打游戏都失去了乐趣，有时候想找个人聊聊天，看着空荡荡的宿舍也没有谁能和自己搭一句话。总结起来就是孤独。为了每天做完兼职回到宿舍以后的能稍微不那么孤独，我总是想方设法的弄出些声响，假装热闹，那段时间我回到宿舍就是打开电脑一直网易云，后面网易云也听腻了，干脆买了一个人工智障——天猫精灵。这个小东西后面我回老家就被我带回去，不过说起来，它也算是陪伴了我一个假期，就是那种你一叫它，它就会回应你，你问它啥它也会回应你，即便回答在牛头不对马嘴，但是它会回复你。时间一晃，在学校的时光已经结束了，工作两年之后，有同学买车了，有同学买房了，反观自己一无所有的时候，难免会有些难过。人的一生似乎都是在围绕着RMB转圈，都是为了RMB而活一样。有时候就会觉得很烦躁，想放空自己，想出趟远门，漫无目的的四处漂流。最近这段时间，经人提点，想通了，原来一直是自己在给自己施加压力，一下子被戳中了一样，开始试着以一个躺平的想法去度日，感觉轻松了许多。嗯，挺好，一身轻松。奈何手上还有些事情需要处理，承诺了的东西不能一拖再拖，等到把这些杂事处理干净，兴许就更轻松了。人们似乎就是喜欢把自己脆弱的一面隐藏起来，把自己伪装成坚不可摧的样子，或者说有时候是不得不变成这样。嗯，至少我认识的很多人都是这样。互联网的发展让我们每个人在各种社交媒体上留下了大量的痕迹，有时候，我们会尝试把这写痕迹抹去，希望把自己藏起来，让人看不到真实的自己甚至看不到自己，或许这是人与生俱来的对自己的保护机制吧哈哈哈。前段时间，因为一些事情，心里充满了愧疚感，最近算是调节了差不多了，只想过的快乐一点，每天一个人安静的在网易云听听歌，每天大多数时候都还是能够不忧伤的哈哈哈。最后吐槽一件事，就上个周，对币圈有一点兴趣，于是找了一些资源准备挖狗狗币，结果损失惨重，一个周，损失1000U，妈了个巴子，好在心态好，安慰自己说无所吊谓，吃一堑长一智吧。emmm，洋洋洒洒的写了一堆，也没个啥中心思想，无所吊谓了，主打一个无人观看自由发挥。博客写了三年多了，第一次写一篇个人生活的记录，就这样吧，以后这里就是一半博客一半吐槽了！