如何使用 JPOS Q2 Server 动态路由 ISO8583 消息？

明确结论

可以实现。JPOS Q2 并不要求所有路由都由 channel.xml 里的固定 IP 决定。XML 适合声明服务器、Channel、MUX、TransactionManager 等基础组件。具体的路由判断应放在 Java 业务代码中，例如 ISORequestListener、TransactionManager 的 TransactionParticipant 或独立的路由服务。

推荐采用以下流程：

1. 从报文的指定字段中提取路由标识。
2. 根据标识查询数据库，得到逻辑路由。
3. 从连接池或 MUX 注册表中选出对应的下游连接。
4. 通过该 MUX 发送请求并等待响应。
5. 使用原始 ISOSource 将下游响应返回给上游。

不要在每笔交易到达时修改 channel.xml，也不要让多个线程并发修改同一个共享 Channel 的 host 和 port。

推荐架构

Source
  │
  │ ISO8583 request
  ▼
QServer / ISORequestListener
  │
  ├─ 解析 identifier
  ├─ 查询或命中路由缓存
  ├─ 执行业务处理
  └─ 选择 Destination MUX
          │
          ▼
    Destination Host
          │
          │ ISO8583 response
          ▼
    Destination MUX
          │
          ▼
原始 ISOSource.send(response)
          │
          ▼
        Source

这里要区分两个概念：

• Channel 负责 TCP 连接、报文打包和收发。
• MUX 负责在同一条连接上匹配请求与响应，通常根据 MTI、字段 11、字段 37 等信息进行关联。

因此，支付网关通常不会让多个业务线程共享一个 Channel，再分别调用 receive()。更常见的做法是通过 QMUX 或其他 MUX 匹配请求和响应。

路由表设计

数据库最好返回逻辑路由名称，避免业务代码直接使用任意 IP 地址。例如：

CREATE TABLE iso_route (
    identifier      VARCHAR(32) PRIMARY KEY,
    route_name      VARCHAR(64) NOT NULL,
    host            VARCHAR(255) NOT NULL,
    port            INTEGER NOT NULL,
    enabled         BOOLEAN NOT NULL,
    timeout_ms      INTEGER NOT NULL
);

路由结果可以抽象为：

public record Route(
    String name,
    String host,
    int port,
    long timeoutMillis
) {
}

如果系统对安全性和可运维性要求较高，建议让数据库中的 identifier 只映射到预先批准的 route_name：

identifier=ACQ001 -> route_name=bank-a
identifier=ACQ002 -> route_name=bank-b

之后由 Q2 中已经配置好的 bank-a、bank-b MUX 管理连接。这样既能使用 Q2 的连接重试、日志、启停和监控能力，也能防止数据库中的错误数据把请求发往未经允许的地址。

在入口处动态选择路由

下面以 ISORequestListener 为例，假设字段 32 是路由标识。实际项目应按照接入协议选择字段，也可能使用字段 24、33、41、42、100 或复合域中的子字段。

import org.jpos.iso.ISOException;
import org.jpos.iso.ISOMsg;
import org.jpos.iso.ISORequestListener;
import org.jpos.iso.ISOSource;
import org.jpos.iso.MUX;

import java.io.IOException;
import java.util.concurrent.ExecutorService;

public class DynamicRouter implements ISORequestListener {
    private final RouteResolver routeResolver;
    private final MuxRegistry muxRegistry;
    private final ExecutorService executor;

    public DynamicRouter(
        RouteResolver routeResolver,
        MuxRegistry muxRegistry,
        ExecutorService executor
    ) {
        this.routeResolver = routeResolver;
        this.muxRegistry = muxRegistry;
        this.executor = executor;
    }

