spring boot rabbitmq

rabbitmq是消息队列中的一种，以下是rabbitmq在spring boot中的集成和实现

spring boot中要集成rabbitmq很简单，在maven添加依赖就可以了

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

接下来是rabbit一些属性配置

srping:
    rabbitmq:
        host: 127.0.0.1
        port: 15672
        username: guest
        password: guest

消息队列和交换机配置

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;

@Configuration
public class RabbitConfig {

    //声明队列
    @Bean
    public Queue queue1() {
        return new Queue("queue1", true); // true表示持久化该队列
    }

    @Bean
    public Queue queue2() {
        return new Queue("queue2");
    }

    //交换机
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }

    //绑定
    @Bean
    public Binding binding1() {
        //通过key绑定
        return BindingBuilder.bind(queue1()).to(topicExchange()).with("key1");
    }


    @Bean
    public Binding binding2() {
        return BindingBuilder.bind(queue2()).to(topicExchange()).with("key2");
    }
}

我配置了两个队列分别和交换机进行绑定，再来配置队列发送者和接受者

@Component
public class Product {

    @Autowired
    private AmqpTemplate rabbitTemplate;

    public void send1(String message) {
        System.out.println("queue1发送消息：" + message);
        rabbitTemplate.convertAndSend("topicExchange", "key1", message);
    }

    public void send2(String message) {
        System.out.println("queue2发送消息：" + message);
        rabbitTemplate.convertAndSend("topicExchange", "key2", message);
    }
}

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

@Component
public class Customer {

    @RabbitListener(queues = "queue1")
    public void consumeMessage1(String message) {
        System.out.println("queue1接收消息："+ message);
    }

    @RabbitListener(queues = "queue2")
    public void consumeMessage2(String message) {
        System.out.println("queue2接收消息："+ message);
    }
}

配置完后进行单元测试

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringBootRabbitmqApplicationTests {

    @Autowired
    private Product product;

    @Test
    public void send() {
        for (int i = 0; i < 10; i++) {
            product.send1("t1--" + i);
            product.send2("t2--" + i);
        }
    }
}

单元测试结果

result.png

可能会出现消息发送完毕但是没有被消费完的情况，因为消息发送到队列后单元测试就结束了，如果重新进行单元测试，queue1被持久化所以上一次的消息能继续被消费queue2不能