Abstract

The integration scenario includes topic working modes (simple/work queue/publish/subscribe/routing via routingKey) and Confirm, return, basicAck, basicNack, and Dead Letter Exchange queue (DLX) implement delayed or scheduled tasks.

integration

Dependencies and Configuration

The following content is consumer versus producer

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4. The 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>
Copy the code
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
Copy the code

The producer configures the message queue rules

The following are two configurations. The first configuration is used in this consolidation example

@Configuration
public class TopicConfig {

   // Declare a queue
   @Bean
   public Queue topicQ1(a) {
      return new Queue("topic_sb_mq_q1");
   }
   
   // Declare a queue and bind it to a dead-letter switch (return value can be written either way)
   // To test the dead letter, you need to disable the listening on the original queue
   @Bean
   public Queue topicQ2(a) {

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


   / / declare the exchange
   @Bean
   public TopicExchange setTopicExchange(a) {
      return new TopicExchange("topicExchange");
   }

   // To declare a binding, you need to declare a routingKey
   @Bean
   public Binding bindTopicHebei1(a) {
      return BindingBuilder.bind(topicQ1()).to(setTopicExchange()).with("changsha.*");
   }
   @Bean
   public Binding bindTopicHebei2(a) {
      return BindingBuilder.bind(topicQ2()).to(setTopicExchange()).with("#.beijing"); }}Copy the code
@Configuration
public class RabbitMqConfig {

    // Define the name of the switch
    public static final String  EXCHANGE_NAME = "boot_topic_exchange";
    // Define the queue name
    public static final String QUEUE_NAME = "boot_queue";

    //1
    @Bean("bootExchange")
    public Exchange bootExchange(a){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //2
    @Bean("bootQueue")
    public Queue bootQueue(a){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //3. Bind queues to switches
    @Bean
    public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue, @Qualifier("bootExchange") Exchange exchange){
        // Topic mode compatible with broadcast mode and routing mode. With ("#") matches all subscribers in broadcast mode; With ("boot.1") matches the specified subscriber in a similar routing pattern
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs(); }}Copy the code

Producer releases message

@RestController
public class ProducerController {

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

       // Define confirm callback
       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 mode only sends messages to exchange. Distribute to all queues under exchange
       rabbitTemplate.send("topicExchange", routingKey, new Message(message.getBytes("UTF-8"),messageProperties));
       return "message sended : routingKey >"+routingKey+"; message > "+message; }}Copy the code

Consumer listening message

@Component
public class ConcumerReceiver {
    / / topic schema
    // Note that this pattern has a preference for matching. For example, send a routingKey=hunan.IT, that matches hunan.*(hunan.IT, Hunan
    @RabbitListener(queues="topic_sb_mq_q1")
    public void topicReceiveq1(String msg,Message message, Channel channel) throws IOException {
       / / message id
       long deliveryTag = message.getMessageProperties().getDeliveryTag();
       try {
          // message.getBody() todosomething

          // Sign the message
          channel.basicAck(deliveryTag, true);
       } catch (Exception e) {

          // Refuse to sign for it
          // Third parameter: requeue: requeue. If set to true, the message is returned to the queue and the broker resends the message to the consumer
          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); }}Copy the code