您的位置：首页 > 数据库 > MySQL

Mysql 基于 Amoeba 的水平和垂直分片

2016-06-20 17:44 615 查看

环境：

Servers:

Amoeba Server (Linux): 192.168.14.129

Mysql 1 Server (Linux): 192.168.14.131

Mysql 2 Server (Linux): 192.168.14.133

Clients

Mysql GUI Tools (Windows): 192.168.14.28

Java Programs (Eclipse): 192.168.14.28

假设以上程序都已经安装好了。

1. Mysql数据库远程访问授权

mysql 1 server 和 mysql 2 server 的 test 数据库，允许 amoeba server 访问。用户名：test_user 密码：1234

Sql代码

grant all on test.* to test_user@192.168.14.129 identified by '1234';

2. 创建测试表

在 mysql 1 server 中：

t_user:

Sql代码

mysql> create table test.t_user (

-> user_id integer unsigned not null,

-> user_name varchar(45),

-> user_address varchar(100),

-> primary key (user_id)

-> )engine=innodb;

Query OK, 0 rows affected (0.01 sec)

t_blog:

Sql代码

mysql> create table test.t_blog (

-> blog_id integer unsigned not null,

-> blog_title varchar(45),

-> blog_content text,

-> user_id integer,

-> primary key (blog_id)

-> )engine=innodb;

Query OK, 0 rows affected (0.00 sec)

t_message:

Sql代码

mysql> create table test.t_message (

-> message_id integer unsigned not null,

-> message_content varchar(255),

-> user_id integer

-> )engine=innodb;

Query OK, 0 rows affected (0.01 sec)

在 mysql 2 server 中

t_user 同上。

t_attention:

Sql代码

mysql> create table test.t_attention (

-> attention_id integer unsigned not null,

-> user_id integer,

-> blog_id integer

-> )engine=innodb;

Query OK, 0 rows affected (0.01 sec)

t_blog_comment:

Sql代码

mysql> create table test.t_blog_comment (

-> comment_id integer unsigned not null,

-> commnet_content text,

-> blog_id integer

-> )engine=innodb;

Query OK, 0 rows affected (0.01 sec)

3. 配置 Amoeba 的切分数据库的规则

Amoeba 的详细使用说明请参见：http://docs.hexnova.com/amoeba/

dbServers.xml

Xml代码

<?xml version="1.0" encoding="gbk"?>

<!DOCTYPE amoeba:dbServers SYSTEM "dbserver.dtd">

<amoeba:dbServers xmlns:amoeba="http://amoeba.meidusa.com/">



<dbServer name="abstractServer" abstractive="true">

<factoryConfig class="com.meidusa.amoeba.mysql.net.MysqlServerConnectionFactory">

<property name="manager">${defaultManager}</property>

<property name="sendBufferSize">64</property>

<property name="receiveBufferSize">128</property>



<property name="port">3306</property>



<property name="schema">test</property>



<property name="user">test_user</property>



<property name="password">1234</property>

</factoryConfig>

<poolConfig class="com.meidusa.amoeba.net.poolable.PoolableObjectPool">

<property name="maxActive">500</property>

<property name="maxIdle">500</property>

<property name="minIdle">10</property>

<property name="minEvictableIdleTimeMillis">600000</property>

<property name="timeBetweenEvictionRunsMillis">600000</property>

<property name="testOnBorrow">true</property>

<property name="testWhileIdle">true</property>

</poolConfig>

</dbServer>

<dbServer name="server1" parent="abstractServer">

<factoryConfig>



<property name="ipAddress">192.168.14.131</property>

</factoryConfig>

</dbServer>

<dbServer name="server2" parent="abstractServer">

<factoryConfig>



<property name="ipAddress">192.168.14.133</property>

</factoryConfig>

</dbServer>

</amoeba:dbServers>

amoeba.xml

Xml代码

<?xml version="1.0" encoding="gbk"?>

<!DOCTYPE amoeba:configuration SYSTEM "amoeba.dtd">

<amoeba:configuration xmlns:amoeba="http://amoeba.meidusa.com/">

