使用CGLIB和Java动态代理实现spring的事务控制

• spring事务控制分析
• DatasourceTransactionUtils数据源和事务控制
• ProxyStrategy策略模式接口
• JdkTransactionProxy，Java动态代理实现事务控制
• CglibTransactionProxy，CGLIG动态代理实现事务控制
• ProxyContext
• UserService接口
• UserServiceImpl
• UserMapper
• HelloServiceImpl
• TestProxy测试类
• 打印结果
• 后记

spring事务控制分析

Spring的事务管理机制实现的原理，就是通过这样一个动态代理对所有需要事务管理的Bean进行加载，并根据配置在invoke方法中对当前调用的 方法名进行判定，并在method.invoke方法前后为其加上合适的事务管理代码，这样就实现了Spring式的事务管理。Spring中的AOP实 现更为复杂和灵活，不过基本原理是一致的。本文使用Java自带的动态代理和CGLIB两种方式来实现一个简单的事务控制

DatasourceTransactionUtils数据源和事务控制

package com.parallel.stomp.test.aop;

import com.alibaba.druid.pool.DruidDataSourceFactory;

import javax.sql.DataSource;
import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Properties;

/**
* 初始化数据库连接池和事务管理器
*
* @author rewnei2
* @version v0.1 2019/5/10 16:25
*/
public class DatasourceTransactionUtils {
private static ThreadLocal<Connection> threadLocal = new ThreadLocal<>();
private static DataSource dataSource;

static {
try {
Properties properties = new Properties();
InputStream inputStream = DatasourceTransactionUtils.class.getClassLoader().getResourceAsStream("druid.properties");
properties.load(inputStream);
dataSource = DruidDataSourceFactory.createDataSource(properties);
} catch (IOException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}

private static DataSource getDataSource() {
return dataSource;
}

private static Connection getConnection() {
try {
return dataSource.getConnection();
} catch (SQLException e) {
throw new RuntimeException("获取连接失败");
}
}

/**
* 获取与线程绑定的connection
*
* @return
*/
public static Connection getCurrentConnection() {
Connection connection = threadLocal.get();
if (connection == null) {
connection = getConnection();
threadLocal.set(connection);
}
//打印当前线程的名称
//System.out.println(Thread.currentThread().getName());
return connection;
}

/**
* 开启事务
*/
public static void beginTransaction() {
try {
getCurrentConnection().setAutoCommit(false);
System.out.println("线程" + Thread.currentThread().getName() + ": start transaction");
} catch (SQLException e) {
throw new RuntimeException("开启事务失败");
}
}

/**
* 提交事务
*/
public static void commit() {
try {
getCurrentConnection().commit();
System.out.println("线程" + Thread.currentThread().getName() + ": commit transaction");
} catch (SQLException e) {
throw new RuntimeException("提交事务失败");
}
}

/**
* 数据库事务回滚
*/
public static void rollback() {
try {
getCurrentConnection().rollback();
System.out.println("线程" + Thread.currentThread().getName() + ": rollback transaction");
} catch (SQLException e) {
throw new RuntimeException("回滚事务失败");
}
}

/**
* 释放资源
*/
public static void close() {
Connection connection = getCurrentConnection();
try {
connection.close();
threadLocal.remove();
} catch (SQLException e) {
throw new RuntimeException("释放资源失败");
}
}

public static void main(String[] args) {
Connection connection = DatasourceTransactionUtils.getCurrentConnection();
System.out.println(connection);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
Connection connection = DatasourceTransactionUtils.getCurrentConnection();
System.out.println(connection);
}
});
thread.start();
}

}

ProxyStrategy策略模式接口

package com.parallel.stomp.test.aop;

/**
* @author rewnei2
* @version v0.1 2019/5/10 18:11
*/
public interface ProxyStrategy {
/**
* 生成代理对象
* @param object
* @return
*/
public Object proxyObject(Object object);
}

JdkTransactionProxy，Java动态代理实现事务控制

package com.parallel.stomp.test.aop;

import org.springframework.transaction.annotation.Transactional;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

