上一篇文章说过,在Spring启动的过程中就会就会初始化TM和RM,建立与TC的长连接。TM,RM,TC都是用netty来处理网络连接的,初始化netty客户端和服务端的过程也非常类似。
本篇文章只分析TM的初始化过程,RM和TM复用了很多方法
// TmNettyRemotingClient
public void init() {
// registry processor
// 注册消息处理器
registerProcessor();
if (initialized.compareAndSet(false, true)) {
super.init();
}
}
// AbstractNettyRemotingClient
public void init() {
// 不断连接seata server
timerExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
clientChannelManager.reconnect(getTransactionServiceGroup());
}
}, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);
// 是否允许批量发送请求
if (NettyClientConfig.isEnableClientBatchSendRequest()) {
mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,
MAX_MERGE_SEND_THREAD,
KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(),
new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));
mergeSendExecutorService.submit(new MergedSendRunnable());
}
// 移除发送超时的消息
super.init();
clientBootstrap.start();
}
clientBootstrap#start是netty启动的模版代码,注册消息处理器和处理消息的套路我在seata server启动的文章分析的比较详细,本篇文章就不深入分析了
TM和RM每隔10s都要TC集群的每个地址建立长连接
// NettyClientChannelManager#reconnect
void reconnect(String transactionServiceGroup) {
List<String> availList = null;
try {
// 获得事务分组对应的集群中每台机器地址
availList = getAvailServerList(transactionServiceGroup);
} catch (Exception e) {
LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);
return;
}
if (CollectionUtils.isEmpty(availList)) {
RegistryService registryService = RegistryFactory.getInstance();
String clusterName = registryService.getServiceGroup(transactionServiceGroup);
if (StringUtils.isBlank(clusterName)) {
LOGGER.error("can not get cluster name in registry config '{}{}', please make sure registry config correct",
ConfigurationKeys.SERVICE_GROUP_MAPPING_PREFIX,
transactionServiceGroup);
return;
}
if (!(registryService instanceof FileRegistryServiceImpl)) {
LOGGER.error("no available service found in cluster '{}', please make sure registry config correct and keep your seata server running", clusterName);
}
return;
}
// 遍历tc服务器地址
for (String serverAddress : availList) {
try {
// 建立与tc的连接
acquireChannel(serverAddress);
} catch (Exception e) {
LOGGER.error("{} can not connect to {} cause:{}",FrameworkErrorCode.NetConnect.getErrCode(), serverAddress, e.getMessage(), e);
}
}
}
Channel acquireChannel(String serverAddress) {
Channel channelToServer = channels.get(serverAddress);
// 与当前serverAddress已经存在连接,直接返回
if (channelToServer != null) {
channelToServer = getExistAliveChannel(channelToServer, serverAddress);
if (channelToServer != null) {
return channelToServer;
}
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info("will connect to " + serverAddress);
}
// 与当前serverAddress不存在连接,新建连接
Object lockObj = CollectionUtils.computeIfAbsent(channelLocks, serverAddress, key -> new Object());
synchronized (lockObj) {
return doConnect(serverAddress);
}
}
private Channel doConnect(String serverAddress) {
Channel channelToServer = channels.get(serverAddress);
// 当前地址已经存在连接
if (channelToServer != null && channelToServer.isActive()) {
return channelToServer;
}
Channel channelFromPool;
try {
NettyPoolKey currentPoolKey = poolKeyFunction.apply(serverAddress);
NettyPoolKey previousPoolKey = poolKeyMap.putIfAbsent(serverAddress, currentPoolKey);
if (previousPoolKey != null && previousPoolKey.getMessage() instanceof RegisterRMRequest) {
RegisterRMRequest registerRMRequest = (RegisterRMRequest) currentPoolKey.getMessage();
((RegisterRMRequest) previousPoolKey.getMessage()).setResourceIds(registerRMRequest.getResourceIds());
}
channelFromPool = nettyClientKeyPool.borrowObject(poolKeyMap.get(serverAddress));
channels.put(serverAddress, channelFromPool);
} catch (Exception exx) {
LOGGER.error("{} register RM failed.",FrameworkErrorCode.RegisterRM.getErrCode(), exx);
throw new FrameworkException("can not register RM,err:" + exx.getMessage());
}
return channelFromPool;
}
TM和RM客户端在启动的时候会和集群中的的每台seata server建立长连接,但是在后续发送请求的时候,比如开启全局事务,注册分支事务只会和其中的一台机器通讯,TM或RM首先根据事务分组找到集群列表,然后根据负载均衡策略从列表中选出一台机器发起请求。具体代码可参见AbstractNettyRemotingClient#sendSyncRequest方法
[1]https://blog.csdn.net/zjj2006/category_10310426.html
[2]https://blog.csdn.net/weixin_38308374/article/details/108944877
文章浏览阅读1.2k次。vue2封装对话框el-dialog组件_
文章浏览阅读4.7k次,点赞5次,收藏6次。MFC 文本框换行 标签: it mfc 文本框1.将Multiline属性设置为True2.换行是使用"\r\n" (宽字符串为L"\r\n")3.如果需要编辑并且按Enter键换行,还要将 Want Return 设置为 True4.如果需要垂直滚动条的话将Vertical Scroll属性设置为True,需要水平滚动条的话将Horizontal Scroll属性设_c++ mfc同一框内输入二行怎么换行
文章浏览阅读832次。检查Linux是否是否开启所需端口,默认为6379,若未打开,将其开启:以root用户执行iptables -I INPUT -p tcp --dport 6379 -j ACCEPT如果还是未能解决,修改redis.conf,修改主机地址:bind 192.168.85.**;然后使用该配置文件,重新启动Redis服务./redis-server redis.conf..._redis-server doesn't support auth command or ismisconfigured. try
文章浏览阅读4.9k次。济大数电实验报告_数据选择器及其应用
文章浏览阅读1w次,点赞3次,收藏5次。题目描述所谓角谷猜想,是指对于任意一个正整数,如果是奇数,则乘3加1,如果是偶数,则除以2,得到的结果再按照上述规则重复处理,最终总能够得到1。如,假定初始整数为5,计算过程分别为16、8、4、2、1。程序要求输入一个整数,将经过处理得到1的过程输出来。输入 一个正整数N(N <= 2,000,000) 输出 从输入整数到1的步骤,每一步为一行,每一部中描述计算过程。最后一行输出"En..._角谷猜想c++代码
文章浏览阅读1k次。1: private float[,] LoadHeightData(Texture2D heightMap) 2: { 3: float minimumHeight = 255; 4: float maximumHeight = 0; 5: 6: int width = heightMap.Width; _positions indices 自动创建地形
文章浏览阅读5.5w次,点赞3次,收藏41次。加密方式主要有3种,摘要加密、对称加密与非对称加密。_java 加密 解密
文章浏览阅读6.5k次,点赞2次,收藏9次。sql语句中,以逗号分隔的字符串中是否包含某个特定字符串,类似于判断一个数组中是否包含某一个元素,例如:判断 ,a,b,c,d,e,f,g,中是否包含 a ,sql语句如何实现?SQL中没有类似VB中的Split的函数。此时,如要匹配 ,6,3,9,22,中的6。可用charindex函数巧妙实现:(注意红色的逗号)charindex( ',' +cast(6 as varchar)+ ',',‘,6,3,9,22,’)>0sqlserver:sqlserver中没有封装好的方法可以实现._sql server 逗号分隔包含
文章浏览阅读1.8k次。 第一次接触这个游戏,还是通过校内网。然后下载了一个单机绿色破解版的。感觉挺有意思,就慢慢玩了起来。后来打开游戏文件夹,发现base里面竟然都是一些xml、lua和jpg文件,感觉好熟悉,于是便开始了我的分析。 游戏目录结构如下:│ Around the World in 80 Days.exe│ bass.dll│ settings.ini│ ├─sc
文章浏览阅读8.4k次。百度普通IP定位是一套以HTTP/HTTPS形式提供的轻量级定位接口,用户可以通过该服务,根据IP定位来获取大致位置。1.首先需要成为百度地图开放平台的注册用户。http://lbsyun.baidu.com/index.php2. 申请Web服务API服务,在控制台创建一个应用:应用类型为浏览器端3.创建好应用后你,将会得到一个访问应用(AK).4.请求URLhttp://..._百度ip归属api
文章浏览阅读4.7k次。DataGrid颜色专题在Flex运用中经常提到的有关DataGrid问题是如何改变DataGrid单元格(cell),列(column)和行(row)的背景颜色(backgroundcolor)这里对这3种颜色做一个总结。 1. 设置行(row)的背景色 主要是通过对DataGrid扩展,对protected函数drawRowBackground()进行重写,具体代码如下:package com{ im_datagrid背景颜色
文章浏览阅读5k次。欢迎关注“小白玩转Python”,发现更多 “有趣”引言如果你刚刚开始学习 PyTorch,并想学习如何做一些基本的图像分类,那么这篇文章你一定不要错过哦~本文将通过组织自己的训练数据..._pytorch torchvision 训练自己的数据 图像分类