问题描述
某产品线的某个RocketMQ GroupID消息大量积压；DBA反馈该应用使用的TBase数据库带宽流量飙升，TBase数据库的CPU、Load处于正常范围。
问题分析
应用逻辑
该应用主要逻辑是从RocketMQ接收消息，如果该消息在TBase中不存在，则插入相关信息到TBase中；如果该消息在TBase中存在，则更新相关信息。
主要执行的SQL类似下面这个样子：
update test set a = ?,b = ?,c = ? where d = ?
TBase JDBC Driver
<dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <version>42.3.3</version> </dependency>
无灰度不变更
检查该应用变更记录，发现该应用是在12点40分左右开始变更的，与问题产生时间吻合，所以回滚应用，但是回滚之后问题依旧存在；
检查该应用配置变更记录，发现该应用最近一次配置变更是2023年12月20日，配置变更之后，发过几次发版，排除配置变更导致故障的情况。
TBase数据库飙升的带宽来自哪里？
检查该应用ECS出口带宽在12点40分左右开始飙升，与TBASE数据库带宽飙升时间吻合。
TBase带宽打满会导致消息处理变慢，从RocketMQ控制台监控指标也可以说明这一点。
消息处理变慢意味着打到TBase的请求减少了，但是带宽却增加了很多，意味着打到TBase的一个请求的字节数增大了，所以关键是需要分析出针对一个请求增加了哪些字节。
为了便于理解，首先介绍下TBase中与本案例相关的通信协议（内容来自网络）。
Extended Query
Extended Query 协议将以上 Simple Query 的处理流程分为若干步骤，每一步都由单独的服务端消息进行确认。该协议可以使用服务端的 perpared-statement 功能，即先发送一条参数化 SQL，服务端收到 SQL（Statement）之后对其进行解析、重写并保存，这里保存的 Statement 也就是所谓 Prepared-statement，可以被复用；执行 SQL 时，直接获取事先保存的 Prepared-statement 生成计划并执行，避免对同类型 SQL 重复解析和重写。
如下例， SELECT * FROM users u, logs l WHERE u.usrid=$1 AND u.usrid=l.usrid AND l.date = $2; 是一条参数化 SQL，执行 PREPARE 时，服务端对该 SQL 进行解析和重写；执行 EXECUTE 时，为 Prepared Statement 生成计划并执行。第二次执行 EXECUTE 时无需再对 SQL 进行解析和重写，直接生成计划并执行即可。
可见，Extended Query 协议通过使用服务端的 Prepared Statement，提升同类 SQL 多次执行的效率。但与 Simple Query 相比，其不允许在一个请求中包含多条 SQL 命令，否则会报语法错误。
Extended Query 协议通常包括 5 个步骤，分别是 Parse，Bind，Describe，Execute 和 Sync。以下分别介绍各个阶段的处理流程。
Parse
客户端首先向服务端发送一个 Parse 消息，该消息包括参数化 SQL，参数占位符以及每个参数的类型，还可以指定 Statement 的名字，若不指定名字，即为一个 “未命名” 的 Statement，该 Statement 会在生成下一个 “未命名” Statement 时予以销毁，若指定名字，则必须在下次发送 Parse 消息前将其显式销毁。
PostgreSQL 服务端收到该消息后，调用 exec_parse_message 函数进行处理，进行语法分析、语义分析和重写，同时会创建一个 Plan Cache 的结构，用于缓存后续的执行计划。
Bind
客户端发送 Bind 消息，该消息携带具体的参数值、参数格式和返回列的格式，如下：
PostgreSQL 收到该消息后，调用 exec_bind_message 函数进行处理。为之前保存的 Prepared Statement 创建执行计划并将其保存在 Plan Cache 中，创建一个 Portal 用于后续执行。关于 Plan Cache 的具体实现和复用逻辑在此不细述，以后单独撰文介绍。
在 PostgreSQL 内核中，Portal 是对查询执行状态的一种抽象，该结构贯穿执行器运行的始终。
Describe
客户端可以发送 Describe 消息获取 Statment 或 Portal 的元信息，即返回结果的列名，类型等信息，这些信息由 RowDescription 消息携带。如果请求获取 Statement 的元信息，还会返回具体的参数信息，由 ParameterDescription 消息携带。
Execute
客户端发送 Execute 消息告知服务端执行请求，服务端收到消息后，执行 Bind 阶段创建的 Portal，执行结果通过 DataRow 消息返回给客户端，执行完成后发送 CommandComplete 。
Execute 消息中可以指定返回的行数，若行数为 0，表示返回所有行。
Sync
使用 Extended Query 协议时，一个请求总是以 Sync 消息结束，服务端接收到 Sync 消息后，关闭隐式开启的事务并回复 ReadyForQuery 消息。
Extended Query 完整的消息流如下：
所以接下来的思路是tcpdump抓包，然后通过wireshark分析出一个请求发送了哪些报文到TBase就可以基本确认问题了。
可惜的是抓的包wireshark并不能识别出TBase协议；另一个思路是TBase服务端能不能分析出一个请求具体涉及哪些报文，TBase技术支持最终没有提供出来，至此通过分析报文的方式没有走通。
另外一个思路是从应用程序侧分析TBase JDBC Driver发送请求的逻辑。
TBase JDBC Driver
通过分析Driver相关代码，找出Driver请求TBase相关逻辑，然后抓取运行时一些参数。
通过arthas抓包发现在PreparedStatement.executeUpdate()真正执行之前，Driver似乎就像TBase服务端发送了报文：
以上这个点是很可疑的，所以接下来抓取调用栈：
ts=2024-01-23 00:23:56;thread_name=ConsumeMessageThread_12;id=14a;is_daemon=false;priority=5;TCCL=org.apache.catalina.loader.ParallelWebappClassLoader@246b0ff2;trace_id=0a04cb0d17059202723904892d7747;rpc_id=0.1.10.1.104207185.11933.1.10413251.1.1.104216118.1.1.10413223 @org.postgresql.core.v3.QueryExecutorImpl.sendQuery() at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:347) at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:315) at org.postgresql.jdbc.PgPreparedStatement.getParameterMetaData(PgPreparedStatement.java:1659) at com.alibaba.druid.pool.DruidPooledPreparedStatement.getParameterMetaData(DruidPooledPreparedStatement.java:710) at org.springframework.jdbc.core.StatementCreatorUtils.setNull(StatementCreatorUtils.java:276) at org.springframework.jdbc.core.StatementCreatorUtils.setParameterValueInternal(StatementCreatorUtils.java:235) at org.springframework.jdbc.core.StatementCreatorUtils.setParameterValue(StatementCreatorUtils.java:169) at org.springframework.jdbc.core.ArgumentPreparedStatementSetter.doSetValue(ArgumentPreparedStatementSetter.java:66) at org.springframework.jdbc.core.ArgumentPreparedStatementSetter.setValues(ArgumentPreparedStatementSetter.java:47) at org.springframework.jdbc.core.JdbcTemplate$2.doInPreparedStatement(JdbcTemplate.java:875) at org.springframework.jdbc.core.JdbcTemplate$2.doInPreparedStatement(JdbcTemplate.java:870) at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:633) at org.springframework.jdbc.core.JdbcTemplate.update(JdbcTemplate.java:870) at org.springframework.jdbc.core.JdbcTemplate.update(JdbcTemplate.java:931) at org.springframework.jdbc.core.JdbcTemplate.update(JdbcTemplate.java:941) at xxxxx at com.aliyun.openservices.ons.api.impl.rocketmq.ConsumerImpl$MessageListenerImpl.consumeMessage(ConsumerImpl.java:116) at com.alibaba.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:710) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)
接下来重点分析下调用栈，为了便于说明，以上调用栈的复现可以用如下代码表示：
import com.alibaba.druid.pool.DruidDataSource; import org.springframework.jdbc.core.JdbcTemplate; import java.sql.SQLException; public class JdbcTemplateTest { public static void main(String[] args) throws SQLException { DruidDataSource dataSource = new DruidDataSource(); dataSource.setDriverClassName("org.postgresql.Driver"); dataSource.setUsername("postgres"); dataSource.setPassword("admin"); dataSource.setUrl("jdbc:postgresql://localhost:5432/postgres"); dataSource.setTimeBetweenEvictionRunsMillis(10 * 60 * 1000); dataSource.setInitialSize(0); dataSource.setPhyTimeoutMillis(1000); dataSource.setTestWhileIdle(false); dataSource.setFilters("stat"); dataSource.init(); JdbcTemplate jdbcTemplate = new JdbcTemplate(); jdbcTemplate.setDataSource(dataSource); String sql = "update test set name = ?,address = ? where id = ?"; Object[] params = new Object[3]; params[0] = null; params[1] = null; params[2] = 1; int size = jdbcTemplate.update(sql,params); System.out.println(size); } }
按照线程栈的调用关系，以此分析相关类的代码：
ArgumentPreparedStatementSetter
public void setValues(PreparedStatement ps) throws SQLException { if (this.args != null) { for (int i = 0; i < this.args.length; i++) { Object arg = this.args[i]; doSetValue(ps, i + 1, arg); } } } protected void doSetValue(PreparedStatement ps, int parameterPosition, Object argValue) throws SQLException { if (argValue instanceof SqlParameterValue) { SqlParameterValue paramValue = (SqlParameterValue) argValue; StatementCreatorUtils.setParameterValue(ps, parameterPosition, paramValue, paramValue.getValue()); } else { StatementCreatorUtils.setParameterValue(ps, parameterPosition, SqlTypeValue.TYPE_UNKNOWN, argValue); } }
StatementCreatorUtils
private static void setParameterValueInternal(PreparedStatement ps, int paramIndex, int sqlType, String typeName, Integer scale, Object inValue) throws SQLException { String typeNameToUse = typeName; int sqlTypeToUse = sqlType; Object inValueToUse = inValue; // override type info? if (inValue instanceof SqlParameterValue) { SqlParameterValue parameterValue = (SqlParameterValue) inValue; if (logger.isDebugEnabled()) { logger.debug("Overriding type info with runtime info from SqlParameterValue: column index " + paramIndex + ", SQL type " + parameterValue.getSqlType() + ", type name " + parameterValue.getTypeName()); } if (parameterValue.getSqlType() != SqlTypeValue.TYPE_UNKNOWN) { sqlTypeToUse = parameterValue.getSqlType(); } if (parameterValue.getTypeName() != null) { typeNameToUse = parameterValue.getTypeName(); } inValueToUse = parameterValue.getValue(); } if (logger.isTraceEnabled()) { logger.trace("Setting SQL statement parameter value: column index " + paramIndex + ", parameter value [" + inValueToUse + "], value class [" + (inValueToUse != null ? inValueToUse.getClass().getName() : "null") + "], SQL type " + (sqlTypeToUse == SqlTypeValue.TYPE_UNKNOWN ? "unknown" : Integer.toString(sqlTypeToUse))); } if (inValueToUse == null) { setNull(ps, paramIndex, sqlTypeToUse, typeNameToUse); } else { setValue(ps, paramIndex, sqlTypeToUse, typeNameToUse, scale, inValueToUse); } } private static void setNull(PreparedStatement ps, int paramIndex, int sqlType, String typeName) throws SQLException { if (sqlType == SqlTypeValue.TYPE_UNKNOWN || sqlType == Types.OTHER) { boolean useSetObject = false; Integer sqlTypeToUse = null; DatabaseMetaData dbmd = null; String jdbcDriverName = null; boolean tryGetParameterType = true; if (shouldIgnoreGetParameterType == null) { try { dbmd = ps.getConnection().getMetaData(); jdbcDriverName = dbmd.getDriverName(); tryGetParameterType = !driversWithNoSupportForGetParameterType.contains(jdbcDriverName); if (tryGetParameterType && jdbcDriverName.startsWith("Oracle")) { // Avoid getParameterType use with Oracle 12c driver by default: // needs to be explicitly activated through spring.jdbc.getParameterType.ignore=false tryGetParameterType = false; driversWithNoSupportForGetParameterType.add(jdbcDriverName); } } catch (Throwable ex) { logger.debug("Could not check connection metadata", ex); } } else { tryGetParameterType = !shouldIgnoreGetParameterType; } if (tryGetParameterType) { try { // 问题出在ps.getParameterMetaData()调用上 sqlTypeToUse = ps.getParameterMetaData().getParameterType(paramIndex); } catch (Throwable ex) { if (logger.isDebugEnabled()) { logger.debug("JDBC 3.0 getParameterType call not supported - using fallback method instead: " + ex); } } } if (sqlTypeToUse == null) { // JDBC driver not compliant with JDBC 3.0 -> proceed with database-specific checks sqlTypeToUse = Types.NULL; try { if (dbmd == null) { dbmd = ps.getConnection().getMetaData(); } if (jdbcDriverName == null) { jdbcDriverName = dbmd.getDriverName(); } if (shouldIgnoreGetParameterType == null) { // Register JDBC driver with no support for getParameterType, except for the // Oracle 12c driver where getParameterType fails for specific statements only // (so an exception thrown above does not indicate general lack of support). driversWithNoSupportForGetParameterType.add(jdbcDriverName); } String databaseProductName = dbmd.getDatabaseProductName(); if (databaseProductName.startsWith("Informix") || (jdbcDriverName.startsWith("Microsoft") && jdbcDriverName.contains("SQL Server"))) { // "Microsoft SQL Server JDBC Driver 3.0" versus "Microsoft JDBC Driver 4.0 for SQL Server" useSetObject = true; } else if (databaseProductName.startsWith("DB2") || jdbcDriverName.startsWith("jConnect") || jdbcDriverName.startsWith("SQLServer")|| jdbcDriverName.startsWith("Apache Derby")) { sqlTypeToUse = Types.VARCHAR; } } catch (Throwable ex) { logger.debug("Could not check connection metadata", ex); } } if (useSetObject) { ps.setObject(paramIndex, null); } else { ps.setNull(paramIndex, sqlTypeToUse); } } else if (typeName != null) { ps.setNull(paramIndex, sqlType, typeName); } else { ps.setNull(paramIndex, sqlType); } }
PgPreparedStatement
public ParameterMetaData getParameterMetaData() throws SQLException { int flags = QueryExecutor.QUERY_ONESHOT | QueryExecutor.QUERY_DESCRIBE_ONLY | QueryExecutor.QUERY_SUPPRESS_BEGIN; StatementResultHandler handler = new StatementResultHandler(); connection.getQueryExecutor().execute(preparedQuery.query, preparedParameters, handler, 0, 0, flags); int[] oids = preparedParameters.getTypeOIDs(); return createParameterMetaData(connection, oids); }
QueryExecutorImpl
private void sendOneQuery(SimpleQuery query, SimpleParameterList params, int maxRows, int fetchSize, int flags) throws IOException { boolean asSimple = (flags & QueryExecutor.QUERY_EXECUTE_AS_SIMPLE) != 0; if (asSimple) { assert (flags & QueryExecutor.QUERY_DESCRIBE_ONLY) == 0 : "Simple mode does not support describe requests. sql = " + query.getNativeSql() + ", flags = " + flags; sendSimpleQuery(query, params); return; } assert !query.getNativeQuery().multiStatement : "Queries that might contain ; must be executed with QueryExecutor.QUERY_EXECUTE_AS_SIMPLE mode. " + "Given query is " + query.getNativeSql(); // Per https://www.postgresql.org/docs/current/static/protocol-flow.html#PROTOCOL-FLOW-EXT-QUERY // A Bind message can use the unnamed prepared statement to create a named portal. // If the Bind is successful, an Execute message can reference that named portal until either // the end of the current transaction // or the named portal is explicitly destroyed boolean noResults = (flags & QueryExecutor.QUERY_NO_RESULTS) != 0; boolean noMeta = (flags & QueryExecutor.QUERY_NO_METADATA) != 0; boolean describeOnly = (flags & QueryExecutor.QUERY_DESCRIBE_ONLY) != 0; // extended queries always use a portal // the usePortal flag controls whether or not we use a *named* portal boolean usePortal = (flags & QueryExecutor.QUERY_FORWARD_CURSOR) != 0 && !noResults && !noMeta && fetchSize > 0 && !describeOnly; boolean oneShot = (flags & QueryExecutor.QUERY_ONESHOT) != 0; boolean noBinaryTransfer = (flags & QUERY_NO_BINARY_TRANSFER) != 0; boolean forceDescribePortal = (flags & QUERY_FORCE_DESCRIBE_PORTAL) != 0; // Work out how many rows to fetch in this pass. int rows; if (noResults) { rows = 1; // We're discarding any results anyway, so limit data transfer to a minimum } else if (!usePortal) { rows = maxRows; // Not using a portal -- fetchSize is irrelevant } else if (maxRows != 0 && fetchSize > maxRows) { // fetchSize > maxRows, use maxRows (nb: fetchSize cannot be 0 if usePortal == true) rows = maxRows; } else { rows = fetchSize; // maxRows > fetchSize } sendParse(query, params, oneShot); // Must do this after sendParse to pick up any changes to the // query's state. // boolean queryHasUnknown = query.hasUnresolvedTypes(); boolean paramsHasUnknown = params.hasUnresolvedTypes(); boolean describeStatement = describeOnly || (!oneShot && paramsHasUnknown && queryHasUnknown && !query.isStatementDescribed()); if (!describeStatement && paramsHasUnknown && !queryHasUnknown) { int[] queryOIDs = castNonNull(query.getPrepareTypes()); int[] paramOIDs = params.getTypeOIDs(); for (int i = 0; i < paramOIDs.length; i++) { // Only supply type information when there isn't any // already, don't arbitrarily overwrite user supplied // type information. if (paramOIDs[i] == Oid.UNSPECIFIED) { params.setResolvedType(i + 1, queryOIDs[i]); } } } if (describeStatement) { sendDescribeStatement(query, params, describeOnly); if (describeOnly) { return; } } // Construct a new portal if needed. Portal portal = null; if (usePortal) { String portalName = "C_" + (nextUniqueID++); portal = new Portal(query, portalName); } sendBind(query, params, portal, noBinaryTransfer); // A statement describe will also output a RowDescription, // so don't reissue it here if we've already done so. // if (!noMeta && !describeStatement) { /* * don't send describe if we already have cached the row description from previous executions * * XXX Clearing the fields / unpreparing the query (in sendParse) is incorrect, see bug #267. * We might clear the cached fields in a later execution of this query if the bind parameter * types change, but we're assuming here that they'll still be valid when we come to process * the results of this query, so we don't send a new describe here. We re-describe after the * fields are cleared, but the result of that gets processed after processing the results from * earlier executions that we didn't describe because we didn't think we had to. * * To work around this, force a Describe at each execution in batches where this can be a * problem. It won't cause more round trips so the performance impact is low, and it'll ensure * that the field information available when we decoded the results. This is undeniably a * hack, but there aren't many good alternatives. */ if (!query.isPortalDescribed() || forceDescribePortal) { sendDescribePortal(query, portal); } } sendExecute(query, portal, rows); }
通过以上分析，可以确认针对每个值为null的字段，多发送了Parse、Describe、Execute、Sync报文。
Parse
Describe
解决方法
通过以上分析可知，造成流量增大的原因是Spring JdbcTemplate中针对值为null的字段，调用了PgPreparedStatement.getParameterMetaData导致的。
所以规避方式是不触发这个调用。
方法一
自定义字段的设置方式的实现类。
jdbcTemplate.update(sql, new PreparedStatementSetter() { @Override public void setValues(PreparedStatement ps) throws SQLException { ps.setString(1,null); ps.setString(2,null); ps.setInt(3,1); } });
方法二
依然调用原有JdbcTemplate方法，需要保证字段值不为null；
方法三
依然调用原有JdbcTemplate方法，JVM启动参数配置（需测试）：
-Dspring.jdbc.getParameterType.ignore=true
MySQL JDBC Driver
MySQL JDBC Driver是否存在这个问题？经过分析相关代码，及ECS带宽流量，可以确定不存在这个问题。
public ParameterMetaData getParameterMetaData() throws SQLException { synchronized (checkClosed().getConnectionMutex()) { if (this.parameterMetaData == null) { if (this.connection.getGenerateSimpleParameterMetadata()) { this.parameterMetaData = new MysqlParameterMetadata(this.parameterCount); } else { this.parameterMetaData = new MysqlParameterMetadata(null, this.parameterCount, getExceptionInterceptor()); } } return this.parameterMetaData; } }
