import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
private SubscribableChannel provisionDestination(String name, boolean pubSub) {
	String destinationName = name + ".destination";
	SubscribableChannel destination = this.provisionedDestinations.get(destinationName);
	if (destination == null) {
		destination = pubSub ? new PublishSubscribeChannel() : new DirectChannel();
		((AbstractMessageChannel)destination).setBeanName(destinationName);
		((AbstractMessageChannel)destination).setComponentName(destinationName);
		this.provisionedDestinations.put(destinationName, destination);
	}
	return destination;
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
private SpringIntegrationChannelBinder createBinder() {
	SpringIntegrationProvisioner provisioningProvider = new SpringIntegrationProvisioner();
	SpringIntegrationChannelBinder binder = new SpringIntegrationChannelBinder(provisioningProvider);
	this.context.registerBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, PublishSubscribeChannel.class);
	this.context.refresh();
	binder.setApplicationContext(this.context);
	return binder;
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
@Bean
public IntegrationFlow storeBookPageByPageErrorChannelHandler(PublishSubscribeChannel storeBookPageByPageErrorChannel,
                                                              ConnectionFactory connectionFactory,
                                                              @Qualifier("createBookResultQueue") Destination createBookResultQueue){
    return integrationFlowDefinition -> integrationFlowDefinition.channel(storeBookPageByPageErrorChannel)
            .handleWithAdapter(adapters -> adapters.jms(connectionFactory).destination(createBookResultQueue));

}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
/**
     * SubscribableChannel for Axon CQRS to use
     * @return
     */
    @Bean(name = "defaultInputChannel")
    public PublishSubscribeChannel defaultInputChannel() {
        PublishSubscribeChannel channel = new PublishSubscribeChannel();
        List<ChannelInterceptor> list = new ArrayList<>(1);
        list.add(messageSelectingInterceptor());
        channel.setInterceptors(list);

//        channel.setDatatypes(Object.class); // we've defined it using the PayloadTypeSelector instead and injected it as an interceptor above
        return channel;
    }
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
/**
 * This channel can be used to monitor messages from our main channel without interrupting it.
 * @return
 */
@Bean(name = "monitoringChannel")
public PublishSubscribeChannel monitoringChannel() {
    PublishSubscribeChannel channel = new PublishSubscribeChannel();
    List<ChannelInterceptor> list = new ArrayList<>(1);
    list.add(wireTap());
    channel.setInterceptors(list);

    return channel;
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
@Bean
public MessageChannel officeStream() {
    return new PublishSubscribeChannel();
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
@Bean
public MessageChannel outsideStream() {
    return new PublishSubscribeChannel();
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
@Bean
public MessageChannel temperatureStream() {
    return new PublishSubscribeChannel();
}
 

import org.springframework.integration.channel.PublishSubscribeChannel; //导入依赖的package包/类
@Bean
public PublishSubscribeChannel storeBookPageByPageErrorChannel(){
    return new PublishSubscribeChannel();
}