[java学习笔记]springboot整合rabbitmq

展示一个使用springboot整合rabbitmq发送和接收消息的简单例子：

首先创建一个springboot项目，引入依赖：

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

rabbitmq的服务主机，虚拟机，用户名和密码需要在配置文件application.yml中配置：

server:
  port: 8080
spring:
  rabbitmq:
    virtual-host: /
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
  application:
    name: ly-rabbitmq

一般来讲，对于生产者（消息发送方）而言，需要声明交换机，而对于消费者（消息接收方）而言需要声明队列，交换机，以及绑定二者；这些工作需要以bean的形式配置在spring容器中，因此增加一个配置类：

package cn.ly.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {
    // 队列名称
    public final static String QUEUE = "test_queue";
    // 交换机名称
    public final static String EXCHANG = "test_exchange";
    // 关键字
    public final static String KEY = "test_key";

    // 声明队列
    @Bean(QUEUE)
    public Queue queue() {
        return new Queue(QUEUE);
    }

    // 声明交换机
    @Bean(EXCHANG)
    public Exchange exchange() {
        return ExchangeBuilder.directExchange(EXCHANG).durable(true).build();
    }

    // 绑定交换机和队列
    @Bean
    public Binding binding(@Qualifier(EXCHANG) Exchange exchange,@Qualifier(QUEUE) Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with(KEY).noargs();
    }
}

有不少同名不同包的类，注意引类的正确；

接收的消息可以是字符串，或者一个序列化后的对象，这里定义一个即将被发送的对象的类：

package cn.ly.rabbitmq.doamin;
import lombok.Data;
import lombok.ToString;

import java.io.Serializable;

@Data
@ToString
public class User implements Serializable{
    private String username;
    private String password;
}

消息接收：

定义一个方法来监听队列：

package cn.ly.rabbitmq.mq;

import cn.ly.rabbitmq.config.RabbitConfig;
import cn.ly.rabbitmq.doamin.User;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class TestMq {
    // 监听队列
    @RabbitListener(queues = {RabbitConfig.QUEUE})
    public void lisen(User msg, Message message, Channel channel) {
        // msg是收到的信息，message则是以固定方式封装的信息
        // 这里发送的消息是一个序列化后的对象，一般也可以使用jsonString传递消息，这样接收的话可以使用string接收
        log.info("recive a massage:" + msg);
    }
}

此时消费端便完成了，只要队列接收到消息，就会调用监听程序；

发送消息，这里定义了一个controller来发送消息：

package cn.ly.rabbitmq.controller;

import cn.ly.rabbitmq.config.RabbitConfig;
import cn.ly.rabbitmq.doamin.User;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class TestController {
    @Autowired
    RabbitTemplate rabbitTemplate;

    // 演示一下发送消息\
    @GetMapping("/test")
    public String test() {
        // 发送一个对象
        User user = new User();
        user.setPassword("1111");
        user.setUsername("ly");
        // 发送消息，需要指定交换机，关键字，消息的格式是自己任定义的，可以是json字符串，也可以是一个序列化的对象
        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANG, RabbitConfig.KEY, user);
        return "success";
    }
}

启动后访问localhost:8080/test，即可看到输出日志：

2019-07-16 17:17:48.180  INFO 9012 --- [ntContainer#0-1] cn.ly.rabbitmq.mq.TestMq                 : recive a massage:User(username=ly, password=1111)

多线程处理消息

值得注意的一点是：默认的情况下，消费端处理消息是单线程的，即，如果在执行监听程序时，再次收到消息就会等到监听程序执行完成后才会处理下一条消息，如果需要多线程并行处理多条消息的话，可以对程序进行如下改造：

RabbitConfig配置类中添加如下配置：

@Bean("customContainerFactory")
    public SimpleRabbitListenerContainerFactory containerFactory
    (SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        // 10是并发数量，即支持的同时处理的线程数量
        factory.setConcurrentConsumers(10);
        factory.setMaxConcurrentConsumers(10);
        configurer.configure(factory, connectionFactory);
        return factory;
    }

在监听方法上的注解添加属性：

@RabbitListener(queues = {RabbitConfig.QUEUE},containerFactory = "customContainerFactory")