HelloWorld
登录

TCC

Wesley13

2021-10-11

1381 0 0

1. 概述

本文分享 Dubbo 支持。

TCC-Transaction 通过 Dubbo 隐式传参的功能，避免自己对业务代码的入侵。可能有同学不太理解为什么说 TCC-Transaction 对业务代码有一定的入侵性，一起来看个代码例子：

1public interface CapitalTradeOrderService {
2    String record(TransactionContext transactionContext, CapitalTradeOrderDto tradeOrderDto);
3}
  • 代码来自 tcc-transaction-http-sample 。声明远程调用时，增加了参数 TransactionContext。当然你也可以通过自己使用的远程调用框架做一定封装，避免入侵。

如下是对 Dubbo 封装了后，Dubbo Service 方法的例子：

1public interface CapitalTradeOrderService {
2    @Compensable
3    String record(CapitalTradeOrderDto tradeOrderDto);
4}
  • 代码来自 http-transaction-dubbo-sample 。是不是不需要传入参数 TransactionContext。当然，注解是肯定需要的，否则 TCC-Transaction 怎么知道哪些方法是 TCC 方法。

TCC-Transaction 通过 Dubbo Proxy 的机制，实现 @Compensable 属性自动生成，增加开发体验，也避免出错。

Dubbo 支持( Maven 项目 tcc-transaction-dubbo ) 整体代码结构如下：

  • proxy
  • context

我们分成两个小节分享这两个包实现的功能。

笔者暂时对 Dubbo 了解的不够深入，如果有错误的地方，还烦请指出，谢谢。

你行好事会因为得到赞赏而愉悦
同理，开源项目贡献者会因为 Star 而更加有动力
为 TCC-Transaction 点赞！传送门

ps：笔者假设你已经阅读过《tcc-transaction 官方文档 —— 使用指南1.2.x》。

2. Dubbo 代理

将 Dubbo Service 方法上的注解 @Compensable ，自动生成注解的confirmMethod、cancelMethod、transactionContextEditor 属性，例子代码如下：

1@Compensable(propagation=Propagation.SUPPORTS, confirmMethod="record", cancelMethod="record", transactionContextEditor=DubboTransactionContextEditor.class)
2public String record(RedPacketTradeOrderDto paramRedPacketTradeOrderDto) {
3    // ... 省略代码
4}
  • 该代码通过 Javassist 生成的 Proxy 代码的示例。
  • propagation=Propagation.SUPPORTS ：支持当前事务，如果当前没有事务，就以非事务方式执行。为什么不使用 REQUIRED ？如果使用 REQUIRED 事务传播级别，事务恢复重试时，会发起新的事务。
  • confirmMethod、cancelMethod 使用和 try 方法相同方法名：本地发起远程服务 TCC confirm / cancel 阶段，调用相同方法进行事务的提交或回滚。远程服务的 CompensableTransactionInterceptor 会根据事务的状态是 CONFIRMING / CANCELLING 来调用对应方法。
  • transactionContextEditor=DubboTransactionContextEditor.class，使用 Dubbo 事务上下文编辑器，在「3. Dubbo 事务上下文编辑器」详细分享。

Dubbo Service Proxy 提供了两种生成方式：

  • JavassistProxyFactory，基于 Javassist 方式
  • JdkProxyFactory，基于 JDK 动态代理机制

这块内容我们不拓展开，感兴趣的同学点击如下文章：

Dubbo 的 Invoker 模型是非常关键的概念，看下图：

2.1 JavassistProxyFactory

2.1.1 Javassist

Javassist 是一个开源的分析、编辑和创建 Java 字节码的类库。通过使用Javassist 对字节码操作可以实现动态 ”AOP” 框架。

关于 Java 字节码的处理，目前有很多工具，如 bcel，asm( cglib只是对asm又封装了一层 )。不过这些都需要直接跟虚拟机指令打交道。

Javassist 的主要的优点，在于简单，而且快速，直接使用 Java 编码的形式，而不需要了解虚拟机指令，就能动态改变类的结构，或者动态生成类。

2.1.2 TccJavassistProxyFactory

org.mengyun.tcctransaction.dubbo.proxy.javassist.TccJavassistProxyFactory，TCC Javassist 代理工厂。实现代码如下：

1public class TccJavassistProxyFactory extends JavassistProxyFactory {
2    @SuppressWarnings("unchecked")
3    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
4        return (T) TccProxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
5    }
6}
  • 项目启动时，调用 TccJavassistProxyFactory#getProxy(...) 方法，生成 Dubbo Service 调用 Proxy。
  • com.alibaba.dubbo.rpc.proxy.InvokerInvocationHandler，Dubbo 调用处理器，点击连接查看代码。

2.1.3 TccProxy & TccClassGenerator

org.mengyun.tcctransaction.dubbo.proxy.javassist.TccProxy，TCC Proxy 工厂，生成 Dubbo Service 调用 Proxy 。笔者认为，TccProxy 改成 TccProxyFactory 更合适，原因在下文。

org.mengyun.tcctransaction.dubbo.proxy.javassist.TccClassGenerator，TCC 类代码生成器，基于 Javassist 实现。

