第四章: SpringBoot整合RabbitMQ

导入依赖

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

编写配置文件

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    virtual-host: /
    username: guest
    password: guest

编写配置类

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 RabbitMQConfig {

    /**
     * 创建交换机
     * @return
     */
    @Bean("testExchange")
    public Exchange testExchange() {
        return ExchangeBuilder.topicExchange("test_exchange").durable(true).build();
    }

    /**
     * 创建队列
     */
    @Bean("testQueue1")
    public Queue testQueue1() {
        return QueueBuilder.durable("test_queue_1").build();
    }
    @Bean("testQueue2")
    public Queue testQueue2() {
        return QueueBuilder.durable("test_queue_2").build();
    }
    @Bean("testQueue3")
    public Queue testQueue3() {
        return QueueBuilder.durable("test_queue_3").build();
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(@Qualifier("testQueue1") Queue queue, @Qualifier("testExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("#.world").noargs();
    }
    @Bean
    public Binding bindingQueue2(@Qualifier("testQueue2") Queue queue, @Qualifier("testExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("#.world").noargs();
    }
    @Bean
    public Binding bindingQueue3(@Qualifier("testQueue3") Queue queue, @Qualifier("testExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("#.world").noargs();
    }

}

编写监听器

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

@Component
public class QueueListener1 {

    @RabbitListener(queues = "test_queue_1")
    public void handler(Message message) {
        String consumerQueue = message.getMessageProperties().getConsumerQueue();
        System.out.println(consumerQueue + ":" + new String(message.getBody()));
    }

}
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class QueueListener2 {

    @RabbitListener(queues = "test_queue_2")
    public void handler(Message message) {
        String consumerQueue = message.getMessageProperties().getConsumerQueue();
        System.out.println(consumerQueue + ":" + new String(message.getBody()));
    }

}
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class QueueListener3 {

    @RabbitListener(queues = "test_queue_3")
    public void handler(Message message) {
        String consumerQueue = message.getMessageProperties().getConsumerQueue();
        System.out.println(consumerQueue + ":" + new String(message.getBody()));
    }

}

编写测试类

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

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

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void test() {
        for (int i = 0; i < 5; i++) {
            rabbitTemplate.convertAndSend("test_exchange", "hello.world", "hello, world");
        }
    }

}
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容