您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

高可用池化 Thrift Client 實現：ThriftJ 設計與源碼深度解析

編輯：Java 綜合教學

高可用池化 Thrift Client 實現：ThriftJ 設計與源碼深度解析

Thrift 原生缺乏 Smart Client 能力（服務發現、負載均衡、故障轉移、連接池）。ThriftJ 針對此痛點，提供鏈式 API、連接池、多策略負載均衡、異常隔離恢復、服務級別降級的生產級 Thrift 客戶端封裝。

核心特性

鏈式調用 API，配置簡潔直觀
完善默認值，開箱即用
基於 Commons Pool 2 池化 TTransport，高效管理連接生命周期
異常服務自動隔離與恢復（Guava Cache 實現）
四種負載均衡策略：隨機、輪詢、權重、哈希
三級服務級別：SERVERS_ONLY / ALL_SERVERS / NOT_EMPTY，自動降級

Maven 依賴

<dependency>
    <groupId>com.github.cyfonly</groupId>
    <artifactId>thriftj</artifactId>
    <version>1.0.1</version>
</dependency>
<!-- 需額外添加 slf4j 實現，如 logback-classic -->

使用示例

// 服務列表
String servers = "127.0.0.1:10001,127.0.0.1:10002";

// 連接驗證器
ConnectionValidator validator = transport -> transport.isOpen();

// 連接池配置
GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig();

// Failover 策略
FailoverStrategy failoverStrategy = new FailoverStrategy(); // 默認 10次/1分/1分

// 構建客戶端
ThriftClient thriftClient = new ThriftClient();
thriftClient.servers(servers)
    .loadBalance(Constant.LoadBalance.RANDOM)
    .connectionValidator(validator)
    .poolConfig(poolConfig)
    .failoverStrategy(failoverStrategy)
    .connTimeout(5)              // 秒
    .backupServers("")           // 備用集群
    .serviceLevel(Constant.ServiceLevel.NOT_EMPTY)
    .start();

// 獲取可用服務列表
for (ThriftServer s : thriftClient.getAvailableServers()) {
    System.out.println(s.getHost() + ":" + s.getPort());
}

// 業務調用
TestThriftJ.Client client = thriftClient.iface(TestThriftJ.Client.class);
QryResult result = client.qryTest(1);
System.out.println("result[code=" + result.code + " msg=" + result.msg + "]");

架構設計

ThriftClient (門面)
├─ ServerRegistry          # 服務節點註冊、健康狀態
├─ LoadBalancer            # 策略模式：Random/RoundRobin/Weighted/Hash
├─ ConnectionPoolManager   # KeyedObjectPool
│   └─ ThriftConnectionFactory # makeObject/validateObject/destroyObject
├─ FailoverChecker         # 異步健康檢查線程 + Guava Cache 狀態機
└─ ServiceLevelResolver    # 可用節點篩選與降級邏輯

關鍵源碼解析

1. 連接工廠與池管理

public class ThriftConnectionFactory extends KeyedPooledObjectFactory {
    @Override
    public PooledObject makeObject(ThriftServer server) {
        TSocket socket = new TSocket(server.getHost(), server.getPort());
        socket.setTimeout(timeout);
        TFramedTransport transport = new TFramedTransport(socket);
        transport.open();
        return new DefaultPooledObject<>(transport);
    }

    @Override
    public boolean validateObject(ThriftServer server, PooledObject pooled) {
        boolean valid = pooled.getObject().isOpen();
        if (failoverChecker != null) {
            valid &= failoverChecker.getConnectionValidator().isValid(pooled.getObject());
        }
        if (!valid) failoverChecker.getFailoverStrategy().fail(server);
        return valid;
    }

    @Override
    public void destroyObject(ThriftServer server, PooledObject pooled) {
        pooled.getObject().close();
    }
}

2. Failover 狀態機

public class FailoverStrategy {
    private final Cache> failCountMap; // 失敗時間戳隊列
    private final Cache failedList;               // 失效節點集合

    public void fail(ThriftServer server) {
        EvictingQueue q = failCountMap.get(server);
        synchronized (q) {
            q.add(System.currentTimeMillis());
            if (q.remainingCapacity() == 0 && q.element() >= (System.currentTimeMillis() - failDuration)) {
                failedList.put(server, true); // 達到閾值，標記失效
            }
        }
    }

    public Set getFailed() { return failedList.asMap().keySet(); }
}

默認參數：failCount=10, failDuration=1min, recoverDuration=1min。失效後經 recoverDuration 自動從 failedList 過期移除，實現恢復。

3. 負載均衡策略

Random：ThreadLocalRandom.current().nextInt(list.size())
RoundRobin：AtomicInteger 計數器取模
Weighted：前綴和數組 + 二分查找
Hash：Hashing.consistentHash(key, list.size())，需調用方提供 key

4. 服務級別降級

private List getAvailableServers(boolean all) {
    List result = new ArrayList<>();
    Set failed = failoverStrategy.getFailed();
    for (ThriftServer s : serverList) if (!failed.contains(s)) result.add(s);
    if (serviceLevel == SERVERS_ONLY) return result;
    if ((all || result.isEmpty()) && !backupServerList.isEmpty()) {
        for (ThriftServer s : backupServerList) if (!failed.contains(s)) result.add(s);
    }
    if (serviceLevel == ALL_SERVERS) return result;
    if (result.isEmpty()) result.addAll(serverList); // NOT_EMPTY: 寧可用不可用節點也不拋異常
    return result;
}