案例

一个 Dubbo Service，TccProxy 会动态生成两个类：

  • Dubbo Service 调用 Proxy
  • Dubbo Service 调用 ProxyFactory，生成对应的 Dubbo Service Proxy

例如 Dubbo Service 接口如下：

1public interface RedPacketTradeOrderService {
2    @Compensable
3    String record(RedPacketTradeOrderDto tradeOrderDto);
4}

生成 Dubbo Service 调用 ProxyFactory 如下 ：

1public class TccProxy3 extends TccProxy implements TccClassGenerator.DC {
2  public Object newInstance(InvocationHandler paramInvocationHandler) {
3    return new proxy3(paramInvocationHandler);
4  }
5}
  • TccProxy 提供 #newInstance(handler) 方法，创建 Proxy，所以笔者认为，TccProxy 改成 TccProxyFactory 更合适。
  • org.mengyun.tcctransaction.dubbo.proxy.javassist.TccClassGenerator.DC 动态生成类标记，标记该类由 TccClassGenerator 生成的。

生成 Dubbo Service 调用 Proxy 如下 ：

1public class proxy3 implements TccClassGenerator.DC, RedPacketTradeOrderService, EchoService {
2    
3    public static Method[] methods;
4    private InvocationHandler handler;
5    public proxy3() {}
6    public proxy3(InvocationHandler paramInvocationHandler) {
7        this.handler = paramInvocationHandler;
8    }
9    @Compensable(propagation = Propagation.SUPPORTS, confirmMethod = "record", cancelMethod = "record", transactionContextEditor = DubboTransactionContextEditor.class)
10    public String record(RedPacketTradeOrderDto paramRedPacketTradeOrderDto) {
11        Object[] arrayOfObject = new Object[1];
12        arrayOfObject[0] = paramRedPacketTradeOrderDto;
13        Object localObject = this.handler.invoke(this, methods[0], arrayOfObject);
14        return (String) localObject;
15    }
16    public Object $echo(Object paramObject) {
17        Object[] arrayOfObject = new Object[1];
18        arrayOfObject[0] = paramObject;
19        Object localObject = this.handler.invoke(this, methods[1], arrayOfObject);
20        return (Object) localObject;
21    }
22}
  • com.alibaba.dubbo.rpc.service.EchoService，Dubbo Service 回声服务接口，用于服务健康检查，Dubbo Service 默认自动实现该接口，点击连接查看代码。
  • org.mengyun.tcctransaction.dubbo.proxy.javassist.TccClassGenerator.DC 动态生成类标记，标记该类由 TccClassGenerator 生成的。

🦅实现

调用 TccProxy#getProxy(...) 方法，获得 TCC Proxy 工厂，实现代码如下：

