seata的TM端启动异常分析

Posted 2021-09-28 熙熙

(1): 在试用seata的1.4.1版本的时候，日志中总有 "no available service found in cluster \'default\', please make sure registry config correct and keep your seata server running" 这个异常日志打印。 在网上翻阅了一下资料，然而不得要领，于是自己翻阅源码，很快解决问题。（说明：seata的1.4.1版本开始，可以不使用registry.conf 和 file.conf 文件，使用springboot的配置文件即可，而且提供了默认值，可以根据需要配置指定的值）。

(2)：查看源码，是在类NettyClientChannelManager的方法 reconnect 中出现。查看代码，是根据配置的事务组获取有效的服务地址为空所致。
里面参数 transactionServiceGroup 的值是 通过配置 seata.service.vgroup-mapping.xxx 指定。该类的源码如下：

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)) {
        // 忽略
        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;
        }
 // 忽略
}

(3)：getAvailServerList 方法的实现很简单，就是根据 seata.registry.type 的配置值，找到对应类型的registry。从对应的服务中获取地址。
本次尝试是配置的值是 eureka，所以是从eureka 中获取配置的值。

private List<String> getAvailServerList(String transactionServiceGroup) throws Exception {
    List<InetSocketAddress> availInetSocketAddressList = RegistryFactory.getInstance()
                                                                        .lookup(transactionServiceGroup);
    if (CollectionUtils.isEmpty(availInetSocketAddressList)) {
        return Collections.emptyList();
    }
    return availInetSocketAddressList.stream().map(NetUtil::toStringAddress).collect(Collectors.toList());
}

(4)：从 EurekaRegistryServiceImpl类的lookup方法中，可以看出是从变量 clusterAddressMap 中获取值。

public List<InetSocketAddress> lookup(String key) throws Exception {
    String clusterName = getServiceGroup(key);
    if (clusterName == null) {
        return null;
    }
    if (!subscribeListener) {
        refreshCluster();
        subscribe(null, event -> {
            try {
                refreshCluster();
                } catch (Exception e) {
                   LOGGER.error("Eureka event listener refreshCluster error:{}", e.getMessage(), e);
                }
          });
    }
    return new ArrayList<>(clusterAddressMap.getOrDefault(clusterName.toUpperCase(), Collections.emptySet()));
}

(5)：clusterAddressMap 的值是在函数 refreshCluster 中赋值的。看函数 refreshCluster 的实现可以看到，refreshCluster的值是把eureka中注册的服务及其IP地址信息建立一个MAP对象。 这样就非常明显了，在seata-server的 registry.type 配置为eureka，而且 eureka的 application 值和客户端的配置 seata.service.vgroup-mapping.xxx= 的值一致即可。代码如下：

private void refreshCluster() {
    List<Application> applications = getEurekaClient(false).getApplications().getRegisteredApplications(); //获取eureka的注册信息。这里是读取的缓存。
    if (CollectionUtils.isEmpty(applications)) {
        clusterAddressMap.clear();
        return;
    }
    ConcurrentMap<String, Set<InetSocketAddress>> collect = new ConcurrentHashMap<>(MAP_INITIAL_CAPACITY);
        for (Application application : applications) {
            List<InstanceInfo> instances = application.getInstances();
            if (CollectionUtils.isNotEmpty(instances)) {
                Set<InetSocketAddress> addressSet = instances.stream()
                    .map(instance -> new InetSocketAddress(instance.getIPAddr(), instance.getPort()))
                    .collect(Collectors.toSet());
               collect.put(application.getName(), addressSet);
           }
        }
    clusterAddressMap = collect;
}