<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
    </parent>

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

server.port=8090
spring.rabbitmq.host=192.168.168.10
spring.rabbitmq.port=5672
spring.rabbitmq.username=zheng123
spring.rabbitmq.password=zheng123
spring.rabbitmq.virtual-host=/zheng
spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.listener.direct.acknowledge-mode=manual

@Configuration
public class TopicConfig {

   // 聲明隊列
   @Bean
   public Queue topicQ1() {
      return new Queue("topic_sb_mq_q1");
   }
   
   // 聲明隊列并綁定該隊列到死信交換機(返回值有兩種寫法，任選一種都可以)
   // 測試死信需要關(guān)閉原隊列的監(jiān)聽
   @Bean
   public Queue topicQ2() {

        return QueueBuilder.durable("topic_sb_mq_q2")
              .withArgument("x-dead-letter-exchange", "topicExchange")
              .withArgument("x-dead-letter-routing-key", "changsha.f")
              .withArgument("x-message-ttl", 10000)
              .build();

        Map<String,Object> arguments = new HashMap<>(2);
        arguments.put("x-dead-letter-exchange","topicExchange");
        arguments.put("x-dead-letter-routing-key","changsha.f");
        arguments.put("x-message-ttl",10000);
        return new Queue("topic_sb_mq_q2",true,false,false,arguments);
   }


   //聲明exchange
   @Bean
   public TopicExchange setTopicExchange() {
      return new TopicExchange("topicExchange");
   }

   //聲明binding，需要聲明一個routingKey
   @Bean
   public Binding bindTopicHebei1() {
      return BindingBuilder.bind(topicQ1()).to(setTopicExchange()).with("changsha.*");
   }
   @Bean
   public Binding bindTopicHebei2() {
      return BindingBuilder.bind(topicQ2()).to(setTopicExchange()).with("#.beijing");
   }
}

@Configuration
public class RabbitMqConfig {

    //定義交換機的名字
    public static final String  EXCHANGE_NAME = "boot_topic_exchange";
    //定義隊列的名字
    public static final String QUEUE_NAME = "boot_queue";

    //1、聲明交換機
    @Bean("bootExchange")
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //2、聲明隊列
    @Bean("bootQueue")
    public Queue bootQueue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //3、隊列與交換機進行綁定
    @Bean
    public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){
        // topic模式兼容廣播模式，路由模式。with("#")則類似廣播模式匹配所有訂閱者；with("boot.1")則類似路由模式匹配指定訂閱者
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }
}

@RestController
public class ProducerController {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @GetMapping(value="/topicSend")
    public Object topicSend(String routingKey,String message) throws AmqpException, UnsupportedEncodingException {

       // 定義 confirm 回調(diào)
       rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
          if (ack) {
             // confirmed
          } else {
             // nack-ed
          }
       });

       rabbitTemplate.setMandatory(true);
       rabbitTemplate.setReturnCallback((msg, replyCode, replyText, exchange, routKey)->{
          // return message
       });

       if(null == routingKey) {
          routingKey="changsha.kf";
       }
       MessageProperties messageProperties = new MessageProperties();
       messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
       //fanout模式只往exchange里發(fā)送消息。分發(fā)到exchange下的所有queue
       rabbitTemplate.send("topicExchange", routingKey, new Message(message.getBytes("UTF-8"),messageProperties));
       return "message sended : routingKey >"+routingKey+";message > "+message;
    }
}

@Component
public class ConcumerReceiver {
    //topic 模式
    //注意這個模式會有優(yōu)先匹配原則。例如發(fā)送routingKey=hunan.IT,那匹配到hunan.*(hunan.IT,hunan.eco),之后就不會再去匹配*.ITd
    @RabbitListener(queues="topic_sb_mq_q1")
    public void topicReceiveq1(String msg,Message message, Channel channel) throws IOException {
       // 消息id
       long deliveryTag = message.getMessageProperties().getDeliveryTag();
       try {
          // message.getBody() todosomething

          // 簽收消息
          channel.basicAck(deliveryTag, true);
       } catch (Exception e) {

          // 拒絕簽收
          // 第三個參數(shù)：requeue：重回隊列。如果設(shè)置為true，則消息重新回到queue，broker會重新發(fā)送該消息給消費端
          channel.basicNack(deliveryTag, true, true);
       }
    }

    @RabbitListener(queues="topic_sb_mq_q2")
    public void topicReceiveq2(String message) {
       System.out.println("Topic模式 topic_sb_mq_q2 received  message : " +message);
    }
}