1  1: // 【TccProxy.java】
2  2: public static TccProxy getProxy(ClassLoader cl, Class<?>... ics) {
3  3:     // 校验接口超过上限
4  4:     if (ics.length > 65535) {
5  5:         throw new IllegalArgumentException("interface limit exceeded");
6  6:     }
7  7: 
8  8:     // use interface class name list as key.
9  9:     StringBuilder sb = new StringBuilder();
10 10:     for (Class<?> ic : ics) {
11 11:         String itf = ic.getName();
12 12:         // 校验是否为接口
13 13:         if (!ic.isInterface()) {
14 14:             throw new RuntimeException(itf + " is not a interface.");
15 15:         }
16 16:         // 加载接口类
17 17:         Class<?> tmp = null;
18 18:         try {
19 19:             tmp = Class.forName(itf, false, cl);
20 20:         } catch (ClassNotFoundException ignored) {
21 21:         }
22 22:         if (tmp != ic) { // 加载接口类失败
23 23:             throw new IllegalArgumentException(ic + " is not visible from class loader");
24 24:         }
25 25:         sb.append(itf).append(';');
26 26:     }
27 27:     String key = sb.toString();
28 28: 
29 29:     // get cache by class loader.
30 30:     Map<String, Object> cache;
31 31:     synchronized (ProxyCacheMap) {
32 32:         cache = ProxyCacheMap.get(cl);
33 33:         if (cache == null) {
34 34:             cache = new HashMap<String, Object>();
35 35:             ProxyCacheMap.put(cl, cache);
36 36:         }
37 37:     }
38 38: 
39 39:     // 获得 TccProxy 工厂
40 40:     TccProxy proxy = null;
41 41:     synchronized (cache) {
42 42:         do {
43 43:             // 从缓存中获取 TccProxy 工厂
44 44:             Object value = cache.get(key);
45 45:             if (value instanceof Reference<?>) {
46 46:                 proxy = (TccProxy) ((Reference<?>) value).get();
47 47:                 if (proxy != null) {
48 48:                     return proxy;
49 49:                 }
50 50:             }
51 51:             // 缓存中不存在，设置生成 TccProxy 代码标记。创建中时，其他创建请求等待，避免并发。
52 52:             if (value == PendingGenerationMarker) {
53 53:                 try {
54 54:                     cache.wait();
55 55:                 } catch (InterruptedException ignored) {
56 56:                 }
57 57:             } else {
58 58:                 cache.put(key, PendingGenerationMarker);
59 59:                 break;
60 60:             }
61 61:         }
62 62:         while (true);
63 63:     }
64 64: 
65 65:     long id = PROXY_CLASS_COUNTER.getAndIncrement();
66 66:     String pkg = null;
67 67:     TccClassGenerator ccp = null; // proxy class generator
68 68:     TccClassGenerator ccm = null; // proxy factory class generator
69 69:     try {
70 70:         // 创建 Tcc class 代码生成器
71 71:         ccp = TccClassGenerator.newInstance(cl);
72 72: 
73 73:         Set<String> worked = new HashSet<String>(); // 已处理方法签名集合。key：方法签名
74 74:         List<Method> methods = new ArrayList<Method>(); // 已处理方法集合。
75 75: 
76 76:         // 处理接口
77 77:         for (Class<?> ic : ics) {
78 78:             // 非 public 接口，使用接口包名
79 79:             if (!Modifier.isPublic(ic.getModifiers())) {
80 80:                 String npkg = ic.getPackage().getName();
81 81:                 if (pkg == null) {
82 82:                     pkg = npkg;
83 83:                 } else {
84 84:                     if (!pkg.equals(npkg)) { // 实现了两个非 public 的接口，
85 85:                         throw new IllegalArgumentException("non-public interfaces from different packages");
86 86:                     }
87 87:                 }
88 88:             }
89 89:             // 添加接口
90 90:             ccp.addInterface(ic);
91 91:             // 处理接口方法
92 92:             for (Method method : ic.getMethods()) {
93 93:                 // 添加方法签名到已处理方法签名集合
94 94:                 String desc = ReflectUtils.getDesc(method);
95 95:                 if (worked.contains(desc)) {
96 96:                     continue;
97 97:                 }
98 98:                 worked.add(desc);
99 99:                 // 生成接口方法实现代码
100100:                 int ix = methods.size();
101101:                 Class<?> rt = method.getReturnType();
102102:                 Class<?>[] pts = method.getParameterTypes();
103103:                 StringBuilder code = new StringBuilder("Object[] args = new Object[").append(pts.length).append("];");
104104:                 for (int j = 0; j < pts.length; j++) {
105105:                     code.append(" args[").append(j).append("] = ($w)$").append(j + 1).append(";");
106106:                 }
107107:                 code.append(" Object ret = handler.invoke(this, methods[").append(ix).append("], args);");
108108:                 if (!Void.TYPE.equals(rt)) {
109109:                     code.append(" return ").append(asArgument(rt, "ret")).append(";");
110110:                 }
111111:                 methods.add(method);
112112:                 // 添加方法
113113:                 Compensable compensable = method.getAnnotation(Compensable.class);
114114:                 if (compensable != null) {
115115:                     ccp.addMethod(true, method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());
116116:                 } else {
117117:                     ccp.addMethod(false, method.getName(), method.getModifiers(), rt, pts, method.getExceptionTypes(), code.toString());
118118:                 }
119119:             }
120120:         }
121121: 
122122:         // 设置包路径
123123:         if (pkg == null) {
124124:             pkg = PACKAGE_NAME;
125125:         }
126126: 
127127:         // create ProxyInstance class.
128128:         // 设置类名
129129:         String pcn = pkg + ".proxy" + id;
130130:         ccp.setClassName(pcn);
131131:         // 添加静态属性 methods
132132:         ccp.addField("public static java.lang.reflect.Method[] methods;");
133133:         // 添加属性 handler
134134:         ccp.addField("private " + InvocationHandler.class.getName() + " handler;");
135135:         // 添加构造方法，参数 handler
136136:         ccp.addConstructor(Modifier.PUBLIC, new Class<?>[]{InvocationHandler.class}, new Class<?>[0], "handler=$1;");
137137:         // 添加构造方法，参数 空
138138:         ccp.addDefaultConstructor();
139139:         // 生成类
140140:         Class<?> clazz = ccp.toClass();
141141:         // 设置静态属性 methods
142142:         clazz.getField("methods").set(null, methods.toArray(new Method[0]));
143143: 
144144:         // create TccProxy class.
145145:         // 创建 Tcc class 代码生成器
146146:         ccm = TccClassGenerator.newInstance(cl);
147147:         // 设置类名
148148:         String fcn = TccProxy.class.getName() + id;
149149:         ccm.setClassName(fcn);
150150:         // 添加构造方法，参数 空
151151:         ccm.addDefaultConstructor();
152152:         // 设置父类为 TccProxy.class
153153:         ccm.setSuperClass(TccProxy.class);
154154:         // 添加方法 #newInstance(handler)
155155:         ccm.addMethod("public Object newInstance(" + InvocationHandler.class.getName() + " h){ return new " + pcn + "($1); }");
156156:         // 生成类
157157:         Class<?> pc = ccm.toClass();
158158:         // 创建 TccProxy 对象
159159:         proxy = (TccProxy) pc.newInstance();
160160:     } catch (RuntimeException e) {
161161:         throw e;
162162:     } catch (Exception e) {
163163:         throw new RuntimeException(e.getMessage(), e);
164164:     } finally {
165165:         // release TccClassGenerator
166166:         if (ccp != null) {
167167:             ccp.release();
168168:         }
169169:         if (ccm != null) {
170170:             ccm.release();
171171:         }
172172:         // 唤醒缓存 wait
173173:         synchronized (cache) {
174174:             if (proxy == null) {
175175:                 cache.remove(key);
176176:             } else {
177177:                 cache.put(key, new WeakReference<TccProxy>(proxy));
178178:             }
179179:             cache.notifyAll();
180180:         }
181181:     }
182182:     return proxy;
183183: }

  • 第 3 至 7 行 ：校验接口超过上限。

  • 第 8 至 27 行 ：使用接口集合类名以 ; 分隔拼接，作为 Proxy 的唯一标识。例如 key=org.mengyun.tcctransaction.sample.dubbo.redpacket.api.RedPacketAccountService;com.alibaba.dubbo.rpc.service.EchoService;。

  • 第 29 至 37 行 ：获得 Proxy 对应的 ClassLoader。这里我们看下静态属性 ProxyCacheMap的定义：

    1/**
2* Proxy 对象缓存
3* key ：ClassLoader
4* value.key ：Tcc Proxy 标识。使用 Tcc Proxy 实现接口名拼接
5* value.value ：Tcc Proxy 工厂对象
6*/
7private static final Map<ClassLoader, Map<String, Object>> ProxyCacheMap = new WeakHashMap<ClassLoader, Map<String, Object>>();

  • 第 39 至 63 行 ：一直获得 TCC Proxy 工厂直到成功。

  • 第 43 至 50 行 ：从缓存中获取 TCC Proxy 工厂。

  • 第 51 至 60 行 ：若缓存中不存在，设置正在生成 TccProxy 代码标记。创建中时，其他创建请求等待，避免并发。

  • 第 65 行 ：PROXY_CLASS_COUNTER，Proxy Class 计数，用于生成 Proxy 类名自增。代码如下：

    private static final AtomicLong PROXY_CLASS_COUNTER = new AtomicLong(0);

  • 第 66 至 67 行

    • ccm，生成 Dubbo Service 调用 ProxyFactory 的代码生成器
    • ccp，生成 Dubbo Service 调用 Proxy 的代码生成器

  • 第 70 至 142 行 ：生成 Dubbo Service 调用 Proxy 的代码。

  • 第 70 至 71 行 ：调用 TccClassGenerator#newInstance(loader) 方法， 创建生成 Dubbo Service 调用 Proxy 的代码生成器。实现代码如下：

  • 1  // TccClassGenerator.java
