1-fescar(seata)源码分析-server端

2019-02-19 本文已影响0人致虑

1-fescar源码分析-server端

一.官网介绍（以下截取至自官网）

https://github.com/alibaba/fescar/wiki/%E6%A6%82%E8%A7%88

1.1 定义一个分布式事务

首先，很自然的，我们可以把一个分布式事务理解成一个包含了若干分支事务的全局事务。全局事务的职责是协调其下管辖的分支事务达成一致，要么一起成功提交，要么一起失败回滚。此外，通常分支事务本身就是一个满足 ACID 的本地事务。这是我们对分布式事务结构的基本认识，与 XA 是一致的。
image.png

定义三个组件

image.png

Transaction Coordinator (TC)：事务协调器，维护全局事务的运行状态，负责协调并驱动全局事务的提交或回滚。
Transaction Manager (TM)：控制全局事务的边界，负责开启一个全局事务，并最终发起全局提交或全局回滚的决议。
Resource Manager (RM)：控制分支事务，负责分支注册、状态汇报，并接收事务协调器的指令，驱动分支（本地）事务的提交和回滚。

具体配合

image.png
具体流程

TM 向 TC 申请开启一个全局事务，全局事务创建成功并生成一个全局唯一的 XID。
XID 在微服务调用链路的上下文中传播。
RM 向 TC 注册分支事务，将其纳入 XID 对应全局事务的管辖。
TM 向 TC 发起针对 XID 的全局提交或回滚决议。
TC 调度 XID 下管辖的全部分支事务完成提交或回滚请求。

提交与回滚

image.png

二、主要流程（模块）概要

1.fescar-server：TC模块，server端启动：fescar的主服务，主要是netty服务端，接收TM的提交或者回滚消息以及向RM发送提交或者回滚消息指令等。
2.fescar-tm：TM模块，主要提供事务管理，向TC发送消息，触发事务的提交或者回滚。
3.rm-datasource: RM模块，数据库相关，覆盖rpc服务的执行逻辑，备份脚本执行前后结果，生成回滚脚本，向TC注册事务分支，执行正向脚本（逻辑）。
4.fescar-core：核心对象打包
5.fescar-spring：提供注解，保证对业务的非侵入性

1.服务启动，通过@GlobalTransactional执行到GlobalTransactionalInterceptor拦截器，进入对业务逻辑进行增强，执行TransactionalTemplate逻辑
2.TransactionalTemplate对全局事务的获取、执行业务逻辑、异常回滚、事务提交进行统筹，并同时通过服务的维持channel将每个步骤与server的netty进行交涉.
3.同时TransactionalTemplate在执行业务逻辑的同时，通过spring jdbc的覆盖，通过直接调用RM模块生成相关的备份及回滚脚本。
4.server接收到相关服务的消息后，触发相关的动作，并最终触发RM去进行数据库层面的操作。

三、（原理）源码分析

其实细看上面官网的介绍，基本能清楚fescar的大致思想。下面从源码角度出发一步一步的将流程以代码的形式读完。

3.1 demo

先看下官网的结构图：

image.png

项目中存在官方的example模块，里面就模拟了上图的相关流程：先回到本节主题：server

3.2.TC server

3.2.1.Server服务端启动

public class Server {

    private static final ThreadPoolExecutor WORKING_THREADS = new ThreadPoolExecutor(100, 500, 500, TimeUnit.SECONDS, new LinkedBlockingQueue(20000), new ThreadPoolExecutor.CallerRunsPolicy());
    public static void main(String[] args) throws IOException {

        // netty实现的一个简单的rpc服务端
        RpcServer rpcServer = new RpcServer(WORKING_THREADS);
        
        int port = 8091;
        if (args.length == 0) {
            rpcServer.setListenPort(port);
        }

        if (args.length > 0) {
            try {
                port = Integer.parseInt(args[0]);
            } catch (NumberFormatException e) {
                System.err.println("Usage: sh fescar-server.sh $LISTEN_PORT $PATH_FOR_PERSISTENT_DATA");
                System.exit(0);
            }
            rpcServer.setListenPort(port);
        }
        
        String dataDir = null;
        if (args.length > 1) {
            dataDir = args[1];
        }

        /**
         * SessionHolder主要负责事务信息存储，对应的ROOT_SESSION_MANAGER，ASYNC_COMMITTING_SESSION_MANAGER，RETRY_COMMITTING_SESSION_MANAGER，RETRY_ROLLBACKING_SESSION_MANAGER分别对应相应的文件存储到本地
         */
        SessionHolder.init(dataDir);

        //协调者初始化
        /**
         * DefaultCoordinator 继承至 AbstractTCInboundHandler, AbstractTCInboundHandler handle方法会调用doGlobalRollback()方法，doGlobalRollback()方法会调用DefaultCore的rollback()
         */
        DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer);
        coordinator.init();
        rpcServer.setHandler(new DefaultCoordinator(rpcServer));

