import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(userName);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);
factory.setHost(hostName);
factory.setPort(portNumber);
Connection conn = factory.newConnection(); // 커넥션 생성

.
.
.

Channel channel = conn.createChannel(); // 채널 생성

import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.ShutdownListener;

connection.addShutdownListener(new ShutdownListener() {
    public void shutdownCompleted(ShutdownSignalException cause)
    {
        ...
    }
});

package hello;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

@SpringBootApplication
public class Application {

    final static String queueName = "spring-boot";

    @Bean
    Queue queue() {
        return new Queue(queueName, false);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("spring-boot-exchange");
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(queueName);
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

    public static void main(String[] args) throws InterruptedException {
        SpringApplication.run(Application.class, args);
    }

}

org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder


public Binding<MessageChannel> doBindConsumer(
.
.
.
//익스체인지 선언
TopicExchange exchange = new TopicExchange(exchangeName);
.
.
.

org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder


public Binding<MessageChannel> doBindConsumer(
.
.
.
//큐 선언
declareQueue(queueName, queue);

.
.
.

//실제 큐를 생성하는 로직
public void declareQueue(String beanName, Queue queue) {
	try {
		this.rabbitAdmin.declareQueue(queue);
	}
	catch (AmqpConnectException e) {
		if (logger.isDebugEnabled()) {
			logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available");
		}
	}
	addToAutoDeclareContext(beanName, queue);
}

org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder

public Binding<MessageChannel> doBindConsumer(
.
.
.
//익스체인지-큐 바인딩 호출
if (partitioned) {
	String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex());
	declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey));
}
else {
	declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#"));
}

.
.
.

//실제로 익스체인지-큐 바인딩하는 부분
public void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) {
	try {
		this.rabbitAdmin.declareBinding(binding);
	}
	catch (AmqpConnectException e) {
		if (logger.isDebugEnabled()) {
			logger.debug("Declaration of binding: " + rootName + ".binding deferred - connection not available");
		}
	}
	addToAutoDeclareContext(rootName + ".binding", binding);
}

//스트림 연동하기
@EnableBinding(Sink.class)
public class listener {

  ...

  @StreamListener(Sink.INPUT)
  public void processVote(Vote vote) {
      //service 
  }
}

org.springframework.cloud.stream.binder.AbstractMessageChannelBinder

public final Binding<MessageChannel> doBindConsumer(
...
    final ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group, properties);
...

org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner


public ConsumerDestination provisionConsumerDestination(String name, String group, ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
	boolean anonymous = !StringUtils.hasText(group);
	String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName())
			: groupedName(name, group);
	if (this.logger.isInfoEnabled()) {
		this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name);
	}
	String prefix = properties.getExtension().getPrefix();
	final String exchangeName = applyPrefix(prefix, name);
	Exchange exchange = buildExchange(properties.getExtension(), exchangeName);
	if (properties.getExtension().isDeclareExchange()) {
		declareExchange(exchangeName, exchange);
	}
	String queueName = applyPrefix(prefix, baseQueueName);
	boolean partitioned = !anonymous && properties.isPartitioned();
	boolean durable = !anonymous && properties.getExtension().isDurableSubscription();
	Queue queue;
	if (anonymous) {
		queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false));
	}
	else {
		if (partitioned) {
			String partitionSuffix = "-" + properties.getInstanceIndex();
			queueName += partitionSuffix;
		}
		if (durable) {
			queue = new Queue(queueName, true, false, false,
					queueArgs(queueName, properties.getExtension(), false));
		}
		else {
			queue = new Queue(queueName, false, false, true,
					queueArgs(queueName, properties.getExtension(), false));
		}
	}
	declareQueue(queueName, queue);
	Binding binding = null;
	if (properties.getExtension().isBindQueue()) {
		binding = declareConsumerBindings(name, properties, exchange, partitioned, queue);
	}
	if (durable) {
		autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName,
				properties.getExtension());
	}
	return new RabbitConsumerDestination(queue, binding);
}

build.gradle

dependencies {
	
	.
	.
	.
	compile('org.springframework.cloud:spring-cloud-starter-stream-rabbit:1.2.0.RELEASE')

application-dev.properties

dependencies {
	
	spring.cloud.stream.bindings.SINK-INPUT-TEST.destination=news
	spring.cloud.stream.bindings.SINK-INPUT-TEST.group=서버의 유니크한 값 설정
	spring.cloud.stream.rabbit.bindings.SINK-TEST.consumer.bindingRoutingKey=newsbox.update.test
	spring.cloud.stream.rabbit.bindings.SINK-TEST.consumer.exchangeType=direct
	spring.cloud.stream.rabbit.bindings.SINK-TEST.consumer.durableSubscription=false
	spring.cloud.stream.binders.rabbit.type=rabbit
	
	//참고로 rabbitMQ 관련 설정 정보를 따로 properties 에 분리했습니다.
	spring.rabbitmq.host=${rabbitmq.host}
	spring.rabbitmq.virtual-host=testhost
	spring.rabbitmq.username=${rabbitmq.username}
	spring.rabbitmq.password=${rabbitmq.password}

listener.java
	
@EnableBinding(*****Listener.Sink.class)
public class *****Listener {

    @Autowired
    private News*****Dao news*****Dao;

    @StreamListener(Sink.inboundTest)
    public void subscribe(PojoTest pojotest) {

        if(pojotest.is***()){
            
            //뉴스 업데이트 비즈니스 로직 
           
        }
    }

    public interface Sink {
        String inboundTest = "SINK-TEST";

        @Input(inboundTest)
        SubscribableChannel *****Listener();
    }

}

org.springframework.amqp.rabbit.connection.CachingConnectionFactory

.
.

@Override
public void shutdownCompleted(ShutdownSignalException cause) {
	this.closeExceptionLogger.log(logger, "Channel shutdown", cause);
}

org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration


@Configuration
	@ConditionalOnMissingBean(ConnectionFactory.class)
	protected static class RabbitConnectionFactoryCreator {

		@Bean
		public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config)
				throws Exception {
		.
		.
		.
		//실제로 이 부분을 변경할 것입니다...
		CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
					factory.getObject());

@Bean
public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config)
        throws Exception {
        .
        .
        //라이브러스 소스와 똑같이 구현
        .
        .
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(factory.getObject()){


            //셧다웃 이벤트 발생 시 비즈니스 로직 적용
            @Override
            public void shutdownCompleted(ShutdownSignalException cause) {

                //비즈니스 로직 적용
                ***Dao.~~~; 
                
                super.shutdownCompleted(cause);
            }
        };
        .
        .
        //라이브러리 소스와 똑같이
		.

@Bean
public CachingConnectionFactory rabbitConnectionFactory(RabbitProperties config)
        throws Exception {
        .
        .
        //라이브러스 소스와 똑같이 구현
        .
        .
	//팩토리 생성 이후에 로직에 추가
	//초기 연결 시 타이아웃 60초 -> 3초
	connectionFactory.setConnectionTimeout(3000);