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);