RabbitMQ 与 Spring整合 (Spring AMQP)

引入依赖

  <!--消息队列-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

配置文件

spring:
  application:
    name: sms-service
  rabbitmq:
    host: 192.168.236.130
    username: guest
    password: guest
    virtual-host: /guest
    template:
      #      指定默认交换机
      exchange: TEST.SMS.EXCHANGE

代码

消费者

		// 消费者
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(
                    // 队列名称
                    value = "LEYOU.SMS.QUEUE",
                    // 是否持久化
                    durable = "true"),
            exchange = @Exchange(
                    // 交换机名称 与 name属性 相同  名称尽量使用大写
                    value = "LEYOU.SMS.EXCHANGE",
                    // 忽略声明时异常 交换机名称相同但是属性不同声明的时候会出错  继续使用原来的
                    ignoreDeclarationExceptions = "true",
                    // 交换机的类型
                    type = ExchangeTypes.TOPIC
            ),
            // RoutingKey 路由key TOPIC类型可以使用通配符 * : 代表一个单词 # : 代表一个或多个单词
            key = {"verifycode.sms"}))
  /*
     * 注: 消息队列类的异常一般是相外抛出 Spring 调用ACK 就是判断该方法是否抛出异常 通过AOP实现的
     */
    public void sendSms(Map<String, String> data) throws ClientException {
        // 判断参数
        if (data == null) {
            return;
        }
        // 手机号
        String phone = data.get("phone");
        // 验证码
        String code = data.get("code");

        // 判断参数
        if (StringUtils.isBlank(phone) || StringUtils.isBlank(code)) {
            return;
        }

        // 发送短信
        smsUtil.sendSms(phone, code, smsProperties.getSignName(), smsProperties.getVerifyCodeTemplate());

    }

@Component
@RabbitListener(queues = "itcast")
public class Customer1 {
    
    @RabbitHandler
    public void getMsg (String msg){
        System.out.println("msg = " + msg);
    }
}

在监听的方法上可以加上一个参数修改信道的参数

		// 消费者
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(
                    // 队列名称
                    value = "TEST.SMS.QUEUE",
                    // 是否持久化
                    durable = "true"),
            exchange = @Exchange(
                    // 交换机名称 与 name属性 相同  名称尽量使用大写
                    value = "TEST.SMS.EXCHANGE",
                    // 忽略声明时异常 交换机名称相同但是属性不同声明的时候会出错  继续使用原来的
                    ignoreDeclarationExceptions = "true",
                    // 交换机的类型
                    type = ExchangeTypes.TOPIC
            ),
            // RoutingKey 路由key TOPIC类型可以使用通配符 * : 代表一个单词 # : 代表一个或多个单词
            key = {"verifycode.sms"}))
  /*
     * 注: 消息队列类的异常一般是相外抛出 Spring 调用ACK 就是判断该方法是否抛出异常 通过AOP实现的
     */
public void save(Long id, Channel channel) throws IOException {
  // 设置这个消费者每次这能处理一条消息  实现能者多劳 参考: 乐优商城15天
  channel.basicQos(1);
    if(id == null){
        return;
    }
    goodsHtmlService.createHtml(id);
}

生产者

    // 消息队列
    @Autowired
    private AmqpTemplate amqpTemplate;


   // 发送消息  参数 RoutingKey 和 参数 也可以有第三个参数是指交换机
    amqpTemplate.convertAndSend("verifycode.sms", map);