<proxy>



<service name="Amoeba for Mysql" class="com.meidusa.amoeba.net.ServerableConnectionManager">



<property name="port">8066</property>





<property name="manager">${clientConnectioneManager}</property>

<property name="connectionFactory">

<bean class="com.meidusa.amoeba.mysql.net.MysqlClientConnectionFactory">

<property name="sendBufferSize">128</property>

<property name="receiveBufferSize">64</property>

</bean>

</property>

<property name="authenticator">

<bean class="com.meidusa.amoeba.mysql.server.MysqlClientAuthenticator">

<property name="user">root</property>

<property name="password">root</property>

<property name="filter">

<bean class="com.meidusa.amoeba.server.IPAccessController">

<property name="ipFile">${amoeba.home}/conf/access_list.conf</property>

</bean>

</property>

</bean>

</property>

</service>



<service name="Amoeba Monitor Server" class="com.meidusa.amoeba.monitor.MonitorServer">







<property name="ipAddress">127.0.0.1</property>

<property name="daemon">true</property>

<property name="manager">${clientConnectioneManager}</property>

<property name="connectionFactory">

<bean class="com.meidusa.amoeba.monitor.net.MonitorClientConnectionFactory"></bean>

</property>

</service>

<runtime class="com.meidusa.amoeba.mysql.context.MysqlRuntimeContext">



<property name="readThreadPoolSize">20</property>



<property name="clientSideThreadPoolSize">30</property>



<property name="serverSideThreadPoolSize">30</property>



<property name="statementCacheSize">500</property>



<property name="queryTimeout">60</property>

</runtime>

</proxy>



<connectionManagerList>

<connectionManager name="clientConnectioneManager" class="com.meidusa.amoeba.net.MultiConnectionManagerWrapper">

<property name="subManagerClassName">com.meidusa.amoeba.net.ConnectionManager</property>



</connectionManager>

<connectionManager name="defaultManager" class="com.meidusa.amoeba.net.MultiConnectionManagerWrapper">

<property name="subManagerClassName">com.meidusa.amoeba.net.AuthingableConnectionManager</property>



</connectionManager>

</connectionManagerList>



<dbServerLoader class="com.meidusa.amoeba.context.DBServerConfigFileLoader">

<property name="configFile">${amoeba.home}/conf/dbServers.xml</property>

</dbServerLoader>

<queryRouter class="com.meidusa.amoeba.mysql.parser.MysqlQueryRouter">

<property name="ruleLoader">

<bean class="com.meidusa.amoeba.route.TableRuleFileLoader">

<property name="ruleFile">${amoeba.home}/conf/rule.xml</property>

<property name="functionFile">${amoeba.home}/conf/ruleFunctionMap.xml</property>

</bean>

</property>

<property name="sqlFunctionFile">${amoeba.home}/conf/functionMap.xml</property>

<property name="LRUMapSize">1500</property>

<property name="defaultPool">server1</property>



<property name="needParse">true</property>

</queryRouter>

</amoeba:configuration>

rule.xml

t_user 表根据 user_id 字段的奇偶性水平切分，偶数分到 server1 ，奇数分到 server2。其中 server1，server2 是在 dbServers.xml 中定义的。

t_attention 表垂直切分到 server2

主数据库是 server1 ，在 amoeba.xml 中定义

Xml代码

<property name="defaultPool">server1</property>

主数据库说明：连接到 Amoeba 代理的时候，主数据库的所有表均可访问。但是其他的数据库的表，需要在分片规则中有涉及的表，才能访问，其他表不能访问。

Xml代码

<?xml version="1.0" encoding="gbk"?>

<!DOCTYPE amoeba:rule SYSTEM "rule.dtd">

<amoeba:rule xmlns:amoeba="http://amoeba.meidusa.com/">

<tableRule name="t_user" schema="test" defaultPools="server1,server2">

<rule name="rule1">

<parameters>user_id</parameters>

<expression><![CDATA[ user_id % 2 == 0 ]]></expression>

<defaultPools>server1</defaultPools>