    @Override
    public boolean process(ISOSource source, ISOMsg incoming) {
        ISOMsg request = (ISOMsg) incoming.clone();

        executor.execute(() -> {
            try {
                String identifier = request.getString(32);
                if (identifier == null || identifier.isBlank()) {
                    sendError(source, request, "30");
                    return;
                }

                Route route = routeResolver.resolve(identifier);
                if (route == null) {
                    sendError(source, request, "92");
                    return;
                }

                MUX mux = muxRegistry.get(route.name());
                if (mux == null || !mux.isConnected()) {
                    sendError(source, request, "91");
                    return;
                }

                ISOMsg response = mux.request(
                    request,
                    route.timeoutMillis()
                );

                if (response == null) {
                    sendError(source, request, "68");
                    return;
                }

                source.send(response);
            } catch (Exception e) {
                try {
                    sendError(source, request, "96");
                } catch (Exception ignored) {
                    // 此处应记录原始异常和发送失败异常
                }
            }
        });

        return true;
    }

    private void sendError(
        ISOSource source,
        ISOMsg request,
        String responseCode
    ) throws ISOException, IOException {
        ISOMsg response = (ISOMsg) request.clone();
        response.setResponseMTI();
        response.set(39, responseCode);
        source.send(response);
    }
}

示例中的响应码只是为了说明控制流程。字段 39 的实际取值必须遵循接入方协议，不能在所有网络中照搬 91、92、96 或 68。

process() 返回 true，表示当前监听器已经接管这条消息。示例将下游请求交给线程池处理，避免长时间占用 QServer 的接收线程。线程池必须配置容量、队列长度和拒绝策略，不能直接使用无限队列。

查询数据库时不要为每笔交易重建连接

路由查询可以使用连接池，并配合短时间缓存：

public interface RouteResolver {
    Route resolve(String identifier) throws Exception;
}

