Skip to content

Commit 71be3e2

Browse files
xlornexlorne
committed
support JdbcQuery.java
1 parent ce2f88a commit 71be3e2

18 files changed

Lines changed: 208 additions & 59 deletions

‎README.md‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -75,13 +75,12 @@ spring.datasource.password=12345678
7575

7676
import com.codingapi.dbstream.event.DBEvent;
7777
import com.codingapi.dbstream.event.DBEventPusher;
78+
import com.codingapi.dbstream.query.JdbcQuery;
7879

7980
// 在应用启动时注册事件推送器（如 @PostConstruct、@Configuration 等）
80-
DBStreamContext.getInstance().
81-
82-
addEventPusher(new DBEventPusher() {
81+
DBStreamContext.getInstance().addEventPusher(new DBEventPusher() {
8382
@Override
84-
public void push (List < DBEvent > events) {
83+
public void push(JdbcQuery jdbcQuery,List<DBEvent> events){
8584
// 处理数据库变更事件
8685
for (DBEvent event : events) {
8786
System.out.println("表名: " + event.getTableName());
@@ -135,7 +134,7 @@ DBStreamContext.getInstance().addListener(new MySQLListener());
135134

136135
### 5. 通过设置DBTableSupportProvider订阅对那些表进行监听（可选）
137136

138-
```
137+
```java
139138

140139
import com.codingapi.dbstream.scanner.DbTable;
141140

@@ -273,6 +272,8 @@ mvn clean test -P travis
273272

274273
3. **使用场景限制**：
275274
- 数据库表必须有主键的定义，在DELETE事件需要明确主键信息，主键物理表不存在时可通过外部key文件配置的方式添加。
275+
- 注意：INSERT语句在INSERT INTO SELECT非手动指定ID字段，或者INSERT中传入函数等形式可能导致无法获取到数据ID的情况。
276+
- INSERT语句中VALUES不支持函数模式传递数据，建议通过指定参数的方式进行插入数据，若使用了函数无法获取到真实数据时，可自行通过ID进行二次查询获取落库数据,使用`jdbcQuery`对象。
276277
- 若INSERT INTO SELECT 语句中，采用主键自增模式，受限于JDBC的支持将无法解析到自增ID，建议修改单条保存或修改ID为手动传递。
277278

278279
4. **元数据缓存**：

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
<modelVersion>4.0.0</modelVersion>
55
<groupId>com.codingapi.dbstream</groupId>
66
<artifactId>dbstream-driver</artifactId>
7-
<version>1.0.9</version>
7+
<version>1.0.10</version>
88

99
<url>https://github.com/codingapi/dbstream-driver</url>
1010
<name>dbstream-driver</name>

‎src/main/java/com/codingapi/dbstream/event/DBEventContext.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.codingapi.dbstream.event;
22

3+
import com.codingapi.dbstream.query.JdbcQuery;
34
import lombok.Getter;
45

56
import java.util.List;
@@ -21,7 +22,7 @@ private DBEventContext() {
2122

2223
}
2324

24-
void push(List<DBEvent> events) {
25+
void push(JdbcQuery jdbcQuery, List<DBEvent> events) {
2526
if (events == null || events.isEmpty()) {
2627
return;
2728
}
@@ -30,10 +31,10 @@ void push(List<DBEvent> events) {
3031
});
3132
if (!this.pushers.isEmpty()) {
3233
for (DBEventPusher pusher : pushers) {
33-
pusher.push(events);
34+
pusher.push(jdbcQuery,events);
3435
}
3536
} else {
36-
defaultDBEventPusher.push(events);
37+
defaultDBEventPusher.push(jdbcQuery,events);
3738
}
3839
}
3940

‎src/main/java/com/codingapi/dbstream/event/DBEventPusher.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package com.codingapi.dbstream.event;
22

3+
import com.codingapi.dbstream.query.JdbcQuery;
4+
35
import java.util.List;
46

57
/**
@@ -9,7 +11,9 @@ public interface DBEventPusher {
911

1012
/**
1113
* 推送事件
14+
* @param jdbcQuery JDBC数据查询对象
15+
* @param events DB事件消息
1216
*/
13-
void push(List<DBEvent> events);
17+
void push(JdbcQuery jdbcQuery,List<DBEvent> events);
1418

1519
}

‎src/main/java/com/codingapi/dbstream/event/DefaultDBEventPusher.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package com.codingapi.dbstream.event;
22

3+
import com.codingapi.dbstream.query.JdbcQuery;
4+
35
import java.util.List;
46

57
/**
@@ -9,7 +11,7 @@
911
public class DefaultDBEventPusher implements DBEventPusher {
1012

1113
@Override
12-
public void push(List<DBEvent> events) {
14+
public void push(JdbcQuery jdbcQuery,List<DBEvent> events) {
1315
System.out.println("<=== DBStream DBEvent Total " + events.size() + " ===> ");
1416
for (DBEvent event : events) {
1517
System.out.println(event);

‎src/main/java/com/codingapi/dbstream/event/TransactionEventPools.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.codingapi.dbstream.event;
22

3+
import com.codingapi.dbstream.query.JdbcQuery;
34
import lombok.Getter;
45

56
import java.util.ArrayList;
@@ -39,7 +40,7 @@ public boolean isAutoCommit() {
3940
/**
4041
* 添加事务事件
4142
*/
42-
public void addEvents(String transactionKey, List<DBEvent> events) {
43+
public void addEvents(JdbcQuery jdbcQuery,String transactionKey, List<DBEvent> events) {
4344
List<DBEvent> currentEvents = pools.get();
4445
if (currentEvents == null) {
4546
currentEvents = new ArrayList<>();
@@ -48,23 +49,24 @@ public void addEvents(String transactionKey, List<DBEvent> events) {
4849
currentEvents.addAll(events);
4950

5051
if (this.isAutoCommit()) {
51-
this.commitEvents(transactionKey);
52+
this.commitEvents(jdbcQuery,transactionKey);
5253
}
5354

5455
}
5556

5657
/**
5758
* 提交消息
5859
*
60+
* @param jdbcQuery 数据查询对象
5961
* @param transactionKey 事务标示
6062
*/
61-
public void commitEvents(String transactionKey) {
63+
public void commitEvents(JdbcQuery jdbcQuery, String transactionKey) {
6264
List<DBEvent> currentEvents = pools.get();
6365
if (currentEvents != null && !currentEvents.isEmpty()) {
6466
currentEvents.forEach(dbEvent -> {
6567
dbEvent.setTransactionKey(transactionKey);
6668
});
67-
DBEventContext.getInstance().push(currentEvents);
69+
DBEventContext.getInstance().push(jdbcQuery,currentEvents);
6870
currentEvents.clear();
6971
}
7072
this.clear();
@@ -73,9 +75,10 @@ public void commitEvents(String transactionKey) {
7375
/**
7476
* 回滚消息
7577
*
78+
* @param jdbcQuery 数据查询对象
7679
* @param transactionKey 事务标示
7780
*/
78-
public void rollbackEvents(String transactionKey) {
81+
public void rollbackEvents(JdbcQuery jdbcQuery,String transactionKey) {
7982
this.clear();
8083
}
8184

‎src/main/java/com/codingapi/dbstream/interceptor/SQLExecuteState.java‎

Lines changed: 15 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,18 @@
11
package com.codingapi.dbstream.interceptor;
22

33
import com.codingapi.dbstream.proxy.ConnectionProxy;
4+
import com.codingapi.dbstream.query.JdbcQuery;
45
import com.codingapi.dbstream.scanner.DBMetaData;
56
import com.codingapi.dbstream.scanner.DBScanner;
67
import com.codingapi.dbstream.scanner.DbColumn;
78
import com.codingapi.dbstream.scanner.DbTable;
89
import lombok.Getter;
910
import lombok.Setter;
1011

11-
import java.sql.*;
12+
import java.sql.ResultSet;
13+
import java.sql.ResultSetMetaData;
14+
import java.sql.SQLException;
15+
import java.sql.Statement;
1216
import java.util.*;
1317

1418
/**
@@ -76,11 +80,18 @@ public class SQLExecuteState {
7680
@Getter
7781
private long afterTimestamp;
7882

83+
/**
84+
* JDBC数据查询
85+
*/
86+
@Getter
87+
private final JdbcQuery jdbcQuery;
88+
7989

8090
public SQLExecuteState(String sql, ConnectionProxy connectionProxy, Statement statement, DBMetaData metaData) {
8191
this.sql = sql;
8292
this.connectionProxy = connectionProxy;
8393
this.statement = statement;
94+
this.jdbcQuery = new JdbcQuery(connectionProxy);
8495
this.metaData = metaData;
8596
this.sqlExecuteParams = new ArrayList<>();
8697

@@ -255,7 +266,7 @@ public String getTransactionKey() {
255266
* @throws SQLException 查询异常
256267
*/
257268
public List<Map<String, Object>> query(String sql) throws SQLException {
258-
return this.query(sql, new ArrayList<>());
269+
return this.jdbcQuery.query(sql);
259270
}
260271

261272
/**
@@ -267,23 +278,7 @@ public List<Map<String, Object>> query(String sql) throws SQLException {
267278
* @throws SQLException 查询异常
268279
*/
269280
public List<Map<String, Object>> query(String sql, List<Object> params) throws SQLException {
270-
PreparedStatement preparedStatement = connectionProxy.getConnection().prepareStatement(sql);
271-
for (int i = 0; i < params.size(); i++) {
272-
Object param = params.get(i);
273-
preparedStatement.setObject(i + 1, param);
274-
}
275-
ResultSet resultSet = preparedStatement.executeQuery();
276-
ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
277-
List<Map<String, Object>> list = new ArrayList<>();
278-
while (resultSet.next()) {
279-
Map<String, Object> map = new HashMap<>();
280-
for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
281-
map.put(resultSetMetaData.getColumnName(i), resultSet.getObject(i));
282-
}
283-
list.add(map);
284-
}
285-
resultSet.close();
286-
return list;
281+
return jdbcQuery.query(sql,params);
287282
}
288283

289284
/**
@@ -316,6 +311,7 @@ public List<Map<String, Object>> getStatementGenerateKeys(DbTable dbTable) {
316311
}
317312

318313

314+
319315
/**
320316
* 获取驱动配置信息
321317
*

‎src/main/java/com/codingapi/dbstream/listener/dbevent/DBEventExecuteListener.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,10 +124,11 @@ public void after(SQLExecuteState executeState, Object result) throws SQLExcepti
124124

125125
for (int i = 0; i < batchSize; i++) {
126126
DBEventParser dataParser = ThreadLocalContext.getInstance().get(i);
127+
SQLExecuteState sqlExecuteState = executeStateList.get(i);
127128
if (dataParser != null) {
128129
// 获取DB事件信息
129130
List<DBEvent> eventList = dataParser.loadEvents(arrays.get(i));
130-
TransactionEventPools.getInstance().addEvents(transactionKey, eventList);
131+
TransactionEventPools.getInstance().addEvents(sqlExecuteState.getJdbcQuery(),transactionKey, eventList);
131132
}
132133
}
133134
// 清空本地缓存数据
@@ -138,7 +139,7 @@ public void after(SQLExecuteState executeState, Object result) throws SQLExcepti
138139
if (dataParser != null) {
139140
// 获取DB事件信息
140141
List<DBEvent> eventList = dataParser.loadEvents(result);
141-
TransactionEventPools.getInstance().addEvents(transactionKey, eventList);
142+
TransactionEventPools.getInstance().addEvents(executeState.getJdbcQuery(),transactionKey, eventList);
142143
}
143144
// 清空本地缓存数据
144145
ThreadLocalContext.getInstance().remove();

‎src/main/java/com/codingapi/dbstream/parser/InsertDBEventParser.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ private void loadDefaultInsertDataList() throws SQLException {
128128
}
129129
}
130130
} else if (insertValue.isJdbc()) {
131-
value = paramList.get(i);
131+
value = paramList.get(insertValue.getJdbcParamIndex()-1);
132132
} else {
133133
value = insertValue.getValue();
134134
}

‎src/main/java/com/codingapi/dbstream/parser/InsertSQLParser.java‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -107,16 +107,20 @@ public String getValuesSQL() {
107107
public List<InsertValue> getValues() {
108108
List<InsertValue> insertValues = new ArrayList<>();
109109
List<String> values = SQLUtils.parseInsertSQLValues(this.getValuesSQL());
110-
for (String value : values) {
110+
int jdbcIndex = 0;
111+
for (String value:values) {
111112
InsertValue insertValue = new InsertValue();
112113
if (value.trim().equals("?")) {
113114
insertValue.setType(ValueType.JDBC);
115+
jdbcIndex++;
116+
insertValue.setValue(value+jdbcIndex);
114117
} else if (SQLUtils.isSQLKeyword(value.trim()) || value.trim().startsWith("(")) {
115118
insertValue.setType(ValueType.SELECT);
119+
insertValue.setValue(value);
116120
} else {
117121
insertValue.setType(ValueType.STATIC);
122+
insertValue.setValue(value);
118123
}
119-
insertValue.setValue(value);
120124
insertValues.add(insertValue);
121125
}
122126
return insertValues;
@@ -135,6 +139,10 @@ public boolean isSelect() {
135139
public boolean isJdbc() {
136140
return ValueType.JDBC == this.type;
137141
}
142+
143+
public int getJdbcParamIndex(){
144+
return Integer.parseInt(this.value.replace("?",""));
145+
}
138146
}
139147

140148
public static enum ValueType {

0 commit comments

Comments
 (0)