</rule>

<rule name="rule2">

<parameters>user_id</parameters>

<expression><![CDATA[ user_id % 2 == 1 ]]></expression>

<defaultPools>server2</defaultPools>

</rule>

</tableRule>

<tableRule name="t_attention" schema="test" defaultPools="server2" />

<tableRule name="t_blog_comment" schema="test" defaultPools="server2" />

</amoeba:rule>

启动 msyql1, mysql2, amoeba 切分完成。

4. 测试

访问的时候，需要通过 Amoeba 代理来访问。不能直接访问实体数据库，不然切分无效。

使用 Mysql GUI Tools 连接 Amoeba

用户名：root 密码： root 端口号：8066

以上信息在 amoeba.xml 中定义

Xml代码

<service name="Amoeba for Mysql" class="com.meidusa.amoeba.net.ServerableConnectionManager">



<property name="port">8066</property>





<property name="manager">${clientConnectioneManager}</property>

<property name="connectionFactory">

<bean class="com.meidusa.amoeba.mysql.net.MysqlClientConnectionFactory">

<property name="sendBufferSize">128</property>

<property name="receiveBufferSize">64</property>

</bean>

</property>

<property name="authenticator">

<bean class="com.meidusa.amoeba.mysql.server.MysqlClientAuthenticator">



<property name="user">root</property>

<property name="password">root</property>

<property name="filter">

<bean class="com.meidusa.amoeba.server.IPAccessController">

<property name="ipFile">${amoeba.home}/conf/access_list.conf</property>

</bean>

</property>

</bean>

</property>

</service>

可以发现，在GUI 工具中，能看到的表只有在 server1 中，创建的表。上面已经说明了，因为 server1 中主数据库。

验证垂直切分的 t_attention 表

为了验证，垂直切分出去的 t_attention 表。直接执行 select * from test.t_attention;

可以看到，查询成功。说明的确可以访问，切分出去的 t_attention 表。

验证水平切分的 t_user 表

t_user 的水平切分规则是根据 user_id 的奇偶性进行切分。

上图的左下角可以看到：2 rows affected by the last command ，说明刚才那个插入语句，作用到了两条记录

查询看一下插入结果：

发现有两条一样的数据。到网上找了一下原因。

后来知道 amoeba 是根据 sql 解析来进行水平切分的，需要把切分的关键字段（这里是user_id），加入到 sql 中。否则切分规则无效。无效后，会在 server1, server2 均都插入数据。

即变为：insert into t_user(user_id, user_name, user_address) values(1, 'n1', 'a1')

分别检查一下，两个数据库

两个数据库均插入了一条相同的数据。同时也了解到，amoeba 的查询会将水平切分的表在两个数据库的结果合并。

修改 sql 语句后再执行一次。

可以看到左下角：只作用到一条记录

看下查询结果

因为 user_id 为偶数，所有分派到 server 1 了。

水平切分的排序与分页

排序结果：分别查询的 server1 和 server2 然后简单合并。所以排序不正确。

分页结果：同样是分别执行的 server1 和 server2 然后合并。每个 server 取两条记录，记录也变为了 4 条。分页不正确

验证连接查询

在 t_message , t_blog, t_attention 表分别插入两个数据

分别执行以上查询语句。

发现只有第二和第四可以执行成功。

结果：

1. 如果表1 被水平分片，连接查询时，需要连接的表，在表1 分片的所有数据库中均存在（即相同的分片规则），才可以连接查询。

2. 如果表1 被垂直分片，连接查询时，需要连接的表也同样分片到同一个数据库中（即相同的分片规则）。

3. 综上，只有在连接的表，在同一个数据库中均存在时，才能连接操作。即不支持跨数据库的连接。

验证事务管理

使用程序验证。

1. 验证没有分片的表的事务操作（无回滚）。

程序片段：

Java代码

package prx.dao;

import java.util.List;

import java.util.Map;

import prx.core.string.StringUtil;

import prx.dao.core.IPrxDao;

import prx.dao.core.impl.MySqlDao;

