springboot~rabbitmq的队列初始化和绑定

Posted 2021-01-13 敢于对过去告一个段落，才有信心掀开新的篇章！

配置文件，在rabbit中自动建立exchange,queue和绑定它们的关系

1. 代码里初始化exchange
2. 代码里初始化queue
3. 代码里绑定exchange,queue和routekey
4. 配置文件，直接声明vhost

代码里初始化exchange

   /**
   * rabbitMq里初始化exchange.
   *
   * @return
   */
  @Bean
  public TopicExchange crmExchange() {
    return new TopicExchange(EXCHANGE);
  }

代码里初始化queue

  /**
   * rabbitMq里初始化队列crm.hello.
   *
   * @return
   */
  @Bean
  public Queue helloQueue() {
    return new Queue(HELLO);
  }

代码里绑定exchange,queue和routekey

  /**
   * 绑定exchange & queue & routekey.
   *
   * @param queueMessage 队列
   * @param exchange     交换机
   * @param routekey     路由
   * @return
   */
  public Binding bindingExchange(Queue queueMessage, TopicExchange exchange, String routekey) {
    return BindingBuilder.bind(queueMessage).to(exchange).with(routekey);
  }

配置文件

spring:
    rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: lind

完整代码

package com.lind.microservice.productCenter.mq;

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * amqp配置.
 */
@Configuration
public class AmqpConfig {

  /**
   * 交换机.
   */
  public final static String EXCHANGE = "crm";
  /**
   * hello队列.
   */
  public final static String HELLO = "crm.hello";
  /**
   * 建立订单队列.
   */
  public final static String LIND_GENERATE_ORDER = "crm.generate.order";


  /**
   * 绑定exchange & queue & routekey.
   *
   * @param queueMessage 队列
   * @param exchange     交换机
   * @param routekey     路由
   * @return
   */
  public Binding bindingExchange(Queue queueMessage, TopicExchange exchange, String routekey) {
    return BindingBuilder.bind(queueMessage).to(exchange).with(routekey);
  }


  /**
   * rabbitMq里初始化exchange.
   *
   * @return
   */
  @Bean
  public TopicExchange crmExchange() {
    return new TopicExchange(EXCHANGE);
  }

  /**
   * rabbitMq里初始化队列crm.hello.
   *
   * @return
   */
  @Bean
  public Queue helloQueue() {
    return new Queue(HELLO);
  }

  /**
   * rabbitMq里初始化队列crm.generate.order.
   *
   * @return
   */
  @Bean
  public Queue orderQueue() {
    return new Queue(LIND_GENERATE_ORDER);
  }

}

队列发布者

package com.lind.microservice.productCenter.mq;

import java.util.Date;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

@Configuration
public class HelloPublisher {
  @Autowired
  AmqpTemplate rabbitTemplate;
  @Autowired
  AmqpConfig amqpConfig;

  public void hello() {
    String context = "hello " + new Date();
    System.out.println("HelloPublisher : " + context);
    amqpConfig.bindingExchange(
        amqpConfig.helloQueue(),
        amqpConfig.crmExchange(),
        "crm.hello.#"
    );
    this.rabbitTemplate.convertAndSend(AmqpConfig.EXCHANGE, AmqpConfig.HELLO, context);
  }


}

队列订阅者

package com.lind.microservice.productCenter.mq;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = AmqpConfig.HELLO)
public class HelloSubscriber {
  @RabbitHandler
  public void process(String hello) {
    System.out.println("HelloSubscriber  : " + hello);
  }

}