启动firstJob时多DirectChannels引发Dispatcher无订阅者异常排查

核心原因：队列复用导致消息串流 #

你的问题根源在于两个Job的RabbitMQ队列完全复用，导致firstJob运行时产生的消息被second相关的通道接收，但second通道没有订阅者（因为secondJob未启动），从而触发无订阅者异常。

具体看RabbitMqQueueConfig的实现：

• 不管是first还是second的IntegrationFlow，调用getInboundAdapter/getOutboundAdapter时，生成的队列名都是前缀 + JOB_REPLIES_QUEUE或前缀 + JOB_REQUESTS_QUEUE，没有区分first/second Job的独立标识。
• 这意味着：firstManagerInBoundFlow和secondManagerInBoundFlow监听的是同一个回复队列；firstManagerOutboundFlow和secondManagerOutboundFlow发送到同一个请求队列。

当firstJob运行时：

1. Worker节点处理完成后，将回复消息发送到共享的回复队列
2. 所有监听该队列的InboundFlow（包括secondManagerInBoundFlow）都会接收到消息
3. secondManagerInBoundFlow将消息转发到secondReplies通道，但此时secondJobManagerStep并未启动（你只启动了firstJob），secondReplies通道没有订阅者，因此抛出Dispatcher has no subscribers异常

解决方案 #

1. 为每个Job配置独立的队列 #

修改QueueConstants，为first和second Job分别定义独立的队列常量：

public class QueueConstants {
    public static final String FIRST_JOB_REQUESTS_QUEUE = "first.job.requests";
    public static final String FIRST_JOB_REPLIES_QUEUE = "first.job.replies";
    public static final String SECOND_JOB_REQUESTS_QUEUE = "second.job.requests";
    public static final String SECOND_JOB_REPLIES_QUEUE = "second.job.replies";
    // 原其他常量...
}

2. 调整RabbitMqQueueConfig的队列生成逻辑 #

修改getInboundAdapter和getOutboundAdapter方法，允许传入Job标识来区分队列：

@Override
public IntegrationFlow getInboundAdapter(boolean isManager, DirectChannel channel, String jobPrefix) {
    String queueName = isManager ? queueConstants.getConstantWithPrefix(jobPrefix + ".job.replies")
            : queueConstants.getConstantWithPrefix(jobPrefix + ".job.requests");
    return IntegrationFlow.from(Amqp.inboundAdapter(connectionFactory, queueName)).channel(channel).get();
}

@Override
public IntegrationFlow getOutboundAdapter(boolean isManager, DirectChannel channel, String jobPrefix) {
    String queueName = isManager ? queueConstants.getConstantWithPrefix(jobPrefix + ".job.requests")
            : queueConstants.getConstantWithPrefix(jobPrefix + ".job.replies");
    AmqpOutboundChannelAdapterSpec messageHandlerSpec = Amqp.outboundAdapter(defaultRabbitTemplate).routingKey(queueName);
    return IntegrationFlow.from(channel).handle(messageHandlerSpec).get();
}

3. 更新FlowConfig中的IntegrationFlow配置 #

为每个Job的Flow传入对应的jobPrefix：

// first相关Flow
@Bean("firstManagerInBoundFlow")
@Profile("manager")
public IntegrationFlow firstManagerInBoundFlow() {
    return queueConfig.getInboundAdapter(true, firstReplies(), "first");
}

@Bean("firstWorkerInBoundFlow")
@Profile("worker")
public IntegrationFlow firstWorkerInBoundFlow() {
    return queueConfig.getInboundAdapter(false, firstRequests(), "first");
}

@Bean("firstManagerOutboundFlow")
@Profile("manager")
public IntegrationFlow firstManagerOutboundFlow() {
    return queueConfig.getOutboundAdapter(true, firstRequests(), "first");
}

@Bean("firstWorkerOutboundFlow")
@Profile("worker")
public IntegrationFlow firstWorkerOutboundFlow() {
    return queueConfig.getOutboundAdapter(false, firstReplies(), "first");
}

// second相关Flow
@Bean("secondManagerInBoundFlow")
@Profile("manager")
public IntegrationFlow secondManagerInBoundFlow() {
    return queueConfig.getInboundAdapter(true, secondReplies(), "second");
}

@Bean("secondWorkerInBoundFlow")
@Profile("worker")
public IntegrationFlow secondInBoundFlow() {
    return queueConfig.getInboundAdapter(false, secondRequests(), "second");
}

@Bean("secondManagerOutboundFlow")
@Profile("manager")
public IntegrationFlow secondManagerOutboundFlow() {
    return queueConfig.getOutboundAdapter(true, secondRequests(), "second");
}

@Bean("secondWorkerOutboundFlow")
@Profile("worker")
public IntegrationFlow secondWorkerOutboundFlow() {
    return queueConfig.getOutboundAdapter(false, secondReplies(), "second");
}

这样每个Job的请求/回复队列完全独立，消息不会交叉到其他Job的通道，就能避免无订阅者的异常。

内容的提问来源于stack exchange，提问作者Dilyan Galabov