/**
* 使用JDK动态代理实现事务
*
* @author rewnei2
* @version v0.1 2019/5/10 17:21
*/
public class JdkTransactionProxy implements ProxyStrategy {

public Object proxyObject(Object object) {
/**
*ClassLoader loader:类加载器；被代理对象.getClass().getClassLoader()
*Class<?>[] interfaces:当前类的接口，jdk的动态代理对象必须实现接口；被代理对象.getClass().getInterfaces()
*InvocationHandler h：处理器
*/
return Proxy.newProxyInstance(object.getClass().getClassLoader(),
object.getClass().getInterfaces(), new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
/**
* 1. proxy:代理对象
* 2. method：代理对象的方法，被封装成为对象
* 3. args:代理对象调用的方法时，传递的实际参数
*/
//获取原始对象
Method originalMethod = object.getClass().getMethod(method.getName(), method.getParameterTypes());
//System.out.println(originalMethod.isAnnotationPresent(Transactional.class));
//当前的方法没有被@Transactional注解修饰
if (!originalMethod.isAnnotationPresent(Transactional.class)) {
return method.invoke(object, args);
}
//事务方法调用
DatasourceTransactionUtils.beginTransaction();
Object result = null;
try {
result = method.invoke(object, args);
DatasourceTransactionUtils.commit();
} catch (Exception e) {
DatasourceTransactionUtils.rollback();
} finally {
DatasourceTransactionUtils.close();
}
return result;
}
});
}

}

CglibTransactionProxy，CGLIG动态代理实现事务控制

package com.parallel.stomp.test.aop;

import org.springframework.cglib.proxy.Enhancer;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;
import org.springframework.transaction.annotation.Transactional;

import java.lang.reflect.Method;

/**
* 使用
*
* @author rewnei2
* @version v0.1 2019/5/10 16:14
*/
public class CglibTransactionProxy implements ProxyStrategy {
/**
* 获取代理对象
*
* @param clazz
* @param <T>
* @return
*/
public <T> T getProxy(Class<T> clazz) {
//使用CGLIB生成代理对象
Enhancer enhancer = new Enhancer();
enhancer.setSuperclass(clazz);
// 设置回调方法
enhancer.setCallback(new MethodInterceptor() {
/**
* 事务方法拦截
* @param object 被代理的对象（service对象）
* @param method 被调用的方法，原始service方法
* @param args 被代理方法的参数
* @param methodProxy 多出来的参数是MethodProxy 类型的，它应该是cglib生成用来代替Method对象的一个对象
* @return
* @throws Throwable
*/
@Override
public Object intercept(Object object, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
Object result = null;
boolean flag = method.isAnnotationPresent(Transactional.class);
if (!flag) {
return methodProxy.invokeSuper(object, args);
}
//开启事务
DatasourceTransactionUtils.beginTransaction();
try {
result = methodProxy.invokeSuper(object, args);
//事务提交
DatasourceTransactionUtils.commit();
} catch (Exception e) {
//事务回滚
DatasourceTransactionUtils.rollback();
throw e;
} finally {
DatasourceTransactionUtils.close();
}
return result;
}
});
return (T) enhancer.create();
}

@Override
public Object proxyObject(Object object) {
return getProxy(object.getClass());
}
}

ProxyContext

package com.parallel.stomp.test.aop;

import java.util.HashMap;
import java.util.Map;

/**
* @author rewnei2
* @version v0.1 2019/5/10 18:00
*/
public class ProxyContext {
private static final String JDK_PROXY_KEY = "jdk";

private static final String CGLIB_PROXY_KEY = "cglib";

private static Map<String, ProxyStrategy> strategyMap = new HashMap<>();

static {
strategyMap.put(JDK_PROXY_KEY, new JdkTransactionProxy());
strategyMap.put(CGLIB_PROXY_KEY, new CglibTransactionProxy());
}

public Object createProxy(Object object) {
if (object == null) {
throw new RuntimeException("参数异常");
}
/**
* 获取对象的接口信息
*/
Class<?>[] interfaces = object.getClass().getInterfaces();
ProxyStrategy strategy = null;
if (interfaces == null || interfaces.length == 0) {
strategy = strategyMap.get(CGLIB_PROXY_KEY);
} else {
strategy = strategyMap.get(JDK_PROXY_KEY);
}
return strategy.proxyObject(object);
}

}

UserService接口

package com.parallel.stomp.test.aop;

