<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring: 
 rabbitmq: 
 host: localhost 
 port: 5672 
 username: guest 
 password: guest

  @Configuration
public class RabbitP2PConfigure {  
 public static final String QUEUE_NAME = "p2p-queue";
  @Bean
  public Queue queue() {
    return new Queue(QUEUE_NAME, true);
  }

}

@RunWith(SpringRunner.class)
@SpringBootTest(classes = BootCoreTestApplication.class)
@Slf4j
public class RabbitTest {
  @Autowired
  private AmqpTemplate amqpTemplate;

  /**
  * 发送
  */
  @Test
  public void sendLazy() throws InterruptedException {
    City city = new City(234556666L, "direct_name", "direct_code");
    amqpTemplate.convertAndSend(RabbitLazyConfigure.QUEUE_NAME, city);
  }
  
  /**
  * 领取
  */
  @Test
  public void receive() throws InterruptedException {
    Object obj = amqpTemplate.receiveAndConvert(RabbitLazyConfigure.QUEUE_NAME);
    Assert.notNull(obj, "");
    log.debug(obj.toString());
  }
}

@Configuration
public class RabbitFanoutConfigure {
  public static final String EXCHANGE_NAME = "fanout-exchange";
  public static final String FANOUT_A = "fanout.A";
  public static final String FANOUT_B = "fanout.B";
  public static final String FANOUT_C = "fanout.C";

  @Bean
  public Queue AMessage() {
    return new Queue(FANOUT_A);
  }

  @Bean
  public Queue BMessage() {
    return new Queue(FANOUT_B);
  }

  @Bean
  public Queue CMessage() {
    return new Queue(FANOUT_C);
  }

  @Bean
  public FanoutExchange fanoutExchange() {
    return new FanoutExchange(EXCHANGE_NAME);
  }

  @Bean
  public Binding bindingExchangeA(Queue AMessage, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(AMessage).to(fanoutExchange);
  }

  @Bean
  public Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(BMessage).to(fanoutExchange);
  }

  @Bean
  public Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(CMessage).to(fanoutExchange);
  }

}

@Slf4j
public class Sender {

  @Autowired
  private AmqpTemplate rabbitTemplate;

  public void sendFanout(Object message) {
    log.debug("begin send fanout message<" + message + ">");
    rabbitTemplate.convertAndSend(RabbitFanoutConfigure.EXCHANGE_NAME, "", message);
  }

}

@Slf4j
@RabbitListener(queues = {
    RabbitFanoutConfigure.FANOUT_A,
    RabbitFanoutConfigure.FANOUT_B,
    RabbitFanoutConfigure.FANOUT_C
})
public class Receiver {

  @RabbitHandler
  public void receiveMessage(String message) {
    log.debug("Received <" + message + ">");
  }
}

@Configuration
public class RabbitTopicConfigure {
  public static final String EXCHANGE_NAME = "topic-exchange";
  public static final String TOPIC = "topic";
  public static final String TOPIC_A = "topic.A";
  public static final String TOPIC_B = "topic.B";

  @Bean
  public Queue queueTopic() {
    return new Queue(RabbitTopicConfigure.TOPIC);
  }

  @Bean
  public Queue queueTopicA() {
    return new Queue(RabbitTopicConfigure.TOPIC_A);
  }

  @Bean
  public Queue queueTopicB() {
    return new Queue(RabbitTopicConfigure.TOPIC_B);
  }

  @Bean
  public TopicExchange exchange() {
    TopicExchange topicExchange = new TopicExchange(EXCHANGE_NAME);
    topicExchange.setDelayed(true);
    return new TopicExchange(EXCHANGE_NAME);
  }

  @Bean
  public Binding bindingExchangeTopic(Queue queueTopic, TopicExchange exchange) {
    return BindingBuilder.bind(queueTopic).to(exchange).with(RabbitTopicConfigure.TOPIC);
  }

  @Bean
  public Binding bindingExchangeTopics(Queue queueTopicA, TopicExchange exchange) {
    return BindingBuilder.bind(queueTopicA).to(exchange).with("topic.#");
  }
}

@Slf4j
@RabbitListener(queues = {
    RabbitTopicConfigure.TOPIC,
    RabbitTopicConfigure.TOPIC_A,
    RabbitTopicConfigure.TOPIC_B
})
public class Receiver {
  @RabbitHandler
  public void receiveMessage(String message) {
    log.debug("Received <" + message + ">");
  }
}

@RunWith(SpringRunner.class)
@SpringBootTest(classes = BootCoreTestApplication.class)
public class RabbitTest {
  @Autowired
  private AmqpTemplate rabbitTemplate;

  @Test
  public void sendAll() {
    rabbitTemplate.convertAndSend(RabbitTopicConfigure.EXCHANGE_NAME, "topic.test", "send All");
  }

  @Test
  public void sendTopic() {
    rabbitTemplate.convertAndSend(RabbitTopicConfigure.EXCHANGE_NAME, RabbitTopicConfigure.TOPIC, "send Topic");
  }

  @Test
  public void sendTopicA() {
    rabbitTemplate.convertAndSend(RabbitTopicConfigure.EXCHANGE_NAME, RabbitTopicConfigure.TOPIC_A, "send TopicA");
  }
}

@Configuration
public class RabbitLazyConfigure {
  public static final String QUEUE_NAME = "lazy-queue-t";
  public static final String EXCHANGE_NAME = "lazy-exchange-t";

  @Bean
  public Queue queue() {
    return new Queue(QUEUE_NAME, true);
  }

  @Bean
  public DirectExchange defaultExchange() {
    DirectExchange directExchange = new DirectExchange(EXCHANGE_NAME, true, false);
    directExchange.setDelayed(true);
    return directExchange;
  }

  @Bean
  public Binding binding() {
    return BindingBuilder.bind(queue()).to(defaultExchange()).with(QUEUE_NAME);
  }

}

@Slf4j
public class Sender {
  @Autowired
  private AmqpTemplate rabbitTemplate;
  public void sendLazy(Object msg) {
    log.debug("begin send lazy message<" + msg + ">");
    rabbitTemplate.convertAndSend(RabbitLazyConfigure.EXCHANGE_NAME,
        RabbitLazyConfigure.QUEUE_NAME, msg, message -> {
          message.getMessageProperties().setHeader("x-delay", 10000);
          return message;
        }
    );
  }
}