2  public final class TccClassGenerator {
3      
4      /**
5       * CtClass hash 集合
6       * key：类名
7       */
8      private ClassPool mPool;
9      
10      public static TccClassGenerator newInstance(ClassLoader loader) {
11          return new TccClassGenerator(getClassPool(loader));
12      }
13      
14      private TccClassGenerator(ClassPool pool) {
15          mPool = pool;
16      }
17  }

ClassPool 是一个 CtClass 对象的 hash 表，类名做为 key 。ClassPool 的#get(key) 搜索 hash 表找到与指定 key 关联的 CtClass 对象。如果没有找到 CtClass 对象，#get(key) 读一个类文件构建新的 CtClass 对象，它是被记录在 hash 表中然后返回这个对象。

  • 第 76 至 120 行，处理接口。

  • 第 79 至 88 行，生成类的包名。

  • 第 89 至 90 行，调用 TccClassGenerator#addInterface(cl) 方法，添加生成类的接口( Dubbo Service 接口 )。实现代码如下：

    1/**
2* 生成类的接口集合
3*/
4private Set<String> mInterfaces;
5public TccClassGenerator addInterface(Class<?> cl) {
6   return addInterface(cl.getName());
7}
8public TccClassGenerator addInterface(String cn) {
9   if (mInterfaces == null) {
10       mInterfaces = new HashSet<String>();
11   }
12   mInterfaces.add(cn);
13   return this;
14}

  • 第 93 至 98 行，添加方法签名到已处理方法签名集合。多个接口可能存在相同的接口方法，跳过相同的方法，避免冲突。

  • 第 99 至 110 行，生成 Dubbo Service 调用实现代码。案例代码如下：

  • 第 112 至 118 行 ：调用 TccClassGenerator#addMethod(...) 方法，添加生成的方法。实现代码如下：

    1/**
2* 生成类的方法代码集合
3*/
4private List<String> mMethods;
5/**
6* 带 @Compensable 方法代码集合
7*/
8private Set<String> compensableMethods = new HashSet<String>();
9public TccClassGenerator addMethod(boolean isCompensableMethod, String name, int mod, Class<?> rt, Class<?>[] pts, Class<?>[] ets, String body) {
10   // 拼接方法
11   StringBuilder sb = new StringBuilder();
12   sb.append(modifier(mod)).append(' ').append(ReflectUtils.getName(rt)).append(' ').append(name);
13   sb.append('(');
14   for (int i = 0; i < pts.length; i++) {
15       if (i > 0)
16           sb.append(',');
17       sb.append(ReflectUtils.getName(pts[i]));
18       sb.append(" arg").append(i);
19   }
20   sb.append(')');
21   if (ets != null && ets.length > 0) {
22       sb.append(" throws ");
23       for (int i = 0; i < ets.length; i++) {
24           if (i > 0)
25               sb.append(',');
26           sb.append(ReflectUtils.getName(ets[i]));
27       }
28   }
29   sb.append('{').append(body).append('}');
30   // 是否有 @Compensable 注解
31   if (isCompensableMethod) {
32       compensableMethods.add(sb.toString());
33   }
34   return addMethod(sb.toString());
35}
36public TccClassGenerator addMethod(String code) {
37   if (mMethods == null) {
38       mMethods = new ArrayList<String>();
39   }
40   mMethods.add(code);
41   return this;
42}

  • 第 122 至 130 行，生成类名( 例如，org.mengyun.tcctransaction.dubbo.proxy.javassist.proxy3 )，并调用TccClassGenerator#setClassName(...) 方法，设置类名。实现代码如下：

  • 1  /**
2  * 生成类的类名
3  */
4  private String mClassName;
5  public TccClassGenerator setClassName(String name) {
6     mClassName = name;
7     return this;
8  }

  • 第 131 至 134 行，调用 TccClassGenerator#addField(...) 方法，添加静态属性methods ( Dubbo Service 方法集合 )和属性 handler ( Dubbo InvocationHandler )。实现代码如下：

    1/**
2* 生成类的属性集合
3*/
4private List<String> mFields;
5public TccClassGenerator addField(String code) {
6   if (mFields == null) {
7       mFields = new ArrayList<String>();
8   }
9   mFields.add(code);
10   return this;
11}

  • 第 135 至 136 行，调用 TccClassGenerator#addConstructor(...) 方法，添加参数为 handler 的构造方法。实现代码如下：

  • 1  /**
2  * 生成类的非空构造方法代码集合
3  */
4  private List<String> mConstructors;
5  public TccClassGenerator addConstructor(int mod, Class<?>[] pts, Class<?>[] ets, String body) {
6     // 构造方法代码
7     StringBuilder sb = new StringBuilder();
8     sb.append(modifier(mod)).append(' ').append(SIMPLE_NAME_TAG);
9     sb.append('(');
10     for (int i = 0; i < pts.length; i++) {
11         if (i > 0)
12             sb.append(',');
13         sb.append(ReflectUtils.getName(pts[i]));
14         sb.append(" arg").append(i);
15     }
16     sb.append(')');
17     if (ets != null && ets.length > 0) {
18         sb.append(" throws ");
19         for (int i = 0; i < ets.length; i++) {
20             if (i > 0)
21                 sb.append(',');
22             sb.append(ReflectUtils.getName(ets[i]));
23         }
24     }
25     sb.append('{').append(body).append('}');
26     //
27     return addConstructor(sb.toString());
28  }
29  public TccClassGenerator addConstructor(String code) {
30     if (mConstructors == null) {
31         mConstructors = new LinkedList<String>();
32     }
33     mConstructors.add(code);
34     return this;
35  }
36  public TccClassGenerator addConstructor(String code) {
37     if (mConstructors == null) {
38         mConstructors = new LinkedList<String>();
39     }
40     mConstructors.add(code);
41     return this;
42  }

  • 第 137 至 138 行，调用 TccClassGenerator#addDefaultConstructor() 方法，添加默认空构造方法。实现代码如下：

    1/**
2* 默认空构造方法
3*/
4private boolean mDefaultConstructor = false;
5public TccClassGenerator addDefaultConstructor() {
6   mDefaultConstructor = true;
7   return this;
8}

  • 第 139 行，调用 TccClassGenerator#toClass() 方法，生成类。实现代码如下：

         基于 Javassist 生成类。这里不做拓展解释，配合《Java学习之javassist》一起理解。

  • 第 18 行，添加org.mengyun.tcctransaction.dubbo.proxy.javassist.TccClassGenerator.DC动态生成类标记，标记该类由 TccClassGenerator 生成的。

  • 第 34 至 50 行，设置 @Compensable 默认属性。

  • 1   1: public Class<?> toClass() {
2   2:    // mCtc 非空时，进行释放；下面会进行创建 mCtc
3   3:    if (mCtc != null) {
4   4:        mCtc.detach();
5   5:    }
6   6:    long id = CLASS_NAME_COUNTER.getAndIncrement();
7   7:    try {
8   8:        CtClass ctcs = mSuperClass == null ? null : mPool.get(mSuperClass);
9   9:        if (mClassName == null) { // 类名
10  10:            mClassName = (mSuperClass == null || javassist.Modifier.isPublic(ctcs.getModifiers())
11  11:                    ? TccClassGenerator.class.getName() : mSuperClass + "$sc") + id;
12  12:        }
13  13:        // 创建 mCtc
14  14:        mCtc = mPool.makeClass(mClassName);
15  15:        if (mSuperClass != null) { // 继承类
16  16:            mCtc.setSuperclass(ctcs);
17  17:        }
18  18:        mCtc.addInterface(mPool.get(DC.class.getName())); // add dynamic class tag.
19  19:        if (mInterfaces != null) { // 实现接口集合
20  20:            for (String cl : mInterfaces) {
21  21:                mCtc.addInterface(mPool.get(cl));
22  22:            }
23  23:        }
24  24:        if (mFields != null) { // 属性集合
25  25:            for (String code : mFields) {
26  26:                mCtc.addField(CtField.make(code, mCtc));
27  27:            }
28  28:        }
29  29:        if (mMethods != null) { // 方法集合
30  30:            for (String code : mMethods) {
31  31:                if (code.charAt(0) == ':') {
32  32:                    mCtc.addMethod(CtNewMethod.copy(getCtMethod(mCopyMethods.get(code.substring(1))), code.substring(1, code.indexOf('(')), mCtc, null));
33  33:                } else {
34  34:                    CtMethod ctMethod = CtNewMethod.make(code, mCtc);
35  35:                    if (compensableMethods.contains(code)) {
36  36:                        // 设置 @Compensable 属性
37  37:                        ConstPool constpool = mCtc.getClassFile().getConstPool();
38  38:                        AnnotationsAttribute attr = new AnnotationsAttribute(constpool, AnnotationsAttribute.visibleTag);
39  39:                        Annotation annot = new Annotation("org.mengyun.tcctransaction.api.Compensable", constpool);
40  40:                        EnumMemberValue enumMemberValue = new EnumMemberValue(constpool);
41  41:                        enumMemberValue.setType("org.mengyun.tcctransaction.api.Propagation");
42  42:                        enumMemberValue.setValue("SUPPORTS");
43  43:                        annot.addMemberValue("propagation", enumMemberValue);
44  44:                        annot.addMemberValue("confirmMethod", new StringMemberValue(ctMethod.getName(), constpool));
45  45:                        annot.addMemberValue("cancelMethod", new StringMemberValue(ctMethod.getName(), constpool));
46  46:                        ClassMemberValue classMemberValue = new ClassMemberValue("org.mengyun.tcctransaction.dubbo.context.DubboTransactionContextEditor", constpool);
47  47:                        annot.addMemberValue("transactionContextEditor", classMemberValue);
48  48:                        attr.addAnnotation(annot);
49  49:                        ctMethod.getMethodInfo().addAttribute(attr);
50  50:                    }
51  51:                    mCtc.addMethod(ctMethod);
52  52:                }
53  53:            }
54  54:        }
55  55:        if (mDefaultConstructor) { // 空参数构造方法
56  56:            mCtc.addConstructor(CtNewConstructor.defaultConstructor(mCtc));
57  57:        }
58  58:        if (mConstructors != null) { // 带参数构造方法
59  59:            for (String code : mConstructors) {
60  60:                if (code.charAt(0) == ':') {
61  61:                    mCtc.addConstructor(CtNewConstructor.copy(getCtConstructor(mCopyConstructors.get(code.substring(1))), mCtc, null));
62  62:                } else {
63  63:                    String[] sn = mCtc.getSimpleName().split("\\$+"); // inner class name include $.
64  64:                    mCtc.addConstructor(CtNewConstructor.make(code.replaceFirst(SIMPLE_NAME_TAG, sn[sn.length - 1]), mCtc));
65  65:                }
66  66:            }
67  67:        }
68  68: //            mCtc.debugWriteFile("/Users/yunai/test/" + mCtc.getSimpleName().replaceAll(".", "/") + ".class");
69  69:        // 生成
70  70:        return mCtc.toClass();
71  71:    } catch (RuntimeException e) {
72  72:        throw e;
73  73:    } catch (NotFoundException e) {
74  74:        throw new RuntimeException(e.getMessage(), e);
75  75:    } catch (CannotCompileException e) {
76  76:        throw new RuntimeException(e.getMessage(), e);
77  77:    }
78  78: }

  • 第 141 至 142 行，设置 Dubbo Service 方法集合设置到静态属性 methods 上。

  • 第 144 至 157 行，生成 Dubbo Service 调用 Proxy 工厂的代码。

  • 第 146 行，调用 TccClassGenerator#newInstance(loader) 方法， 创建生成 Dubbo Service 调用 Proxy 工厂 的代码生成器。

  • 第 147 至 149 行，生成类名( 例如，org.mengyun.tcctransaction.dubbo.proxy.javassist.TccProxy3 )，并调用TccClassGenerator#setClassName(...) 方法，设置类名。

  • 第 150 至 151 行，调用 TccClassGenerator#addDefaultConstructor() 方法，添加默认空构造方法。

  • 第 152 至 153 行，调用 TccClassGenerator#mSuperClass() 方法，设置继承父类**TccProxy**。实现代码如下：

    1/**
2* 生成类的父类名字
3*/
4private String mSuperClass;
5public TccClassGenerator setSuperClass(Class<?> cl) {
6   mSuperClass = cl.getName();
7   return this;
8}

  • 第 154 至 155 行，调用 TccClassGenerator#addInterface(cl) 方法，添加生成 Proxy 实现代码的方法。代码案例如下：

  • 1  public Object newInstance(InvocationHandler paramInvocationHandler) {
2       return new proxy3(paramInvocationHandler);
3  }

  • 第 156 至 157 行，调用 TccClassGenerator#toClass() 方法，生成类。

  • 第 159 行，调用 TccProxy#newInstance() 方法，创建 Proxy 。实现代码如下：

    1/**
2* get instance with default handler.
3*
4* @return instance.
5*/
6public Object newInstance() {
7   return newInstance(THROW_UNSUPPORTED_INVOKER);
8}
9/**
10* get instance with special handler.
11*
12* @return instance.
13*/
14abstract public Object newInstance(InvocationHandler handler);

