開篇
?這篇文章的目的主要是理清楚Fescar的TM發(fā)送部分的邏輯蓝牲，從時序圖和源碼兩個層面進行分析。
?文章中間會解答兩個自己閱讀代碼中遇到的困惑（估計大部分人看代碼的時候也會遇到這個困惑）泰讽，包括TmRpcClient的初始化過程和配置加載過程例衍。
?文章的最后會附上GlobalAction相關(guān)Request的類關(guān)系圖，便于理解依賴關(guān)系已卸。
?
Fescar TM發(fā)送流程
TM Sender.jpg
說明：
1.DefaultGlobalTransaction執(zhí)行begin/commit/rollback等調(diào)用DefaultTransactionManager佛玄。
2.DefaultTransactionManager內(nèi)部調(diào)用syncCall()方法，進而調(diào)用TmRpcClient的sendMsgWithResponse()方法累澡。
3.TmRpcClient調(diào)用父類AbstractRpcRemoting的sendAsyncRequest()方法構(gòu)建發(fā)送隊列梦抢。
4.AbstractRpcRemotingClient的MergedSendRunnable線程消費發(fā)送隊列構(gòu)建MergedWarpMessage調(diào)用sendRequest發(fā)送。
5.sendRequest()方法內(nèi)部調(diào)用writeAndFlush完成消息發(fā)送愧哟。
TmRcpClient
說明：
TmRpcClient的類依賴關(guān)系圖如上奥吩。
TmRpcClient繼承自AbstractRpcRemotingClient類。
Fescar TM發(fā)送源碼分析
public class DefaultTransactionManager implements TransactionManager {
private static class SingletonHolder {
private static final TransactionManager INSTANCE = new DefaultTransactionManager();
}
/**
* Get transaction manager.
*
* @return the transaction manager
*/
public static TransactionManager get() {
return SingletonHolder.INSTANCE;
}
private DefaultTransactionManager() {
}
@Override
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout)
throws TransactionException {
GlobalBeginRequest request = new GlobalBeginRequest();
request.setTransactionName(name);
request.setTimeout(timeout);
GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
return response.getXid();
}
@Override
public GlobalStatus commit(String xid) throws TransactionException {
long txId = XID.getTransactionId(xid);
GlobalCommitRequest globalCommit = new GlobalCommitRequest();
globalCommit.setTransactionId(txId);
GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit);
return response.getGlobalStatus();
}
@Override
public GlobalStatus rollback(String xid) throws TransactionException {
long txId = XID.getTransactionId(xid);
GlobalRollbackRequest globalRollback = new GlobalRollbackRequest();
globalRollback.setTransactionId(txId);
GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback);
return response.getGlobalStatus();
}
@Override
public GlobalStatus getStatus(String xid) throws TransactionException {
long txId = XID.getTransactionId(xid);
GlobalStatusRequest queryGlobalStatus = new GlobalStatusRequest();
queryGlobalStatus.setTransactionId(txId);
GlobalStatusResponse response = (GlobalStatusResponse) syncCall(queryGlobalStatus);
return response.getGlobalStatus();
}
private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
try {
return (AbstractTransactionResponse) TmRpcClient.getInstance().sendMsgWithResponse(request);
} catch (TimeoutException toe) {
throw new TransactionException(TransactionExceptionCode.IO, toe);
}
}
}
說明：
DefaultTransactionManager的beigin/commit/rollback方法內(nèi)部最終調(diào)用syncCall()方法蕊梧。
syncCall方法內(nèi)部執(zhí)行TmRpcClient.getInstance().sendMsgWithResponse(request)調(diào)用TmRpcClient方法霞赫。
public final class TmRpcClient extends AbstractRpcRemotingClient {
@Override
public Object sendMsgWithResponse(Object msg) throws TimeoutException {
return sendMsgWithResponse(msg, NettyClientConfig.getRpcRequestTimeout());
}
@Override
public Object sendMsgWithResponse(String serverAddress, Object msg, long timeout)
throws TimeoutException {
return sendAsyncRequestWithResponse(serverAddress, connect(serverAddress), msg, timeout);
}
}
說明：
- TmRpcClient內(nèi)部執(zhí)行發(fā)送sendMsgWithResponse調(diào)用sendAsyncRequestWithResponse。
- sendAsyncRequestWithResponse的實現(xiàn)在父類AbstractRpcRemoting當(dāng)中肥矢。
public abstract class AbstractRpcRemoting extends ChannelDuplexHandler {
protected Object sendAsyncRequestWithResponse(String address, Channel channel, Object msg, long timeout) throws
TimeoutException {
if (timeout <= 0) {
throw new FrameworkException("timeout should more than 0ms");
}
return sendAsyncRequest(address, channel, msg, timeout);
}
private Object sendAsyncRequest(String address, Channel channel, Object msg, long timeout)
throws TimeoutException {
if (channel == null) {
LOGGER.warn("sendAsyncRequestWithResponse nothing, caused by null channel.");
return null;
}
// 構(gòu)建RpcMessage對象
final RpcMessage rpcMessage = new RpcMessage();
rpcMessage.setId(RpcMessage.getNextMessageId());
rpcMessage.setAsync(false);
rpcMessage.setHeartbeat(false);
rpcMessage.setRequest(true);
rpcMessage.setBody(msg);
// 通過MessageFuture包裝實現(xiàn)超時
final MessageFuture messageFuture = new MessageFuture();
messageFuture.setRequestMessage(rpcMessage);
messageFuture.setTimeout(timeout);
futures.put(rpcMessage.getId(), messageFuture);
// 測試代碼走的是這個分支
if (address != null) {
// 根據(jù)address進行hash放置到不同的Map當(dāng)中
ConcurrentHashMap<String, BlockingQueue<RpcMessage>> map = basketMap;
BlockingQueue<RpcMessage> basket = map.get(address);
if (basket == null) {
map.putIfAbsent(address, new LinkedBlockingQueue<RpcMessage>());
basket = map.get(address);
}
basket.offer(rpcMessage);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("offer message: " + rpcMessage.getBody());
}
// 發(fā)送其實是另外一個線程單獨執(zhí)行發(fā)送操作的
if (!isSending) {
synchronized (mergeLock) {
mergeLock.notifyAll();
}
}
} else {
ChannelFuture future;
channelWriteableCheck(channel, msg);
future = channel.writeAndFlush(rpcMessage);
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
if (!future.isSuccess()) {
MessageFuture messageFuture = futures.remove(rpcMessage.getId());
if (messageFuture != null) {
messageFuture.setResultMessage(future.cause());
}
destroyChannel(future.channel());
}
}
});
}
// 通過Future實現(xiàn)限時超時機制
if (timeout > 0) {
try {
return messageFuture.get(timeout, TimeUnit.MILLISECONDS);
} catch (Exception exx) {
LOGGER.error("wait response error:" + exx.getMessage() + ",ip:" + address + ",request:" + msg);
if (exx instanceof TimeoutException) {
throw (TimeoutException)exx;
} else {
throw new RuntimeException(exx);
}
}
} else {
return null;
}
}
}
說明：
- 構(gòu)建RpcMessage對象端衰，包裝Request。
- 構(gòu)建MessageFuture對象甘改，包裝RpcMessage旅东，實現(xiàn)超時等待功能。
- 通過basket進行分桶操作十艾，真正執(zhí)行發(fā)送的代碼在AbstractRpcRemotingClient類的MergedSendRunnable抵代。
- Request的發(fā)送類似生成消費者模型，上述代碼只是生產(chǎn)者部分疟羹。
public abstract class AbstractRpcRemotingClient extends AbstractRpcRemoting
implements RemotingService, RegisterMsgListener, ClientMessageSender {
public class MergedSendRunnable implements Runnable {
@Override
public void run() {
while (true) {
synchronized (mergeLock) {
try {
mergeLock.wait(MAX_MERGE_SEND_MILLS);
} catch (InterruptedException e) {}
}
isSending = true;
for (String address : basketMap.keySet()) {
BlockingQueue<RpcMessage> basket = basketMap.get(address);
if (basket.isEmpty()) { continue; }
MergedWarpMessage mergeMessage = new MergedWarpMessage();
while (!basket.isEmpty()) {
RpcMessage msg = basket.poll();
mergeMessage.msgs.add((AbstractMessage)msg.getBody());
mergeMessage.msgIds.add(msg.getId());
}
if (mergeMessage.msgIds.size() > 1) {
printMergeMessageLog(mergeMessage);
}
Channel sendChannel = connect(address);
try {
sendRequest(sendChannel, mergeMessage);
} catch (FrameworkException e) {
if (e.getErrcode() == FrameworkErrorCode.ChannelIsNotWritable
&& address != null) {
destroyChannel(address, sendChannel);
}
LOGGER.error("", "client merge call failed", e);
}
}
isSending = false;
}
}
}
說明：
- MergedSendRunnable 負責(zé)消費待發(fā)送消息體并組裝成MergedWarpMessage對象主守。
- sendRequest()方法內(nèi)部將MergedWarpMessage再次包裝成RpcMessage進行發(fā)送。
public abstract class AbstractRpcRemoting extends ChannelDuplexHandler {
protected void sendRequest(Channel channel, Object msg) {
RpcMessage rpcMessage = new RpcMessage();
rpcMessage.setAsync(true);
rpcMessage.setHeartbeat(msg instanceof HeartbeatMessage);
rpcMessage.setRequest(true);
rpcMessage.setBody(msg);
rpcMessage.setId(RpcMessage.getNextMessageId());
if (msg instanceof MergeMessage) {
mergeMsgMap.put(rpcMessage.getId(), (MergeMessage)msg);
}
channelWriteableCheck(channel, msg);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("write message:" + rpcMessage.getBody() + ", channel:" + channel + ",active?"
+ channel.isActive() + ",writable?" + channel.isWritable() + ",isopen?" + channel.isOpen());
}
channel.writeAndFlush(rpcMessage);
}
}
說明：
- RpcMessage再次包裝MergeMessage進行發(fā)送榄融。
TmRpcClient初始化
public class GlobalTransactionScanner extends AbstractAutoProxyCreator implements InitializingBean {
public GlobalTransactionScanner(String applicationId, String txServiceGroup, int mode,
FailureHandler failureHandlerHook) {
setOrder(ORDER_NUM);
setProxyTargetClass(true);
this.applicationId = applicationId;
this.txServiceGroup = txServiceGroup;
this.mode = mode;
this.failureHandlerHook = failureHandlerHook;
}
private void initClient() {
TMClient.init(applicationId, txServiceGroup);
if ((AT_MODE & mode) > 0) {
RMClientAT.init(applicationId, txServiceGroup);
}
}
public void afterPropertiesSet() {
initClient();
}
}
說明：
- GlobalTransactionScanner的構(gòu)造函數(shù)執(zhí)行后執(zhí)行afterPropertiesSet并執(zhí)行initClient()操作参淫。
- initClient()內(nèi)部執(zhí)行TMClient.init(applicationId, txServiceGroup)進行TMClient的初始化。
public class TMClient {
public static void init(String applicationId, String transactionServiceGroup) {
TmRpcClient tmRpcClient = TmRpcClient.getInstance(
applicationId, transactionServiceGroup);
tmRpcClient.init();
}
}
public final class TmRpcClient extends AbstractRpcRemotingClient {
public void init() {
if (initialized.compareAndSet(false, true)) {
init(SCHEDULE_INTERVAL_MILLS, SCHEDULE_INTERVAL_MILLS);
}
}
public void init(long healthCheckDelay, long healthCheckPeriod) {
// 注意initVars()方法
initVars();
ExecutorService mergeSendExecutorService = new ThreadPoolExecutor(
MAX_MERGE_SEND_THREAD, MAX_MERGE_SEND_THREAD,
KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
new NamedThreadFactory(getThreadPrefix(MERGE_THREAD_PREFIX),
MAX_MERGE_SEND_THREAD));
mergeSendExecutorService.submit(new MergedSendRunnable());
timerExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
reconnect();
} catch (Exception ignore) {
LOGGER.error(ignore.getMessage());
}
}
}, healthCheckDelay, healthCheckPeriod, TimeUnit.SECONDS);
}
private void initVars() {
enableDegrade = CONFIG.getBoolean(
ConfigurationKeys.SERVICE_PREFIX + ConfigurationKeys.ENABLE_DEGRADE_POSTFIX);
super.init();
}
}
說明：
- 核心在于關(guān)注initVars()方法愧杯。
public abstract class AbstractRpcRemotingClient extends AbstractRpcRemoting
implements RemotingService, RegisterMsgListener, ClientMessageSender {
public void init() {
NettyPoolableFactory keyPoolableFactory = new NettyPoolableFactory(this);
// 核心構(gòu)建發(fā)送的對象的連接池
nettyClientKeyPool = new GenericKeyedObjectPool(keyPoolableFactory);
nettyClientKeyPool.setConfig(getNettyPoolConfig());
serviceManager = new ServiceManagerStaticConfigImpl();
super.init();
}
}
public abstract class AbstractRpcRemoting extends ChannelDuplexHandler {
public void init() {
timerExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
List<MessageFuture> timeoutMessageFutures = new ArrayList<MessageFuture>(futures.size());
for (MessageFuture future : futures.values()) {
if (future.isTimeout()) {
timeoutMessageFutures.add(future);
}
}
for (MessageFuture messageFuture : timeoutMessageFutures) {
futures.remove(messageFuture.getRequestMessage().getId());
messageFuture.setResultMessage(null);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("timeout clear future : " + messageFuture.getRequestMessage().getBody());
}
}
nowMills = System.currentTimeMillis();
}
}, TIMEOUT_CHECK_INTERNAL, TIMEOUT_CHECK_INTERNAL, TimeUnit.MILLISECONDS);
}
}
說明：
- AbstractRpcRemotingClient的init()方法核心構(gòu)建nettyClientKeyPool工廠涎才。
- nettyClientKeyPool用于獲取連接TC的對象的工廠池痹屹。
配置加載分析
Config
public class FileConfiguration implements Configuration {
private static final Logger LOGGER = LoggerFactory.getLogger(FileConfiguration.class);
private static final Config CONFIG = ConfigFactory.load();
}
package com.typesafe.config;
public final class ConfigFactory {
private ConfigFactory() {
}
public static Config load() {
return load(ConfigParseOptions.defaults());
}
}
說明：
- 配置加載使用了JAVA 配置管理庫 typesafe.config
- 默認加載classpath下的application.conf,application.json和application.properties文件纷跛。通過ConfigFactory.load()加載。
Request的類關(guān)系圖
GlobalActionRequest.png
