嘿，针对你在Spring Integration里用毒丸模式停止生产者和消费者的场景，我来给你拆解一套完全贴合你架构的可行方案——毕竟你是N个继承Supplier的生产者、M个继承GenericHandler的消费者，用无界队列连接，还要精准控制毒丸数量和生产者停止对吧？

首先得明确：毒丸是给消费者的「停止信号」，每个消费者至少需要收到1个毒丸才会停止，所以总共需要发M个毒丸。你说的每个生产者发M/N个毒丸是完全合理的——如果M能被N整除就平均分配，不能的话就让前几个生产者多分担1个，确保总数凑够M个。

另外，生产者发完K条业务消息+分配的毒丸后，必须彻底停止，不能再被TaskExecutor调度，这一步要用到Spring Integration的专属停止机制。

你的生产者是Supplier类型，要在发送K条消息后触发毒丸发送，然后自我停止：

1. 定义毒丸标记：用一个全局唯一的对象当毒丸，避免和业务消息混淆：

   public static final Object POISON_PILL = new Object();
   

2. 修改Supplier的核心逻辑：
   给每个生产者加计数器和运行状态标记，发完K条业务消息后开始发毒丸，全部发完后抛出StopPollingException——这是Spring Integration官方推荐的停止Poller的方式，抛出后这个Supplier就不会再被你的supplierExecutor调度了，彻底停止。

   @Component
   public class BusinessMessageSupplier implements Supplier<Message<?>> {
       private static final Object POISON_PILL = new Object();
       private final AtomicInteger businessMsgCount = new AtomicInteger(0);
       private final AtomicInteger poisonPillCount = new AtomicInteger(0);
       private final AtomicBoolean isRunning = new AtomicBoolean(true);
       
       // 可配置的参数：每个生产者要发的业务消息数、分配的毒丸数
       private final int BUSINESS_MSG_LIMIT = 100; // 对应你的K
       private final int POISON_PILLS_TO_SEND = 2; // 比如M=6，N=3，每个发2个
   
       @Override
       public Message<?> get() {
           if (!isRunning.get()) {
               throw new StopPollingException("Producer has finished all tasks");
           }
   
           int currentBizCount = businessMsgCount.incrementAndGet();
           if (currentBizCount <= BUSINESS_MSG_LIMIT) {
               // 发送业务消息
               return MessageBuilder.withPayload("Biz Msg " + currentBizCount).build();
           } else {
               int currentPoisonCount = poisonPillCount.incrementAndGet();
               if (currentPoisonCount <= POISON_PILLS_TO_SEND) {
                   // 发送毒丸
                   return MessageBuilder.withPayload(POISON_PILL).build();
               } else {
                   // 所有消息发完，标记停止并抛出停止异常
                   isRunning.set(false);
                   throw new StopPollingException("All business messages and poison pills sent");
               }
           }
       }
   }
   

3. 配置生产者的Poller和TaskExecutor：
   确保你的supplierExecutor绑定到生产者的Poller上，让生产者在指定线程池里运行：

   @Bean
   public IntegrationFlow producerFlow(BusinessMessageSupplier supplier, 
                                       @Qualifier("supplierExecutor") Executor supplierExecutor) {
       return IntegrationFlow.from(supplier, spec -> spec
               .poller(Pollers.fixedDelay(100) // 按需调整轮询间隔
                       .taskExecutor(supplierExecutor)))
               .channel("unboundedQueueChannel") // 连接到你的无界队列
               .get();
   }
   
   @Bean(name = "supplierExecutor")
   public Executor supplierExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
       executor.setCorePoolSize(3); // 对应你的N个生产者
       executor.setMaxPoolSize(3);
       executor.setThreadNamePrefix("producer-thread-");
       executor.initialize();
       return executor;
   }
   

消费者是GenericHandler类型，要能识别毒丸，收到后停止自己，并且确保所有消费者都收到毒丸后完成整体停止：

1. 修改GenericHandler的处理逻辑：
   用原子类维护全局毒丸计数器，确保多线程安全。收到毒丸后标记当前消费者停止，当全局计数器达到M时，就说明所有消费者都收到了停止信号：

   @Component
   public class BusinessMessageHandler implements GenericHandler<Object> {
       private static final Object POISON_PILL = new Object();
       private final AtomicBoolean isRunning = new AtomicBoolean(true);
       // 全局毒丸计数器，多消费者共享
       private static final AtomicInteger globalPoisonPillCount = new AtomicInteger(0);
       private final int TOTAL_CONSUMERS = 6; // 对应你的M
   
       @Override
       public Object handle(Object payload, MessageHeaders headers) {
           if (!isRunning.get()) {
               return null;
           }
   
           if (POISON_PILL.equals(payload)) {
               // 收到毒丸，停止当前消费者
               isRunning.set(false);
               int currentGlobalCount = globalPoisonPillCount.incrementAndGet();
               if (currentGlobalCount == TOTAL_CONSUMERS) {
                   System.out.println("All consumers have received poison pills, consumer flow is stopping");
                   // 这里可以按需扩展：比如关闭队列通道、通知Spring容器停止相关组件等
               }
               return null;
           } else {
               // 处理业务消息逻辑
               System.out.println("Processing: " + payload);
               return payload;
           }
       }
   }
   

2. 配置消费者的IntegrationFlow：
   把消费者绑定到无界队列，并用指定线程池调度：

   @Bean
   public IntegrationFlow consumerFlow(BusinessMessageHandler handler,
                                       @Qualifier("consumerExecutor") Executor consumerExecutor) {
       return IntegrationFlow.from("unboundedQueueChannel")
               .handle(handler, spec -> spec.taskExecutor(consumerExecutor))
               .get();
   }
   
   @Bean(name = "consumerExecutor")
   public Executor consumerExecutor() {
       ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
       executor.setCorePoolSize(6); // 对应你的M个消费者
       executor.setMaxPoolSize(6);
       executor.setThreadNamePrefix("consumer-thread-");
       executor.initialize();
       return executor;
   }
   

• 毒丸数量要精准：必须确保总毒丸数等于消费者数量M，不然会出现有的消费者一直等待消息的情况。如果N不能整除M，比如M=5、N=2，就让第一个生产者发3个，第二个发2个，总数凑够5。
• 线程安全不能忘：所有计数器、运行状态标记都要用AtomicInteger/AtomicBoolean，避免多线程下的并发问题。
• 生产者停止的正确姿势：一定要用StopPollingException，而不是自己手动中断线程——这是Spring Integration Poller的标准停止方式，能确保生产者彻底脱离TaskExecutor的调度。
• 无界队列的处理顺序：因为毒丸是生产者最后发送的，所以消费者会先处理完所有业务消息，再处理毒丸，完全符合预期。

如果你的N和M是配置化的（不是硬编码），可以在启动时动态计算每个生产者的毒丸数：

@Value("${producer.count}")
private int producerCount;

@Value("${consumer.count}")
private int consumerCount;

private int poisonPillsPerProducer;
private int remainder;

@PostConstruct
public void init() {
    poisonPillsPerProducer = consumerCount / producerCount;
    remainder = consumerCount % producerCount;
    // 给前remainder个生产者多分配1个毒丸，比如用生产者的序号来判断
}

这样就能灵活适配不同的部署配置了。

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