#newInstance(handler)，抽象方法，上面第 154 至 155 行生成。TccJavassistProxyFactory 调用该方法，获得 Proxy 。

  • 第 165 至 171 行，释放 TccClassGenerator 。实现代码如下：

    1public void release() {
2   if (mCtc != null) {
3       mCtc.detach();
4   }
5   if (mInterfaces != null) {
6       mInterfaces.clear();
7   }
8   if (mFields != null) {
9       mFields.clear();
10   }
11   if (mMethods != null) {
12       mMethods.clear();
13   }
14   if (mConstructors != null) {
15       mConstructors.clear();
16   }
17   if (mCopyMethods != null) {
18       mCopyMethods.clear();
19   }
20   if (mCopyConstructors != null) {
21       mCopyConstructors.clear();
22   }
23}

  • 第 172 至 180 行，设置 Proxy 工厂缓存，并唤醒等待线程。

**ps：**代码比较多，收获会比较多，算是 Javassist 实战案例了。TCC-Transaction 作者在实现上述类，可能参考了 Dubbo 自带的实现：

2.1.4 配置 Dubbo Proxy

1// META-INF.dubbo/com.alibaba.dubbo.rpc.ProxyFactory
2tccJavassist=org.mengyun.tcctransaction.dubbo.proxy.javassist.TccJavassistProxyFactory
3// tcc-transaction-dubbo.xml
4<dubbo:provider proxy="tccJavassist"/>