下面是一个简化的 JDBC 实现：

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class JdbcRouteResolver implements RouteResolver {
    private static final String SQL = """
        SELECT route_name, host, port, timeout_ms
          FROM iso_route
         WHERE identifier = ?
           AND enabled = TRUE
        """;

    private final DataSource dataSource;

    public JdbcRouteResolver(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public Route resolve(String identifier) throws Exception {
        try (
            Connection connection = dataSource.getConnection();
            PreparedStatement statement = connection.prepareStatement(SQL)
        ) {
            statement.setString(1, identifier);

            try (ResultSet resultSet = statement.executeQuery()) {
                if (!resultSet.next()) {
                    return null;
                }

                return new Route(
                    resultSet.getString("route_name"),
                    resultSet.getString("host"),
                    resultSet.getInt("port"),
                    resultSet.getLong("timeout_ms")
                );
            }
        }
    }
}

实际系统应使用带过期时间的本地缓存，并提前确定数据库不可用时采取哪种策略：

• 使用最近一次有效路由；
• 直接拒绝交易；
• 切换到备用路由；
• 让部分交易进入 store-and-forward 流程。

数据库异常时，金融交易不能静默选择默认目的地，否则可能造成错误清算或资金风险。

下游连接的两种实现方式

方案一：预建多个 QMUX，由数据库结果动态选择

这是相对稳妥的方案。为每个已知目的机构配置独立的 Channel 和 QMUX：

route_name=bank-a -> MUX bank-a
route_name=bank-b -> MUX bank-b
route_name=bank-c -> MUX bank-c

代码中的注册表只需按名称返回 MUX：

import org.jpos.iso.MUX;

import java.util.Map;

public class MuxRegistry {
    private final Map<String, MUX> muxes;

    public MuxRegistry(Map<String, MUX> muxes) {
        this.muxes = Map.copyOf(muxes);
    }

    public MUX get(String routeName) {
        return muxes.get(routeName);
    }
}

项目也可以通过 Q2/NameRegistrar 获取已部署的 QMUX。不同 jPOS 版本的组件获取和注入方式可能不同，应以项目所用版本的 API 为准。原则不变：数据库返回逻辑名称，业务代码选择已经存在的 MUX。

这种方式有以下优点：

• 可以复用长连接；
• 请求与响应由 QMUX 负责关联；
• 连接断开后可以自动重连；
• 各目的端可以使用不同的 packager、header 和超时；
• 配置、日志和运维的边界比较清楚。

方案二：在运行时创建并缓存目的端连接

如果目的地址确实无法提前枚举，可以开发一个 DestinationManager：

Route(host, port, channelType, packager, timeout)
                 │
                 ▼
DestinationManager.getOrCreate(route)
                 │
                 ▼
       Channel + MUX + lifecycle

它至少要处理以下事项：

• 按 host + port + channelType + packager 缓存连接；
• 限制动态目的端的数量；
• 检查地址白名单；
• 管理连接的建立、重连和关闭；
• 防止多个线程重复创建同一连接；
• 清理长期空闲的连接；
• 为每个连接提供请求与响应的关联机制；
• 配置更新后安全替换旧连接。

只执行下面这些操作，还不足以构成可靠的支付路由器：

channel.setHost(route.host(), route.port());
channel.connect();
channel.send(message);
ISOMsg response = channel.receive();

共享 Channel 在并发交易中会产生竞态。一个线程可能在另一个线程收到响应之前修改目标地址，多个线程也无法可靠判断响应属于哪个请求。如果每笔交易都新建 TCP 连接，性能、连接数量、登录握手和故障恢复又会成为问题。

如何让响应沿原路径返回

入口收到请求时，ISOSource source 代表该请求所使用的上游连接。下游响应到达后，只需调用：

source.send(response);

响应便会通过原始上游连接返回，无须再根据 IP 查询一次反向路由。

这需要满足以下条件：

1. 下游请求必须通过 MUX 发送，以便把响应匹配到正确的请求。
2. 处理完成前，上游连接必须保持有效。
3. 总处理时间不能超过上游规定的超时时间。
4. 不能只把 ISOSource、请求报文或关联信息保存在无界的内存集合中。
5. 如果处理过程会跨线程、跨进程或持续较长时间，应保存明确的关联信息，不能只依赖 Java 对象引用。

在 TransactionManager 架构中，这些内容通常会放入 Context：

SOURCE
REQUEST
ROUTE
DESTINATION_REQUEST
DESTINATION_RESPONSE

随后由多个 TransactionParticipant 依次完成校验、查询路由、报文转换、发送下游、处理响应和回送上游。对于需要审计、超时控制、幂等处理、冲正和 store-and-forward 的支付系统，这种结构通常比把所有逻辑集中在一个监听器中更容易维护。

必须处理好报文关联

动态路由不只是选择 IP，还要保证每笔交易都能正确关联。至少需要检查：

• 字段 11（STAN）在各目的端是否唯一；
• 字段 37（Retrieval Reference Number）是保留还是重新生成；
• 请求 MTI 与响应 MTI 是否正确对应；
• 下游是否会修改重要的关联字段；
• 相同 STAN 是否可能同时发送到不同目的端；
• 如何处理超时后到达的响应；
• 冲正报文是否必须发送到原交易所使用的目的端。

冲正尤其不能只根据当前路由表重新计算目的地。路由配置可能在原交易完成后发生变化，因此要在交易日志中保存原始目的路由，让冲正、撤销或查询继续使用该路由。

实施顺序

可以按以下顺序推进：

1. 确定用于路由的 ISO8583 字段及其合法值。
2. 建立 identifier -> route_name 路由表。
3. 为已知目的端配置独立的 Channel 和 QMUX。
4. 实现 RouteResolver，加入数据库连接池和有界缓存。
5. 实现入口监听器或 TransactionManager participant。
6. 使用 mux.request() 发送下游请求并设置超时。
7. 通过原始 ISOSource.send() 回送响应。
8. 加入并发限制、熔断、日志、审计和迟到响应处理。
9. 保存每笔交易实际选择的路由，用于冲正和问题追踪。

如果下游机构数量有限，采用”静态配置连接、动态选择 MUX”通常就能满足动态路由需求。只有当目的地址数量很大或变化频繁时，才有必要实现运行时 Channel/MUX 管理器。

备注：内容仅供参考。