public class Test {

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=3; i<10; i++) {

dao.executeUpdate("insert into t_message(message_id, message_content, user_id) values(?, ?, ?)", new Object[]{i, "c" + i, i});

}

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_message", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

}

执行结果：

Cmd代码

2011/08/12 10:17:33 [INFO] MLog clients using log4j logging.

2011/08/12 10:17:34 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:17:34 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g3olvycduja2|7a78d3, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g3olvycduja2|7a78d3, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:17:36 [INFO] [开始事务]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[3,c3,3]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[4,c4,4]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[5,c5,5]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[6,c6,6]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[7,c7,7]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[8,c8,8]

2011/08/12 10:17:36 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:17:36 [INFO] values:[9,c9,9]

2011/08/12 10:17:36 [INFO] [事务提交]

2011/08/12 10:17:36 [INFO] select * from t_message

[message_id:1, message_content:c1, user_id:1]

[message_id:2, message_content:c2, user_id:2]

[message_id:3, message_content:c3, user_id:3]

[message_id:4, message_content:c4, user_id:4]

[message_id:5, message_content:c5, user_id:5]

[message_id:6, message_content:c6, user_id:6]

[message_id:7, message_content:c7, user_id:7]

[message_id:8, message_content:c8, user_id:8]

[message_id:9, message_content:c9, user_id:9]

执行正确无误。

2. 验证没有分片的表的操作（有回滚）。

程序片段：

Java代码

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=10; i<20; i++) {

dao.executeUpdate("insert into t_message(message_id, message_content, user_id) values(?, ?, ?)", new Object[]{i, "c" + i, i});

}

//主键冲突，出错，回滚。

dao.executeUpdate("insert into t_user(user_id) values(?)", new Object[]{1});

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_message", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

执行结果：

Cmd代码

2011/08/12 10:24:37 [INFO] MLog clients using log4j logging.

2011/08/12 10:24:37 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:24:37 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g3xpaud8zfzf|b169f8, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g3xpaud8zfzf|b169f8, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:24:38 [INFO] [开始事务]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[10,c10,10]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[11,c11,11]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[12,c12,12]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[13,c13,13]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[14,c14,14]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[15,c15,15]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[16,c16,16]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[17,c17,17]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[18,c18,18]

2011/08/12 10:24:38 [INFO] insert into t_message(message_id, message_content, user_id) values(?, ?, ?)

2011/08/12 10:24:38 [INFO] values:[19,c19,19]

2011/08/12 10:24:38 [INFO] insert into t_user(user_id) values(?)

2011/08/12 10:24:38 [INFO] values:[1]

2011/08/12 10:24:38 [ERROR] Duplicate entry '1' for key 'PRIMARY'

com.mysql.jdbc.exceptions.MySQLIntegrityConstraintViolationException: Duplicate entry '1' for key 'PRIMARY'

at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:931)

at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:2870)

at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:1573)

at com.mysql.jdbc.ServerPreparedStatement.serverExecute(ServerPreparedStatement.java:1169)

at com.mysql.jdbc.ServerPreparedStatement.executeInternal(ServerPreparedStatement.java:693)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1404)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1318)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1303)

at com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeUpdate(NewProxyPreparedStatement.java:105)

at prx.dao.core.AbstractPrxDao.executeUpdate(AbstractPrxDao.java:215)

at prx.dao.Test.main(Test.java:21)

2011/08/12 10:24:38 [ERROR] [事务失败，回滚]

2011/08/12 10:24:38 [INFO] select * from t_message

[message_id:1, message_content:c1, user_id:1]

[message_id:2, message_content:c2, user_id:2]

[message_id:3, message_content:c3, user_id:3]

[message_id:4, message_content:c4, user_id:4]

[message_id:5, message_content:c5, user_id:5]

[message_id:6, message_content:c6, user_id:6]

[message_id:7, message_content:c7, user_id:7]

[message_id:8, message_content:c8, user_id:8]

[message_id:9, message_content:c9, user_id:9]

执行正确无误。

3. 验证垂直分片的表的事务（无回滚）