        //全局事务发号器初始化
        UUIDGenerator.init(1);

        XID.setIpAddress(NetUtil.getLocalIp());
        XID.setPort(rpcServer.getListenPort());

        /**
         * 完成服务的启动及监听
         * RpcServer启动之后就会监听来自TM的消息，如上代码所示，在开启RpcServer之前会注册一个DefaultServerMessageListenerImpl用于对TM发过来的消息进行监听。
         * */
        rpcServer.init();
        System.exit(0);
    }
}

RpcServer就是netty一个服务端

public RpcServer(ThreadPoolExecutor messageExecutor) {
    super(new NettyServerConfig(), messageExecutor);
}

#AbstractRpcRemotingServer
public AbstractRpcRemotingServer(final NettyServerConfig nettyServerConfig,
                                 final ThreadPoolExecutor messageExecutor, final ChannelHandler... handlers) {
    super(messageExecutor);
    this.serverBootstrap = new ServerBootstrap();
    this.nettyServerConfig = nettyServerConfig;
    if (NettyServerConfig.enableEpoll()) {
        this.eventLoopGroupBoss = new EpollEventLoopGroup(nettyServerConfig.getBossThreadSize(),
            new NamedThreadFactory(nettyServerConfig.getBossThreadPrefix(), nettyServerConfig.getBossThreadSize()));
        this.eventLoopGroupWorker = new EpollEventLoopGroup(nettyServerConfig.getServerWorkerThreads(),
            new NamedThreadFactory(nettyServerConfig.getWorkerThreadPrefix(),
                nettyServerConfig.getServerWorkerThreads()));
    } else {
        this.eventLoopGroupBoss = new NioEventLoopGroup(nettyServerConfig.getBossThreadSize(),
            new NamedThreadFactory(nettyServerConfig.getBossThreadPrefix(), nettyServerConfig.getBossThreadSize()));
        this.eventLoopGroupWorker = new NioEventLoopGroup(nettyServerConfig.getServerWorkerThreads(),
            new NamedThreadFactory(nettyServerConfig.getWorkerThreadPrefix(),
                nettyServerConfig.getServerWorkerThreads()));
    }
    if (null != handlers) {
        channelHandlers = handlers;
    }
    // init listenPort in constructor so that getListenPort() will always get the exact port
    setListenPort(nettyServerConfig.getDefaultListenPort());
}

看到这里就是netty服务端构造的惯用逻辑了。

紧接着初始化SessionHolder：

SessionHolder.init(dataDir);

# SessionHolder
public static void init(String sessionStorePath) throws IOException {
    if (sessionStorePath == null) {
        ROOT_SESSION_MANAGER = new DefaultSessionManager(ROOT_SESSION_MANAGER_NAME);
        ASYNC_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(ASYNC_COMMITTING_SESSION_MANAGER_NAME);
        RETRY_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(RETRY_COMMITTING_SESSION_MANAGER_NAME);
        RETRY_ROLLBACKING_SESSION_MANAGER = new DefaultSessionManager(RETRY_ROLLBACKING_SESSION_MANAGER_NAME);
    } else {
        if (!sessionStorePath.endsWith("/")) {
            sessionStorePath = sessionStorePath + "/";
        }
        ROOT_SESSION_MANAGER = new FileBasedSessionManager(ROOT_SESSION_MANAGER_NAME, sessionStorePath);
        ASYNC_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(ASYNC_COMMITTING_SESSION_MANAGER_NAME);
        RETRY_COMMITTING_SESSION_MANAGER = new DefaultSessionManager(RETRY_COMMITTING_SESSION_MANAGER_NAME);
        RETRY_ROLLBACKING_SESSION_MANAGER = new DefaultSessionManager(RETRY_ROLLBACKING_SESSION_MANAGER_NAME);
    }
}

