Это 13-й день, когда я участвую в Gengwen Challenge. Проверьте подробности мероприятия:Обновить вызов
Прежде всего, поделитесь всеми предыдущими статьями, ставьте лайки, добавляйте в избранное и пересылайте три раза подряд. >>>>😜😜😜
Сборник статей:🎁nuggets.capable/post/694164…
Github :👉github.com/black-ant
Введение
В этой статье давайте рассмотрим файл конфигурации клиента Seaate, а также метод и процесс настройки.Сначала рассмотрим конфигурацию:
#====================================Seata Config===============================================
seata:
enabled: true
application-id: business-seata-example
tx-service-group: business-service-seata-service-group # 事务群组（可以每个应用独立取名，也可以使用相同的名字）
client:
rm-report-success-enable: true
rm-table-meta-check-enable: false # 自动刷新缓存中的表结构（默认false）
rm-report-retry-count: 5 # 一阶段结果上报TC重试次数（默认5）
rm-async-commit-buffer-limit: 10000 # 异步提交缓存队列长度（默认10000）
rm:
lock:
lock-retry-internal: 10 # 校验或占用全局锁重试间隔（默认10ms）
lock-retry-times: 30 # 校验或占用全局锁重试次数（默认30）
lock-retry-policy-branch-rollback-on-conflict: true # 分支事务与其它全局回滚事务冲突时锁策略（优先释放本地锁让回滚成功）
tm-commit-retry-count: 3 # 一阶段全局提交结果上报TC重试次数（默认1次，建议大于1）
tm-rollback-retry-count: 3 # 一阶段全局回滚结果上报TC重试次数（默认1次，建议大于1）
undo:
undo-data-validation: true # 二阶段回滚镜像校验（默认true开启）
undo-log-serialization: jackson # undo序列化方式（默认jackson）
undo-log-table: undo_log # 自定义undo表名（默认undo_log）
log:
exceptionRate: 100 # 日志异常输出概率（默认100）
support:
spring:
datasource-autoproxy: true
service:
vgroup-mapping:
my_test_tx_group: default # TC 集群（必须与seata-server保持一致）
enable-degrade: false # 降级开关
disable-global-transaction: false # 禁用全局事务（默认false）
grouplist:
default: 127.0.0.1:8091
transport:
shutdown:
wait: 3
thread-factory:
boss-thread-prefix: NettyBoss
worker-thread-prefix: NettyServerNIOWorker
server-executor-thread-prefix: NettyServerBizHandler
share-boss-worker: false
client-selector-thread-prefix: NettyClientSelector
client-selector-thread-size: 1
client-worker-thread-prefix: NettyClientWorkerThread
type: TCP
server: NIO
heartbeat: true
serialization: seata
compressor: none
enable-client-batch-send-request: true # 客户端事务消息请求是否批量合并发送（默认true）
registry:
file:
name: file.conf
type: nacos
nacos:
server-addr: localhost:8848
namespace:
cluster: default
config:
file:
name: file.conf
type: nacos
nacos:
namespace:
server-addr: localhost:8848
2. Объект конфигурации
2.1 Объекты верхнего уровня Seata
Конфигурация seata.xxx, соответствующая объекту
@ConfigurationProperties(prefix = "seata")
@EnableConfigurationProperties(SpringCloudAlibabaConfiguration.class)
public class SeataProperties {
/**
* 是否启用自动配置
*/
private boolean enabled = true;
/**
* application id
*/
private String applicationId;
/**
* 事务服务组
*/
private String txServiceGroup;
/**
* 是否启用数据源bean的自动代理
*/
private boolean enableAutoDataSourceProxy = true;
/**
* 数据源代理模式
*/
private String dataSourceProxyMode = DefaultValues.DEFAULT_DATA_SOURCE_PROXY_MODE;
/**
* 是否使用JDK代理而不是CGLIB代理
*/
private boolean useJdkProxy = false;
/**
* 指定哪个数据源bean不符合自动代理的条件
*/
private String[] excludesForAutoProxying = {};
}
2.2 объект конфигурации seata.client
client:
rm-report-success-enable: true
rm-table-meta-check-enable: false # 自动刷新缓存中的表结构（默认false）
rm-report-retry-count: 5 # 一阶段结果上报TC重试次数（默认5）
tm-commit-retry-count: 3 # 一阶段全局提交结果上报TC重试次数（默认1次，建议大于1）
tm-rollback-retry-count: 3 # 一阶段全局回滚结果上报TC重试次数（默认1次，建议大于1）
// 一级目录下有如下配置 , 对应的对象为 :
public class RmProperties {
private int asyncCommitBufferLimit = 10000;
private int reportRetryCount = 5;
private boolean tableMetaCheckEnable = false;
private boolean reportSuccessEnable = false;
private boolean sagaBranchRegisterEnable = false;
private String sagaJsonParser = fastjson;
}
Выше приведен класс, соответствующий конфигурации первого уровня клиента, давайте взглянем на его подконфигурацию.
// 二级目录 : seata.client.log
public class LogProperties {
private int exceptionRate = 100;
}
// 二级目录 : seata.client.undo
public class UndoProperties {
private boolean dataValidation = true;
private String logSerialization = "jackson";
private String logTable = "undo_log";
private boolean onlyCareUpdateColumns = true;
}
// 二级目录 : seata.client.support
2.3 объект seata.service
@Component
@ConfigurationProperties(prefix = "seata.service")
public class ServiceProperties implements InitializingBean {
/**
* vgroup->rgroup
*/
private Map<String, String> vgroupMapping = new HashMap<>();
/**
* group list
*/
private Map<String, String> grouplist = new HashMap<>();
/**
* degrade current not support
*/
private boolean enableDegrade = false;
/**
* disable globalTransaction
*/
private boolean disableGlobalTransaction = false;
}
2.4 seata.transport
Давайте посмотрим на классы конфигурации и параметры по умолчанию.
@Component
@ConfigurationProperties(prefix = TRANSPORT_PREFIX)
public class TransportProperties {
/**
* tcp, unix-domain-socket
*/
private String type = "TCP";
/**
* NIO, NATIVE
*/
private String server = "NIO";
/**
* enable heartbeat
*/
private boolean heartbeat = true;
/**
* serialization
*/
private String serialization = "seata";
/**
* compressor
*/
private String compressor = "none";
/**
* enable client batch send request
*/
private boolean enableClientBatchSendRequest = true;
}
@Component
@ConfigurationProperties(prefix = "seata.transport.thread-factory")
public class ThreadFactoryProperties {
private String bossThreadPrefix = "NettyBoss";
private String workerThreadPrefix = "NettyServerNIOWorker";
private String serverExecutorThreadPrefix = "NettyServerBizHandler";
private boolean shareBossWorker = false;
private String clientSelectorThreadPrefix = "NettyClientSelector";
private int clientSelectorThreadSize = 1;
private String clientWorkerThreadPrefix = "NettyClientWorkerThread";
}
2.5 seata.config
@Component
@ConfigurationProperties(prefix = CONFIG_PREFIX)
public class ConfigProperties {
/**
* file, nacos, apollo, zk, consul, etcd3, springCloudConfig
*/
private String type = "file";
}
ConfigNacosProperties 等 ConfigXXXProperties
Обратите внимание, что существует несколько классов реализации, каждый из которых соответствует классу конфигурации.
2.6 seata.registry
@Component
@ConfigurationProperties(prefix = "seata.registry")
public class RegistryProperties {
/**
* file, nacos, eureka, redis, zk, consul, etcd3, sofa
*/
private String type = "file";
/**
* the load balance
*/
private String loadBalance = DEFAULT_LOAD_BALANCE;
/**
* 负载均衡虚拟节点
*/
private int loadBalanceVirtualNodes = VIRTUAL_NODES_DEFAULT;
}
3. Процесс инициализации клиента
3.1 Процесс инициализации класса конфигурации
Давайте взглянем на основной класс конфигурации SeataAutoConfiguration, который здесь будет сканироваться.io.seata.spring.boot.autoconfigure.propertiesВсе классы конфигурации в рамках пакета
// Client 的配置化主要基于 SeataAutoConfiguration 类来完成 Client 配置操作
@ComponentScan(basePackages = "io.seata.spring.boot.autoconfigure.properties")
@ConditionalOnProperty(prefix = StarterConstants.SEATA_PREFIX, name = "enabled", havingValue = "true", matchIfMissing = true)
@Configuration
@EnableConfigurationProperties({SeataProperties.class})
public class SeataAutoConfiguration {
private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoConfiguration.class);
@Bean(BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER)
@ConditionalOnMissingBean(name = {BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER})
public SpringApplicationContextProvider springApplicationContextProvider() {
return new SpringApplicationContextProvider();
}
@Bean(BEAN_NAME_FAILURE_HANDLER)
@ConditionalOnMissingBean(FailureHandler.class)
public FailureHandler failureHandler() {
return new DefaultFailureHandlerImpl();
}
@Bean
@DependsOn({BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER, BEAN_NAME_FAILURE_HANDLER})
@ConditionalOnMissingBean(GlobalTransactionScanner.class)
public GlobalTransactionScanner globalTransactionScanner(SeataProperties seataProperties, FailureHandler failureHandler) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Automatically configure Seata");
}
return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup(), failureHandler);
}
/**
* 数据源配置
*/
@Configuration
@ConditionalOnProperty(prefix = StarterConstants.SEATA_PREFIX, name = {"enableAutoDataSourceProxy", "enable-auto-data-source-proxy"}, havingValue = "true", matchIfMissing = true)
static class SeataDataSourceConfiguration {
/**
* The bean seataDataSourceBeanPostProcessor.
*/
@Bean(BEAN_NAME_SEATA_DATA_SOURCE_BEAN_POST_PROCESSOR)
@ConditionalOnMissingBean(SeataDataSourceBeanPostProcessor.class)
public SeataDataSourceBeanPostProcessor seataDataSourceBeanPostProcessor(SeataProperties seataProperties) {
return new SeataDataSourceBeanPostProcessor(seataProperties.getExcludesForAutoProxying(), seataProperties.getDataSourceProxyMode());
}
/**
* The bean seataAutoDataSourceProxyCreator.
*/
@Bean(BEAN_NAME_SEATA_AUTO_DATA_SOURCE_PROXY_CREATOR)
@ConditionalOnMissingBean(SeataAutoDataSourceProxyCreator.class)
public SeataAutoDataSourceProxyCreator seataAutoDataSourceProxyCreator(SeataProperties seataProperties) {
return new SeataAutoDataSourceProxyCreator(seataProperties.isUseJdkProxy(),
seataProperties.getExcludesForAutoProxying(), seataProperties.getDataSourceProxyMode());
}
}
}
Задействованы следующие объекты:
- C- SpringApplicationContextProvider
- C- FailureHandler
- C- GlobalTransactionScanner
- C- SeataDataSourceBeanPostProcessor
- C- SeataAutoDataSourceProxyCreator
3.1.1 SpringApplicationContextProvider
public class SpringApplicationContextProvider implements ApplicationContextAware {
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
ObjectHolder.INSTANCE.setObject(OBJECT_KEY_SPRING_APPLICATION_CONTEXT, applicationContext);
}
}
Как вы можете видеть здесь, ObjectHolder вызывается для хранения applicationContext.
Когда я ранее анализировал Spring, я сказал, что Aware можно использовать для уведомления после создания, вот параметр, установленный для класса перечисления.
Это первый раз, когда я вижу класс перечисления, используемый таким образом, реализует ли он синглтон??
public enum ObjectHolder {
INSTANCE;
private static final int MAP_SIZE = 8;
private static final Map<String, Object> OBJECT_MAP = new ConcurrentHashMap<>(MAP_SIZE);
public Object getObject(String objectKey) {
return OBJECT_MAP.get(objectKey);
}
public <T> T getObject(Class<T> clasz) {
return clasz.cast(OBJECT_MAP.values().stream().filter(clasz::isInstance).findAny().orElseThrow(() -> new ShouldNeverHappenException("Can't find any object of class " + clasz.getName())));
}
public Object setObject(String objectKey, Object object) {
return OBJECT_MAP.putIfAbsent(objectKey, object);
}
}
3.1.2 FailureHandler
В FailureHandler предусмотрены следующие методы:
public interface FailureHandler {
// 启动错误
void onBeginFailure(GlobalTransaction tx, Throwable cause);
// 提交异常
void onCommitFailure(GlobalTransaction tx, Throwable cause);
// 回退异常
void onRollbackFailure(GlobalTransaction tx, Throwable originalException);
// 重试操作
void onRollbackRetrying(GlobalTransaction tx, Throwable originalException);
}
Класс реализации по умолчанию — DefaultFailureHandlerImpl , где настраиваемый означает, что вы можете реализовать и расширить его самостоятельно.
3.1.3 GlobalTransactionScanner
В этом классе задействовано несколько абстракций и интерфейсов:
- AbstractAutoProxyCreator : реализация BeanPostProcessor, которая обертывает bean-компоненты прокси-серверами AOP.
- ConfigurationChangeListener : прослушиватель модификации конфигурации
- InitializingBean: вызов инициализации
- ApplicationContextAware : обработка слушателя с учетом
- DisposableBean: уничтожить обработку
C- AbstractAutoProxyCreator # wrapIfNecessary
AbstractAutoProxyCreator подходит для прокси postProcess, который в основном запускается на этапе postProcessAfterInitialization, который называется:
// 此方法用于校验
protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
try {
synchronized (PROXYED_SET) {
if (PROXYED_SET.contains(beanName)) {
return bean;
}
interceptor = null;
// 检测是否存在 TCC 代理
if (TCCBeanParserUtils.isTccAutoProxy(bean, beanName, applicationContext)) {
// 创建一个 TccActionInterceptor 拦截器 , 并且添加到监听器
interceptor = new TccActionInterceptor(TCCBeanParserUtils.getRemotingDesc(beanName));
ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)interceptor);
} else {
// 获取目标类class
Class<?> serviceInterface = SpringProxyUtils.findTargetClass(bean);
//
Class<?>[] interfacesIfJdk = SpringProxyUtils.findInterfaces(bean);
// 判断是否存在 GlobalTransactional 注解 , 不存在直接返回
if (!existsAnnotation(new Class[]{serviceInterface})
&& !existsAnnotation(interfacesIfJdk)) {
return bean;
}
// 如果存在注解 , 则为其创建一个 GlobalTransactionalInterceptor 用于事务拦截
if (interceptor == null) {
if (globalTransactionalInterceptor == null) {
globalTransactionalInterceptor = new GlobalTransactionalInterceptor(failureHandlerHook);
ConfigurationCache.addConfigListener(
ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)globalTransactionalInterceptor);
}
interceptor = globalTransactionalInterceptor;
}
}
// 如果不是 AOP 代理 , 则直接放回 , 否者 , 进行 AOP 实际处理
if (!AopUtils.isAopProxy(bean)) {
bean = super.wrapIfNecessary(bean, beanName, cacheKey);
} else {
AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean);
Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));
for (Advisor avr : advisor) {
advised.addAdvisor(0, avr);
}
}
PROXYED_SET.add(beanName);
return bean;
}
} catch (Exception exx) {
throw new RuntimeException(exx);
}
}
PS: Роль AdvisedSupport
Проще говоря, этот класс проксируется.Реализация BeanPostProcessor каждого подходящего bean-компонента обернута прокси-сервером AOP, который будет делегировать указанному перехватчику перед вызовом самого bean-компонента.
ConfigurationChangeListener
@Override
public void onChangeEvent(ConfigurationChangeEvent event) {
if (ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION.equals(event.getDataId())) {
disableGlobalTransaction = Boolean.parseBoolean(event.getNewValue().trim());
if (!disableGlobalTransaction && initialized.compareAndSet(false, true)) {
// 可以看到 , 再修改之后会重新初始化一次客户端
initClient();
}
}
}
InitializingBean
public void afterPropertiesSet() {
ConfigurationCache.addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION,
(ConfigurationChangeListener)this);
if (disableGlobalTransaction) {
return;
}
if (initialized.compareAndSet(false, true)) {
// 同样 , 初始化的时候调用了 initClient
initClient();
}
}
См. здесь, что здесь инициализируются клиенты TM и RM.
private void initClient() {
if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {
throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));
}
//init TM
TMClient.init(applicationId, txServiceGroup, accessKey, secretKey);
//init RM
RMClient.init(applicationId, txServiceGroup);
// 注册销毁钩子
registerSpringShutdownHook();
}
ApplicationContextAware
Настройки уведомлений выполняются здесь через Aware ApplicationContext.
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
this.setBeanFactory(applicationContext);
}
DisposableBean
public void destroy() {
ShutdownHook.getInstance().destroyAll();
}
private void registerSpringShutdownHook() {
if (applicationContext instanceof ConfigurableApplicationContext) {
((ConfigurableApplicationContext) applicationContext).registerShutdownHook();
ShutdownHook.removeRuntimeShutdownHook();
}
ShutdownHook.getInstance().addDisposable(TmNettyRemotingClient.getInstance(applicationId, txServiceGroup));
ShutdownHook.getInstance().addDisposable(RmNettyRemotingClient.getInstance(applicationId, txServiceGroup));
}
3.1.4 SeataDataSourceBeanPostProcessor
public class SeataDataSourceBeanPostProcessor implements BeanPostProcessor {
private static final Logger LOGGER = LoggerFactory.getLogger(SeataDataSourceBeanPostProcessor.class);
private final List<String> excludes;
private final BranchType dataSourceProxyMode;
public SeataDataSourceBeanPostProcessor(String[] excludes, String dataSourceProxyMode) {
this.excludes = Arrays.asList(excludes);
this.dataSourceProxyMode = BranchType.XA.name().equalsIgnoreCase(dataSourceProxyMode) ? BranchType.XA : BranchType.AT;
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) {
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof DataSource) {
//当不在排除时，放置和初始化代理
if (!excludes.contains(bean.getClass().getName())) {
//只放置和初始化代理，不返回代理
DataSourceProxyHolder.get().putDataSource((DataSource) bean, dataSourceProxyMode);
}
//如果为SeataDataSourceProxy，则返回原始数据源
if (bean instanceof SeataDataSourceProxy) {
return ((SeataDataSourceProxy) bean).getTargetDataSource();
}
}
return bean;
}
}
SeataDataSourceProxy 是一个接口 , 它提供了多个方法 :
public interface SeataDataSourceProxy extends DataSource {
// 获取目标数据源。
DataSource getTargetDataSource();
// 得到分支类型。
BranchType getBranchType();
}
而该代理类 , 有如下几个实现 :
SeataAutoDataSourceProxyCreator
public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {
private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoDataSourceProxyCreator.class);
private final List<String> excludes;
private final Advisor advisor;
public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes, String dataSourceProxyMode) {
this.excludes = Arrays.asList(excludes);
this.advisor = new DefaultIntroductionAdvisor(new SeataAutoDataSourceProxyAdvice(dataSourceProxyMode));
setProxyTargetClass(!useJdkProxy);
}
@Override
protected Object[] getAdvicesAndAdvisorsForBean(Class<?> beanClass, String beanName, TargetSource customTargetSource) throws BeansException {
return new Object[]{advisor};
}
@Override
protected boolean shouldSkip(Class<?> beanClass, String beanName) {
return !DataSource.class.isAssignableFrom(beanClass) ||
SeataProxy.class.isAssignableFrom(beanClass) ||
excludes.contains(beanClass.getName());
}
}
Суммировать
На самом деле, самая важная операция всей статьи — построитьglobalTransactionalInterceptor, мы будем использовать этот перехватчик в последующем основном процессе
Обратите внимание, что каждый аннотированный метод будет обрабатываться перехватчиком.