目前 Maven 项目 tcc-transaction-dubbo 已经默认配置，引入即可。

2.2 JdkProxyFactory

2.2.1 JDK Proxy

《 Java JDK 动态代理（AOP）使用及实现原理分析》

2.2.2 TccJdkProxyFactory

org.mengyun.tcctransaction.dubbo.proxy.jd.TccJdkProxyFactory，TCC JDK 代理工厂。实现代码如下：

1public class TccJdkProxyFactory extends JdkProxyFactory {
2    @SuppressWarnings("unchecked")
3    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
4        T proxy = (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), interfaces, new InvokerInvocationHandler(invoker));
5        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), interfaces, new TccInvokerInvocationHandler(proxy, invoker));
6    }
7}
  • 项目启动时，调用 TccJavassistProxyFactory#getProxy(...) 方法，生成 Dubbo Service 调用 Proxy。
  • 第一次调用 Proxy#newProxyInstance(...) 方法，创建调用 Dubbo Service 服务的 Proxy。com.alibaba.dubbo.rpc.proxy.InvokerInvocationHandler，Dubbo 调用处理器，点击连接查看代码。
  • 第二次调用 Proxy#newProxyInstance(...) 方法，创建对调用 Dubbo Service 的 Proxy 的 Proxy。为什么会有两层 Proxy？答案在下节 TccInvokerInvocationHandler 。