无非就是暂存各种DefaultSessionManager，共后续任务轮训时使用。DefaultSessionManager继承至
AbstractSessionManager，看下面父类属性sessionMap，无非就是管理GlobalSession了。
而GlobalSession就是server根据XID缓存全局事务的对象，里面属性branchSessions记录了全局事务里对应的所有分支事务，这样就能根据TM的相关触发消息去处理全局事务，进而处理分支事务了。

#AbstractSessionManager
protected Map<Long, GlobalSession> sessionMap = new ConcurrentHashMap<>();

public class GlobalSession implements SessionLifecycle, SessionStorable {
    private private long transactionId;
    private GlobalStatus status;
    private String applicationId;
    private String transactionServiceGroup;
    private String transactionName;
    private int timeout;
    private long beginTime;
    private boolean active;
    ArrayList<BranchSession> branchSessions = new ArrayList<>();
    ...
}

再者初始化coordinator：

coordinator.init();
public void init() {
    retryRollbacking.scheduleAtFixedRate(new Runnable() {

        @Override
        public void run() {
            try {
                handleRetryRollbacking();
            } catch (Exception e) {
                LOGGER.info("Exception retry rollbacking ... ", e);
            }
        }
    }, 0, 5, TimeUnit.MILLISECONDS);

    retryCommitting.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                handleRetryCommitting();
            } catch (Exception e) {
                LOGGER.info("Exception retry committing ... ", e);
            }
        }
    }, 0, 5, TimeUnit.MILLISECONDS);

    asyncCommitting.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                handleAsyncCommitting();
            } catch (Exception e) {
                LOGGER.info("Exception async committing ... ", e);
            }
        }
    }, 0, 10, TimeUnit.MILLISECONDS);

    timeoutCheck.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                timeoutCheck();
            } catch (Exception e) {
                LOGGER.info("Exception timeout checking ... ", e);
            }
        }
    }, 0, 2, TimeUnit.MILLISECONDS);
}

这里无非就是启动一些异步任务处理逻辑，其实这里最主要的就是commit异步任务删除回滚日志的逻辑。因为这里涉及到服务的相关触发逻辑，暂时先暂停到此。

继续rpcServer

rpcServer.init();

#init
@Override
public void init() {
    super.init();
    setChannelHandlers(RpcServer.this);
    DefaultServerMessageListenerImpl defaultServerMessageListenerImpl = new DefaultServerMessageListenerImpl(transactionMessageHandler);
    defaultServerMessageListenerImpl.setServerMessageSender(this);
    this.setServerMessageListener(defaultServerMessageListenerImpl);
    super.start();
}

1.父类初始化逻辑
2.设置处理器handler
3.构造消息设置监听器并监听

4.启动服务

这里最重要的就是设置监听器了，因为这里关乎如何监听TM的事务提交或者回滚消息的。直接看实现：

#DefaultServerMessageListenerImpl
@Override
public void onTrxMessage(long msgId, ChannelHandlerContext ctx, Object message, ServerMessageSender sender) {
    RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
    if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("server received:" + message + ",clientIp:" + NetUtil.toIpAddress(ctx.channel().remoteAddress()) + ",vgroup:" + rpcContext.getTransactionServiceGroup());
    } else {
        messageStrings.offer(message + ",clientIp:" + NetUtil.toIpAddress(ctx.channel().remoteAddress()) + ",vgroup:" + rpcContext.getTransactionServiceGroup());
    }
    if (!(message instanceof AbstractMessage)) { return; }
    if (message instanceof MergedWarpMessage) {
        AbstractResultMessage[] results = new AbstractResultMessage[((MergedWarpMessage)message).msgs.size()];
        for (int i = 0; i < results.length; i++) {
            final AbstractMessage subMessage = ((MergedWarpMessage)message).msgs.get(i);
            results[i] = transactionMessageHandler.onRequest(subMessage, rpcContext);
        }
        MergeResultMessage resultMessage = new MergeResultMessage();
        resultMessage.setMsgs(results);
        sender.sendResponse(msgId, ctx.channel(), resultMessage);
    } else if (message instanceof AbstractResultMessage) {
        transactionMessageHandler.onResponse((AbstractResultMessage)message, rpcContext);
    }
}