Java代码

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=3; i<10; i++) {

dao.executeUpdate("insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)", new Object[]{i, i, i});

}

//主键冲突，出错，回滚。

//dao.executeUpdate("insert into t_user(user_id) values(?)", new Object[]{1});

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_attention", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

执行结果：

Cmd代码

2011/08/12 10:30:49 [INFO] MLog clients using log4j logging.

2011/08/12 10:30:50 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:30:50 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g45oipe8gwx3|b169f8, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g45oipe8gwx3|b169f8, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:30:50 [INFO] [开始事务]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[3,3,3]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[4,4,4]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[5,5,5]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[6,6,6]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[7,7,7]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[8,8,8]

2011/08/12 10:30:50 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:30:50 [INFO] values:[9,9,9]

2011/08/12 10:30:50 [INFO] [事务提交]

2011/08/12 10:30:50 [INFO] select * from t_attention

[attention_id:1, user_id:1, blog_id:1]

[attention_id:2, user_id:2, blog_id:2]

[attention_id:2, user_id:2, blog_id:2]

[attention_id:3, user_id:3, blog_id:3]

[attention_id:4, user_id:4, blog_id:4]

[attention_id:5, user_id:5, blog_id:5]

[attention_id:6, user_id:6, blog_id:6]

[attention_id:7, user_id:7, blog_id:7]

[attention_id:8, user_id:8, blog_id:8]

[attention_id:9, user_id:9, blog_id:9]

执行成功。

4. 验证垂直分片的表（有回滚）

Java代码

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=10; i<20; i++) {

dao.executeUpdate("insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)", new Object[]{i, i, i});

}

//主键冲突，出错，回滚。

dao.executeUpdate("insert into t_user(user_id) values(?)", new Object[]{1});

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_attention", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

执行结果：

Cmd代码

2011/08/12 10:32:46 [INFO] MLog clients using log4j logging.

2011/08/12 10:32:46 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:32:46 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g486lb1n4c9c4|b169f8, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g486lb1n4c9c4|b169f8, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:32:47 [INFO] [开始事务]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[10,10,10]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[11,11,11]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[12,12,12]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[13,13,13]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[14,14,14]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[15,15,15]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[16,16,16]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[17,17,17]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[18,18,18]

2011/08/12 10:32:47 [INFO] insert into t_attention(attention_id, user_id, blog_id) values(?, ?, ?)

2011/08/12 10:32:47 [INFO] values:[19,19,19]

2011/08/12 10:32:47 [INFO] insert into t_user(user_id) values(?)

2011/08/12 10:32:47 [INFO] values:[1]

2011/08/12 10:32:47 [ERROR] Duplicate entry '1' for key 'PRIMARY'

com.mysql.jdbc.exceptions.MySQLIntegrityConstraintViolationException: Duplicate entry '1' for key 'PRIMARY'

at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:931)

at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:2870)

at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:1573)

at com.mysql.jdbc.ServerPreparedStatement.serverExecute(ServerPreparedStatement.java:1169)

at com.mysql.jdbc.ServerPreparedStatement.executeInternal(ServerPreparedStatement.java:693)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1404)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1318)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1303)

at com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeUpdate(NewProxyPreparedStatement.java:105)

at prx.dao.core.AbstractPrxDao.executeUpdate(AbstractPrxDao.java:215)

at prx.dao.Test.main(Test.java:21)

2011/08/12 10:32:47 [ERROR] [事务失败，回滚]

2011/08/12 10:32:47 [INFO] select * from t_attention

[attention_id:1, user_id:1, blog_id:1]

[attention_id:2, user_id:2, blog_id:2]

[attention_id:2, user_id:2, blog_id:2]

[attention_id:3, user_id:3, blog_id:3]

[attention_id:4, user_id:4, blog_id:4]

[attention_id:5, user_id:5, blog_id:5]

[attention_id:6, user_id:6, blog_id:6]

[attention_id:7, user_id:7, blog_id:7]

[attention_id:8, user_id:8, blog_id:8]

[attention_id:9, user_id:9, blog_id:9]

