spring集成在bean之间发布订阅

Question

谢谢你提前阅读。在我的主要方法中，我有一个出版频道

@Bean(name = "feeSchedule")
public SubscribableChannel getMessageChannel() {
    return new PublishSubscribeChannel();
}

在执行长时间运行流程的服务中，它创建了一个收费计划，我将频道注入其中

@Service
public class FeeScheduleCompareServiceImpl implements FeeScheduleCompareService {

    @Autowired
    MessageChannel outChannel;

    public List<FeeScheduleUpdate> compareFeeSchedules(String oldStudyId) {
    List<FeeScheduleUpdate> sortedResultList = longMethod(oldStudyId);
    outChannel.send(MessageBuilder.withPayload(sortedResultList).build());
        return sortedResultList;
    } 
}

这就是我正在努力解决的问题。我想使用CompletableFuture并在另一个Springbean中获得未来事件的有效负载。我需要一个未来来从消息返回有效负载。我想我想创建一个ServiceActivator作为消息的endpoint，但正如我所说的，我需要它来返回未来a的有效负载。

@org.springframework.stereotype.Service
public class SFCCCompareServiceImpl implements SFCCCompareService {
     @Autowired
    private SubscribableChannel outChannel;

     @Override
    public List<SFCCCompareDTO> compareSFCC(String state, int service){
    ArrayList<SFCCCompareDTO> returnList = new ArrayList<SFCCCompareDTO>();
    CompletableFuture<List<FeeScheduleUpdate>> fa =  CompletableFuture.supplyAsync( () ->
            {  //block A   WHAT GOES HERE?!?!
                    outChannel.subscribe()
            }
    );
    CompletableFuture<List<StateFeeCodeClassification>> fb =  CompletableFuture.supplyAsync( () ->
            {  
                  return this.stateFeeCodeClassificationRepository.findAll();       
            }
    );
    CompletableFuture<List<SFCCCompareDTO>> fc = fa.thenCombine(fb,(a,b) ->{
        //block C
        //get in this block when both A & B are complete
        Object theList = b.stream().forEach(new Consumer<StateFeeCodeClassification>() {
            @Override
            public void accept(StateFeeCodeClassification stateFeeCodeClassification) {
                a.stream().forEach(new Consumer<FeeScheduleUpdate>() {
                    @Override
                    public void accept(FeeScheduleUpdate feeScheduleUpdate) {
                        returnList new SFCCCompareDTO();
                    }
                });
            }
        }).collect(Collectors.toList());
        return theList;
    });
    fc.join();
    return returnList;
}

}

我想会有一个服务激活器，比如：

@MessageEndpoint
public class UpdatesHandler implements MessageHandler{

    @ServiceActivator(requiresReply = "true")
    public List<FeeScheduleUpdate> getUpdates(Message m){
        return (List<FeeScheduleUpdate>) m.getPayload();
    }
}

Answer 1

你的问题不清楚，但我会尽力帮你提供一些信息。

Spring集成不提供CompletableFuture支持，但它提供了异步处理和回复。

有关详细信息，请参阅异步网关。另请参见异步服务激活器。

outChannel。subscribe（）应该附带MessageHandler回调。