这里就是监听事务消息的触发点了，看下调用链：

@Override
public void dispatch(long msgId, ChannelHandlerContext ctx, Object msg) {
    if (msg instanceof RegisterRMRequest) {
        serverMessageListener.onRegRmMessage(msgId, ctx, (RegisterRMRequest)msg, this,
            checkAuthHandler);
    } else {
        if (ChannelManager.isRegistered(ctx.channel())) {
            serverMessageListener.onTrxMessage(msgId, ctx, msg, this);
        } else {
            try {
                closeChannelHandlerContext(ctx);
            } catch (Exception exx) {
                LOGGER.error(exx.getMessage());
            }
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
            }
        }
    }
}

继续往上追:

#AbstractRpcRemoting
@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof RpcMessage) {
        final RpcMessage rpcMessage = (RpcMessage)msg;
        if (rpcMessage.isRequest()) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getBody()));
            }
            try {
                AbstractRpcRemoting.this.messageExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            dispatch(rpcMessage.getId(), ctx, rpcMessage.getBody());
                        } catch (Throwable th) {
                            LOGGER.error(FrameworkErrorCode.NetDispatch.errCode, th.getMessage(), th);
                        }
                    }
                });
            }
        }
    }

到了netty的channel层了，SocketChannel就是netty本身的消息读写通道，一切消息交换由此追溯，即netty完成通信后触发了监听器逻辑，进而继续TC服务逻辑的处理。

启动服务

super.start();

#AbstractRpcRemotingServer
@Override
public void start() {
    this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupWorker)
        .channel(nettyServerConfig.SERVER_CHANNEL_CLAZZ)
        .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getSoBackLogSize())
        .option(ChannelOption.SO_REUSEADDR, true)
        .childOption(ChannelOption.SO_KEEPALIVE, true)
        .childOption(ChannelOption.TCP_NODELAY, true)
        .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSendBufSize())
        .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketResvBufSize())
        .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK,
            nettyServerConfig.getWriteBufferHighWaterMark())
        .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, nettyServerConfig.getWriteBufferLowWaterMark())
        .localAddress(new InetSocketAddress(listenPort))
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new IdleStateHandler(nettyServerConfig.getChannelMaxReadIdleSeconds(), 0, 0))
                    .addLast(new MessageCodecHandler());
                if (null != channelHandlers) {
                    addChannelPipelineLast(ch, channelHandlers);
                }

            }
        });

    if (nettyServerConfig.isEnableServerPooledByteBufAllocator()) {
        this.serverBootstrap.childOption(ChannelOption.ALLOCATOR, NettyServerConfig.DIRECT_BYTE_BUF_ALLOCATOR);
    }

    try {
        ChannelFuture future = this.serverBootstrap.bind(listenPort).sync();
        LOGGER.info("Server started ... ");
        future.channel().closeFuture().sync();
    } catch (InterruptedException exx) {
        throw new RuntimeException(exx);
    }
}

惯用的netty服务启动方式，至此，TC Server服务启动OK，下面总结下。

四.TC Server服务端启动总结

1.构造RpcServer的netty服务端
2.初始化SessionHolder，用于初始化各种DefaultSessionManager，DefaultSessionManager就是管理GlobalSession和BranchSession的管理器
3.初始化DefaultCoordinator，DefaultCoordinator就是具体执行提交、回滚、注册分支事务等逻辑的类，同时启动异步执行任务，比如提交全局事务后，异步删除回滚sql。
4.初始化RpcServer，设置事务消息监听器
5.启动RpcServer的netty服务端。

五.未完待续。。。

后续分析主要还是根据example官方实例分为：全局事务的获取、事务逻辑执行、事务回滚、事务提交进行。
同时后续每一流程都紧密关联Server，因此还会频繁回到上叙server启动后，收到消息被触发的后续逻辑。