[attention_id:10, user_id:10, blog_id:10]

[attention_id:11, user_id:11, blog_id:11]

[attention_id:12, user_id:12, blog_id:12]

[attention_id:13, user_id:13, blog_id:13]

[attention_id:14, user_id:14, blog_id:14]

[attention_id:15, user_id:15, blog_id:15]

[attention_id:16, user_id:16, blog_id:16]

[attention_id:17, user_id:17, blog_id:17]

[attention_id:18, user_id:18, blog_id:18]

[attention_id:19, user_id:19, blog_id:19]

执行错误，事务回滚失败

5. 验证水平分片的表的事务（无回滚）

Java代码

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=10; i<20; i++) {

dao.executeUpdate("insert into t_user(user_id, user_name, user_address) values(?, ?, ?)", new Object[]{i, "n"+i, "a"+i});

}

//主键冲突，出错，回滚。

//dao.executeUpdate("insert into t_user(user_id) values(?)", new Object[]{1});

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_user", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

执行结果：

Cmd代码

2011/08/12 10:41:20 [INFO] MLog clients using log4j logging.

2011/08/12 10:41:23 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:41:24 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g4j7pb1170bmp|929206, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g4j7pb1170bmp|929206, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:41:27 [INFO] [开始事务]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[10,n10,a10]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[11,n11,a11]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[12,n12,a12]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[13,n13,a13]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[14,n14,a14]

2011/08/12 10:41:27 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:27 [INFO] values:[15,n15,a15]

2011/08/12 10:41:28 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:28 [INFO] values:[16,n16,a16]

2011/08/12 10:41:28 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:28 [INFO] values:[17,n17,a17]

2011/08/12 10:41:28 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:28 [INFO] values:[18,n18,a18]

2011/08/12 10:41:28 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:41:28 [INFO] values:[19,n19,a19]

2011/08/12 10:41:28 [INFO] [事务提交]

2011/08/12 10:41:28 [INFO] select * from t_user

[user_id:2, user_name:n2, user_address:a2]

[user_id:4, user_name:n4, user_address:a4]

[user_id:6, user_name:n6, user_address:a6]

[user_id:10, user_name:n10, user_address:a10]

[user_id:12, user_name:n12, user_address:a12]

[user_id:14, user_name:n14, user_address:a14]

[user_id:16, user_name:n16, user_address:a16]

[user_id:18, user_name:n18, user_address:a18]

[user_id:1, user_name:n1, user_address:a1]

[user_id:3, user_name:n3, user_address:a3]

[user_id:5, user_name:n5, user_address:a5]

[user_id:11, user_name:n11, user_address:a11]

[user_id:13, user_name:n13, user_address:a13]

[user_id:15, user_name:n15, user_address:a15]

[user_id:17, user_name:n17, user_address:a17]

[user_id:19, user_name:n19, user_address:a19]

执行成功。

6. 验证水平分片的表的事务（有回滚）

Java代码

public static void main(String[] args) throws Exception {

IPrxDao dao = new MySqlDao();

//开启事务，插入几条数据

dao.startTransaction();

for(int i=20; i<30; i++) {

dao.executeUpdate("insert into t_user(user_id, user_name, user_address) values(?, ?, ?)", new Object[]{i, "n"+i, "a"+i});

}

//主键冲突，出错，回滚。

dao.executeUpdate("insert into t_user(user_id) values(?)", new Object[]{1});

dao.commitTransaction();

//查看结果

List<Map<String, Object>> list = dao.queryForMapList("select * from t_user", new Object[]{});

for(Map<String, Object> map : list) {

System.out.println(StringUtil.valueOfMap(map));

}

}

Cmd代码

2011/08/12 10:44:05 [INFO] MLog clients using log4j logging.

2011/08/12 10:44:05 [INFO] Initializing c3p0-0.9.1.2 [built 21-May-2007 15:04:56; debug? true; trace: 10]

