import org.springframework.integration.endpoint.MessageProducerSupport; //导入依赖的package包/类
private MessageProducerSupport createInboundAdapter(ConsumerProperties accessor, String queueName) {
	MessageProducerSupport adapter;
	int concurrency = accessor.getConcurrency();
	concurrency = concurrency > 0 ? concurrency : 1;
	if (concurrency == 1) {
		RedisQueueMessageDrivenEndpoint single = new RedisQueueMessageDrivenEndpoint(queueName,
				this.connectionFactory);
		single.setBeanFactory(getBeanFactory());
		single.setSerializer(null);
		adapter = single;
	}
	else {
		adapter = new CompositeRedisQueueMessageDrivenEndpoint(queueName, concurrency);
	}
	return adapter;
}
 

import org.springframework.integration.endpoint.MessageProducerSupport; //导入依赖的package包/类
private Binding<MessageChannel> doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel,
		MessageProducerSupport adapter, final ConsumerProperties properties) {
	DirectChannel bridgeToModuleChannel = new DirectChannel();
	bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
	bridgeToModuleChannel.setBeanName(channelName + ".bridge");
	MessageChannel bridgeInputChannel = addRetryIfNeeded(channelName, bridgeToModuleChannel, properties);
	adapter.setOutputChannel(bridgeInputChannel);
	adapter.setBeanName("inbound." + channelName);
	adapter.afterPropertiesSet();
	DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<MessageChannel>(bindingName, group, moduleInputChannel, adapter) {

		@Override
		protected void afterUnbind() {
			String key = RedisMessageChannelBinder.CONSUMER_GROUPS_KEY_PREFIX + getName();
			RedisMessageChannelBinder.this.redisOperations.boundZSetOps(key).incrementScore(getGroup(), -1);
		}
	};
	ReceivingHandler convertingBridge = new ReceivingHandler(properties);
	convertingBridge.setOutputChannel(moduleInputChannel);
	convertingBridge.setBeanName(channelName + ".bridge.handler");
	convertingBridge.afterPropertiesSet();
	bridgeToModuleChannel.subscribe(convertingBridge);
	this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).incrementScore(group, 1);
	adapter.start();
	return consumerBinding;
}
 

import org.springframework.integration.endpoint.MessageProducerSupport; //导入依赖的package包/类
@Override
protected Binding<MessageChannel> doBindConsumer(final String name, String group, MessageChannel moduleInputChannel, ConsumerProperties properties) {
	if (!StringUtils.hasText(group)) {
		group = "anonymous." + UUID.randomUUID().toString();
	}
	String queueName = groupedName(name, group);
	if (properties.isPartitioned()) {
		queueName += "-" + properties.getInstanceIndex();
	}
	MessageProducerSupport adapter = createInboundAdapter(properties, queueName);
	return doRegisterConsumer(name, group, queueName, moduleInputChannel, adapter, properties);
}