2.2.3 TccInvokerInvocationHandler

org.mengyun.tcctransaction.dubbo.proxy.jdk.TccInvokerInvocationHandler，TCC 调用处理器，在调用 Dubbo Service 服务时，使用 ResourceCoordinatorInterceptor 拦截处理。实现代码如下：

1 1: public class TccInvokerInvocationHandler extends InvokerInvocationHandler {
2 2: 
3 3:     /**
4 4:      * proxy
5 5:      */
6 6:     private Object target;
7 7: 
8 8:     public TccInvokerInvocationHandler(Invoker<?> handler) {
9 9:         super(handler);
1010:     }
1111: 
1212:     public <T> TccInvokerInvocationHandler(T target, Invoker<T> invoker) {
1313:         super(invoker);
1414:         this.target = target;
1515:     }
1616: 
1717:     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
1818:         Compensable compensable = method.getAnnotation(Compensable.class);
1919:         if (compensable != null) {
2020:             // 设置 @Compensable 属性
2121:             if (StringUtils.isEmpty(compensable.confirmMethod())) {
2222:                 ReflectionUtils.changeAnnotationValue(compensable, "confirmMethod", method.getName());
2323:                 ReflectionUtils.changeAnnotationValue(compensable, "cancelMethod", method.getName());
2424:                 ReflectionUtils.changeAnnotationValue(compensable, "transactionContextEditor", DubboTransactionContextEditor.class);
2525:                 ReflectionUtils.changeAnnotationValue(compensable, "propagation", Propagation.SUPPORTS);
2626:             }
2727:             // 生成切面
2828:             ProceedingJoinPoint pjp = new MethodProceedingJoinPoint(proxy, target, method, args);
2929:             // 执行
3030:             return FactoryBuilder.factoryOf(ResourceCoordinatorAspect.class).getInstance().interceptTransactionContextMethod(pjp);
3131:         } else {
3232:             return super.invoke(target, method, args);
3333:         }
3434:     }
3535: 
3636: }

  • 第 18 至 26 行，设置带有 @Compensable 属性的默认属性。

  • 第 28 行，生成方法切面org.mengyun.tcctransaction.dubbo.proxy.jdk.MethodProceedingJoinPoint。实现代码如下：

    1public class MethodProceedingJoinPoint implements ProceedingJoinPoint, JoinPoint.StaticPart {
2    /**
3     * 代理对象
4     */
5    private Object proxy;
6    /**
7     * 目标对象
8     */
9    private Object target;
10    /**
11     * 方法
12     */
13    private Method method;
14    /**
15     * 参数
16     */
17    private Object[] args;
18    
19    @Override
20    public Object proceed() throws Throwable {
21        // Use reflection to invoke the method.
22        try {
23            ReflectionUtils.makeAccessible(method);
24            return method.invoke(target, args);
25        } catch (InvocationTargetException ex) {
26            // Invoked method threw a checked exception.
27            // We must rethrow it. The client won't see the interceptor.
28            throw ex.getTargetException();
29        } catch (IllegalArgumentException ex) {
30            throw new SystemException("Tried calling method [" +
31                    method + "] on target [" + target + "] failed", ex);
32        } catch (IllegalAccessException ex) {
33            throw new SystemException("Could not access method [" + method + "]", ex);
34        }
35    }
36    @Override
37    public Object proceed(Object[] objects) throws Throwable {
38        //        throw new UnsupportedOperationException(); // TODO 芋艿：疑问
39        return proceed();
40    }
41    
42    // ... 省略不重要的方法和对象
43}
    • 该类参考org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint 实现。
    • TODO【1】 proxy 和 target 是否保留一个即可？
    • 在切面处理完成后，调用 #proceed(...) 方法，进行远程 Dubbo Service 服务调用。
    • TODO【2】#proceed(objects) 抛出 throw new UnsupportedOperationException();。需要跟作者确认下。

  • 调用 ResourceCoordinatorAspect#interceptTransactionContextMethod(...) 方法，对方法切面拦截处理。为什么无需调用 CompensableTransactionAspect 切面？因为传播级别为 Propagation.SUPPORTS，不会发起事务。

