一、搭建初始环境

1.引入依赖

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

2. 配置

spring:
  application:
    name: rabbitmq-springboot
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: test
    password: test
    virtual-host: test

二、Hello World

1.生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testHello() {
        rabbitTemplate.convertAndSend("hello", "hello world");
    }

2.消费者

@Component
@RabbitListener(queuesToDeclare = @Queue("hello"))
public class HelloMq {
    @RabbitHandler
    public void receive(String message) {
        System.out.println("message:" + message);
    }
}

运行结果

三、Work queues

1.生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testWork() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("work", "hello work!" + i);
        }
    }

2.消费者

@Component
public class Work {
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive1(String message) {
        System.out.println("message1:" + message);
    }
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive2(String message) {
        System.out.println("message2:" + message);
    }
}

运行结果

说明:默认在Spring AMQP实现中Work这种方式就是公平调度,如果需要实现能者多劳需要额外配置。

四、Publish/Subscribe

1.生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testFanout() throws InterruptedException {
        rabbitTemplate.convertAndSend("logs", "", "这是⽇志⼴播");
    }

2.消费者

@Component
public class Fanout {
    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(name = "logs", type = "fanout")))
    public void receive1(String message) {
        System.out.println("message1=" + message);
    }
    @RabbitListener(bindings = @QueueBinding(value = @Queue, exchange = @Exchange(value = "logs", type = "fanout")))
    public void receive2(String message) {
        System.out.println("message2=" + message);
    }
}

运行结果

五、Routing

1.生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;    
    @Test
    public void testDirect() {
        rabbitTemplate.convertAndSend("directs", "info", "info⽇志信息");
        rabbitTemplate.convertAndSend("directs", "warn", "warn⽇志信息");
        rabbitTemplate.convertAndSend("directs", "error", "error⽇志信息");
    }

2.消费者

@Component
public class RoutingKey {
    @RabbitListener(bindings = @QueueBinding(value = @Queue, key = {"info", "warn", "error"}, exchange = @Exchange(name = "directs", type = "direct")))
    public void receive1(String message) {
        System.out.println("message1 = " + message);
    }
    @RabbitListener(bindings = @QueueBinding(value = @Queue, key = {"error"}, exchange = @Exchange(name = "directs", type = "direct")))
    public void receive2(String message) {
        System.out.println("message2 = " + message);
    }
}

运行结果

五、Routing

1.生产者

    @Autowired
    private RabbitTemplate rabbitTemplate; 
    @Test
    public void testTopic() {
        rabbitTemplate.convertAndSend("topics", "user.save","user.save消息");
        rabbitTemplate.convertAndSend("topics", "user.save.findAll","user.save.findAll消息");
    }

2.消费者

@Component
public class Topic {
    @RabbitListener(bindings = @QueueBinding(value = @Queue, key = "user.*", exchange = @Exchange(name = "topics", type = "topic")))
    public void receive1(String message) {
        System.out.println("message1 = " + message);
    }
    @RabbitListener(bindings = @QueueBinding(value = @Queue, key = "user.#", exchange = @Exchange(name = "topics", type = "topic")))
    public void receive2(String message) {
        System.out.println("message2 = " + message);
    }
}

运行结果

六、参考：

MQ消息中间件之RabbitMQ以及整合SpringBoot2.x实战教程