2011/08/12 10:44:05 [INFO] Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, dataSourceName -> 1hge3oc8h1g4mqbxeevnxo|b169f8, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> com.mysql.jdbc.Driver, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, identityToken -> 1hge3oc8h1g4mqbxeevnxo|b169f8, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:mysql://192.168.14.129:8066/test?autoReconnect=true&characterEncoding=UTF-8, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, numThreadsAwaitingCheckoutDefaultUser -> 0, preferredTestQuery -> null, properties -> {user=******, password=******}, propertyCycle -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, usesTraditionalReflectiveProxies -> false ]

2011/08/12 10:44:05 [INFO] [开始事务]

2011/08/12 10:44:05 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:05 [INFO] values:[20,n20,a20]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[21,n21,a21]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[22,n22,a22]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[23,n23,a23]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[24,n24,a24]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[25,n25,a25]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[26,n26,a26]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[27,n27,a27]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[28,n28,a28]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id, user_name, user_address) values(?, ?, ?)

2011/08/12 10:44:06 [INFO] values:[29,n29,a29]

2011/08/12 10:44:06 [INFO] insert into t_user(user_id) values(?)

2011/08/12 10:44:06 [INFO] values:[1]

2011/08/12 10:44:06 [ERROR] Duplicate entry '1' for key 'PRIMARY'

com.mysql.jdbc.exceptions.MySQLIntegrityConstraintViolationException: Duplicate entry '1' for key 'PRIMARY'

at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:931)

at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:2870)

at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:1573)

at com.mysql.jdbc.ServerPreparedStatement.serverExecute(ServerPreparedStatement.java:1169)

at com.mysql.jdbc.ServerPreparedStatement.executeInternal(ServerPreparedStatement.java:693)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1404)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1318)

at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1303)

at com.mchange.v2.c3p0.impl.NewProxyPreparedStatement.executeUpdate(NewProxyPreparedStatement.java:105)

at prx.dao.core.AbstractPrxDao.executeUpdate(AbstractPrxDao.java:215)

at prx.dao.Test.main(Test.java:21)

2011/08/12 10:44:06 [ERROR] [事务失败，回滚]

2011/08/12 10:44:06 [INFO] select * from t_user

[user_id:1, user_name:n1, user_address:a1]

[user_id:3, user_name:n3, user_address:a3]

[user_id:5, user_name:n5, user_address:a5]

[user_id:11, user_name:n11, user_address:a11]

[user_id:13, user_name:n13, user_address:a13]

[user_id:15, user_name:n15, user_address:a15]

[user_id:17, user_name:n17, user_address:a17]

[user_id:19, user_name:n19, user_address:a19]

[user_id:21, user_name:n21, user_address:a21]

[user_id:23, user_name:n23, user_address:a23]

[user_id:25, user_name:n25, user_address:a25]

[user_id:27, user_name:n27, user_address:a27]

[user_id:29, user_name:n29, user_address:a29]

[user_id:2, user_name:n2, user_address:a2]

[user_id:4, user_name:n4, user_address:a4]

[user_id:6, user_name:n6, user_address:a6]

[user_id:10, user_name:n10, user_address:a10]

[user_id:12, user_name:n12, user_address:a12]

[user_id:14, user_name:n14, user_address:a14]

[user_id:16, user_name:n16, user_address:a16]

[user_id:18, user_name:n18, user_address:a18]

结果：

user_id 为偶数的部分都没有插入数据，回滚成功。 user_id 为偶数分片到 server1 为主数据库

user_id 为奇数的部分都已经插入数据，回滚失败。 user_id 为奇数分片到 server2

综上：所有分片的表，都不能支持事务。

验证存储过程

这个就不写验证过程了。直接写结果好了。

结果：

1. 只可以调用 server1 主数据库的 procedure , 不能调用 server2 的。

2. server1 的 procedure 只能操作 server1 中的表，不支持水平和垂直分片。水平切分的 t_user，不管 user_id 奇偶均插入到 server1。垂直切分的t_attention 表不能访问。

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签： mysql java

相关文章推荐

新的分享

章节导航

Mysql 基于 Amoeba 的 水平和垂直 分片

Mysql 基于 Amoeba 的水平和垂直分片