导入依赖
<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");
}
}
}