2.2.4 配置 Dubbo Proxy

1// META-INF.dubbo/com.alibaba.dubbo.rpc.ProxyFactory
2tccJdk=org.mengyun.tcctransaction.dubbo.proxy.jdk.TccJdkProxyFactory
3// appcontext-service-dubbo.xml
4<dubbo:provider proxy="tccJdk"/>
5<dubbo:reference proxy="tccJdk" id="captialTradeOrderService"
6                     interface="org.mengyun.tcctransaction.sample.dubbo.capital.api.CapitalTradeOrderService" timeout="5000"/>
  • ProxyFactory 的 tccJdk 在 Maven 项 tcc-transaction-dubbo 已经声明。
  • 声明 dubbo:provider 的 proxy="tccJdk"。
  • 声明 dubbo:reference 的 proxy="tccJdk"，否则不生效。

3. Dubbo 事务上下文编辑器

org.mengyun.tcctransaction.dubbo.context.DubboTransactionContextEditor，Dubbo 事务上下文编辑器实现，实现代码如下：

1public class DubboTransactionContextEditor implements TransactionContextEditor {
2    @Override
3    public TransactionContext get(Object target, Method method, Object[] args) {
4        String context = RpcContext.getContext().getAttachment(TransactionContextConstants.TRANSACTION_CONTEXT);
5        if (StringUtils.isNotEmpty(context)) {
6            return JSON.parseObject(context, TransactionContext.class);
7        }
8        return null;
9    }
10    @Override
11    public void set(TransactionContext transactionContext, Object target, Method method, Object[] args) {
12        RpcContext.getContext().setAttachment(TransactionContextConstants.TRANSACTION_CONTEXT, JSON.toJSONString(transactionContext));
13    }
14}
  • 通过 Dubbo 的隐式传参的方式，避免在 Dubbo Service 接口上声明 TransactionContext 参数，对接口产生一定的入侵。
点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时，报错：原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等，这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

皕杰报表之UUID

​在我们用皕杰报表工具设计填报报表时，如何在新增行里自动增加id呢？能新增整数排序id吗？目前可以在新增行里自动增加id，但只能用uuid函数增加UUID编码，不能新增整数排序id。uuid函数说明：获取一个UUID，可以在填报表中用来创建数据ID语法：uuid()或uuid(sep)参数说明：sep布尔值，生成的uuid中是否包含分隔符''，缺省为

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段，为你的工作保驾护航

有空的时候，自己总结了几个代码段，在开发中也经常使用，谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )