Cat 跨线程之 ForkedTransaction 用法和原理分析

叁歲伎倆 2021-11-04 10:10 334阅读 0赞

package com.dianping.cat.message.internal;
    
    import com.dianping.cat.Cat;
    import com.dianping.cat.message.ForkedTransaction;
    import com.dianping.cat.message.Message;
    import com.dianping.cat.message.Transaction;
    import org.junit.After;
    import org.junit.Before;
    import org.junit.Test;
    import org.unidal.helper.Threads;
    
    import java.io.File;
    import java.util.concurrent.TimeUnit;
    
    public class MultiThreadingTest {
        @After
        public void after() {
        }
    
        @Before
        public void before() {
            Cat.initialize(new File("/data/appdatas/cat/client.xml"));
        }
    
        @Test
        public void testForkedTransaction() throws Exception {
            Transaction t = Cat.newTransaction("ForkedRoot", "Root");
            ForkedTransaction t1 = Cat.newForkedTransaction("ForkedChild", "Child1"); //在当前消息树中埋点桥接节点
            Threads.forGroup().start(new TimedThread(t1, 100)); //与子线程共享这个对象
            TimeUnit.MILLISECONDS.sleep(200);
            t.setStatus(Message.SUCCESS);
            t.complete();
        }
    
        static class TimedThread extends Thread {
            private ForkedTransaction m_transaction;
    
            private int m_timeout;
    
            public TimedThread(ForkedTransaction t, int timeout) {
                m_transaction = t;
                m_timeout = timeout;
            }
    
            @Override
            public void run() {
                m_transaction.fork();
                try {
                    TimeUnit.MILLISECONDS.sleep(m_timeout);
                    Cat.logEvent("TimedThread", "Timeout." + m_timeout);
                    m_transaction.setStatus(Message.SUCCESS);
                } catch (Exception e) {
                    Cat.logError(e);
                    m_transaction.setStatus(e);
                } finally {
                    m_transaction.complete();
                }
            }
        }
    }

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

模型图

![aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNDkzNDY4Ny0xMDUzNTI4NzEucG5n][]

代码和流程分析

new一个ForkT在干嘛？  
1.初始化当前线程的消息树根节点消息Id  
2.创建子线程中的事务，并设置自己的根节点和父节点的消息Id  
3.在消息树的当前节点加入一个额外的event节点，并且把子线程的msgId记录下来

![aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTExNzg5MC05MjAwNTA4NzUucG5n][]

linkasRunAway的具体逻辑

![aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTIwMDk1Mi0xMzk4NzU0Mzc0LnBuZw][]

子线程的fork在干嘛？

1.在子线程中初始化本线程的消息树

2. 把forkT的消息id作为子线程的根节点消息id，这样就可以通过runway event节点找到子线程的消息树

![aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTIzODA5My03MzUwNzc5NDYucG5n][]

[aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]: /images/20211104/0cea4f26862644c98ee72d2c0464173a.png
[aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNDkzNDY4Ny0xMDUzNTI4NzEucG5n]: /images/20211104/7a3e5563b58a4983868ee01cd660bbf0.png
[aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTExNzg5MC05MjAwNTA4NzUucG5n]: /images/20211104/8b188577537c46c388d0a1dc560fe898.png
[aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTIwMDk1Mi0xMzk4NzU0Mzc0LnBuZw]: /images/20211104/8666965463bc4358b08d4959e20080b0.png
[aHR0cHM6Ly9pbWFnZXMyMDE3LmNuYmxvZ3MuY29tL2Jsb2cvMTI1ODUyNC8yMDE3MTEvMTI1ODUyNC0yMDE3MTExNTExNTIzODA5My03MzUwNzc5NDYucG5n]: /images/20211104/eae9a7c667274073843a3a9df9f83dcc.png