/**
* @author rewnei2
* @version v0.1 2019/5/10 17:41
*/
public interface UserService {

public void saveUser();
}

UserServiceImpl

package com.parallel.stomp.test.aop;

import org.springframework.transaction.annotation.Transactional;

/**
* @author rewnei2
* @version v0.1 2019/5/10 16:59
*/
public class UserServiceImpl implements UserService {

@Transactional
public void saveUser() {
System.out.println(Thread.currentThread().getName() + " save user");
UserMapper userMapper = new UserMapper();
userMapper.add();
//主动抛出异常
//throw new RuntimeException("主动异常");
}

}

UserMapper

package com.parallel.stomp.test.aop;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
* @author rewnei2
* @version v0.1 2019/5/13 10:53
*/
public class UserMapper {

public void add() {
System.out.println(Thread.currentThread().getName() + " add");
Connection connection = DatasourceTransactionUtils.getCurrentConnection();
String sql = "insert into test_user(id,name) values(4,'2222')";
try {
PreparedStatement preparedStatement = connection.prepareStatement(sql);
preparedStatement.executeUpdate();
} catch (SQLException e) {
e.printStackTrace();
}

}

}

HelloServiceImpl

package com.parallel.stomp.test.aop;

import org.springframework.transaction.annotation.Transactional;

/**
* @author rewnei2
* @version v0.1 2019/5/10 18:17
*/
public class HelloServiceImpl {
@Transactional
public void hello() {
System.out.println("hello cglib");
}

}

TestProxy测试类

package com.parallel.stomp.test.aop;

/**
* @author rewnei2
* @version v0.1 2019/5/10 17:42
*/
public class TestProxy {
public static void main(String[] args) {
UserService userService = new UserServiceImpl();
ProxyContext proxyContext = new ProxyContext();
UserService proxy = (UserService) proxyContext.createProxy(userService);
proxy.saveUser();
new Thread(new Runnable() {
@Override
public void run() {
proxy.saveUser();
}
}).start();
/* HelloServiceImpl helloService = new HelloServiceImpl();
HelloServiceImpl helloService1 = (HelloServiceImpl) proxyContext.createProxy(helloService);
helloService1.hello();*/
}
}

打印结果

线程main: start transaction
main save user
main add
线程main: commit transaction
线程Thread-1: start transaction
Thread-1 save user
Thread-1 add
com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException: Duplicate entry ‘4’ for key ‘PRIMARY’
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at com.mysql.jdbc.Util.handleNewInstance(Util.java:425)
at com.mysql.jdbc.Util.getInstance(Util.java:408)
at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:935)
at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3973)
at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3909)
at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2527)
at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2680)
at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2487)
at com.mysql.jdbc.PreparedStatement.executeInternal(PreparedStatement.java:1858)
at com.mysql.jdbc.PreparedStatement.executeUpdateInternal(PreparedStatement.java:2079)
at com.mysql.jdbc.PreparedStatement.executeUpdateInternal(PreparedStatement.java:2013)
at com.mysql.jdbc.PreparedStatement.executeLargeUpdate(PreparedStatement.java:5104)
at com.mysql.jdbc.PreparedStatement.executeUpdate(PreparedStatement.java:1998)
at com.alibaba.druid.pool.DruidPooledPreparedStatement.executeUpdate(DruidPooledPreparedStatement.java:253)
at com.parallel.stomp.test.aop.UserMapper.add(UserMapper.java:19)
at com.parallel.stomp.test.aop.UserServiceImpl.saveUser(UserServiceImpl.java:15)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at com.parallel.stomp.test.aop.JdkTransactionProxy1.invoke(JdkTransactionProxy.java:43)atcom.sun.proxy.1.invoke(JdkTransactionProxy.java:43) at com.sun.proxy.1.invoke(JdkTransactionProxy.java:43)atcom.sun.proxy.Proxy0.saveUser(Unknown Source)
at com.parallel.stomp.test.aop.TestProxy$1.run(TestProxy.java:16)
at java.lang.Thread.run(Thread.java:745)
线程Thread-1: commit transaction

后记

最近在研究spring事务的底层原理，为了加深对spring事务的认知写了一个小的demo，由于本人能力有限，文中难免有错误之处，欢迎指正。