[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")