添加依赖
代码语言：javascript复制<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>添加配置
代码语言：javascript复制spring.rabbitmq.host=192.168.2.71
spring.rabbitmq.port=5672
spring.rabbitmq.username=light
spring.rabbitmq.password=light
spring.rabbitmq.virtual-host=/test编码
配置类
代码语言：javascript复制@Configuration
public class AmqpConfirguration {
//=============简单、工作队列模式===============
public static final String SIMPLE_QUEUE = "simple_queue";
@Bean
public Queue queue() {
return new Queue(SIMPLE_QUEUE, true);
}
//===============发布/订阅模式============
public static final String PS_QUEUE_1 = "ps_queue_1";
public static final String PS_QUEUE_2 = "ps_queue_2";
public static final String FANOUT_EXCHANGE = "fanout_exchange";
@Bean
public Queue psQueue1() {
return new Queue(PS_QUEUE_1, true);
}
@Bean
public Queue psQueue2() {
return new Queue(PS_QUEUE_2, true);
}
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(FANOUT_EXCHANGE);
}
@Bean
public Binding fanoutBinding1() {
return BindingBuilder.bind(psQueue1()).to(fanoutExchange());
}
@Bean
public Binding fanoutBinding2() {
return BindingBuilder.bind(psQueue2()).to(fanoutExchange());
}
//===============路由模式============
public static final String ROUTING_QUEUE_1 = "routing_queue_1";
public static final String ROUTING_QUEUE_2 = "routing_queue_2";
public static final String DIRECT_EXCHANGE = "direct_exchange";
@Bean
public Queue routingQueue1() {
return new Queue(ROUTING_QUEUE_1, true);
}
@Bean
public Queue routingQueue2() {
return new Queue(ROUTING_QUEUE_2, true);
}
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE);
}
@Bean
public Binding directBinding1() {
return BindingBuilder.bind(routingQueue1()).to(directExchange()).with("user");
}
@Bean
public Binding directBinding2() {
return BindingBuilder.bind(routingQueue2()).to(directExchange()).with("order");
}
//===============主题模式============
public static final String TOPIC_QUEUE_1 = "topic_queue_1";
public static final String TOPIC_QUEUE_2 = "topic_queue_2";
public static final String TOPIC_EXCHANGE = "topic_exchange";
@Bean
public Queue topicQueue1() {
return new Queue(TOPIC_QUEUE_1, true);
}
@Bean
public Queue topicQueue2() {
return new Queue(TOPIC_QUEUE_2, true);
}
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(TOPIC_EXCHANGE);
}
@Bean
public Binding topicBinding1() {
return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("user.add");
}
@Bean
public Binding topicBinding2() {
return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("user.#");
}
}消息生产者：
代码语言：javascript复制@Component
public class AmqpSender {
@Autowired
private AmqpTemplate amqpTemplate;
/**
* 简单模式发送
*
* @param message
*/
public void simpleSend(String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.SIMPLE_QUEUE, message);
}
/**
* 发布/订阅模式发送
*
* @param message
*/
public void psSend(String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.FANOUT_EXCHANGE, "", message);
}
/**
* 路由模式发送
*
* @param message
*/
public void routingSend(String routingKey, String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.DIRECT_EXCHANGE, routingKey, message);
}
/**
* 主题模式发送
*
* @param routingKey
* @param message
*/
public void topicSend(String routingKey, String message) {
this.amqpTemplate.convertAndSend(AmqpConfirguration.TOPIC_EXCHANGE, routingKey, message);
}
}消息消费者：
代码语言：javascript复制@Component
public class AmqpReceiver {
/**
* 简单模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.SIMPLE_QUEUE)
public void simpleReceive(String message) {
System.out.println("接收消息:" message);
}
/**
* 发布/订阅模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.PS_QUEUE_1)
public void psReceive1(String message) {
System.out.println(AmqpConfirguration.PS_QUEUE_1 "接收消息:" message);
}
@RabbitListener(queues = AmqpConfirguration.PS_QUEUE_2)
public void psReceive2(String message) {
System.out.println(AmqpConfirguration.PS_QUEUE_2 "接收消息:" message);
}
/**
* 路由模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_1)
public void routingReceive1(String message) {
System.out.println(AmqpConfirguration.ROUTING_QUEUE_1 "接收消息:" message);
}
@RabbitListener(queues = AmqpConfirguration.ROUTING_QUEUE_2)
public void routingReceive2(String message) {
System.out.println(AmqpConfirguration.ROUTING_QUEUE_2 "接收消息:" message);
}
/**
* 主题模式接收
*
* @param message
*/
@RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_1)
public void topicReceive1(String message) {
System.out.println(AmqpConfirguration.TOPIC_QUEUE_1 "接收消息:" message);
}
@RabbitListener(queues = AmqpConfirguration.TOPIC_QUEUE_2)
public void topicReceive2(String message) {
System.out.println(AmqpConfirguration.TOPIC_QUEUE_2 "接收消息:" message);
}
}消息消费者使用 @RabbitListener 注解监听消息。
测试
代码语言：javascript复制@RunWith(SpringRunner.class)
@SpringBootTest
public class AmqpTest {
@Autowired
private AmqpSender sender;
@Test
public void testSimpleSend() {
for (int i = 1; i < 6; i ) {
this.sender.simpleSend("test simpleSend " i);
}
}
@Test
public void testPsSend() {
for (int i = 1; i < 6; i ) {
this.sender.psSend("test psSend " i);
}
}
@Test
public void testRoutingSend() {
for (int i = 1; i < 6; i ) {
this.sender.routingSend("order", "test routingSend " i);
}
}
@Test
public void testTopicSend() {
for (int i = 1; i < 6; i ) {
this.sender.topicSend("user.add", "test topicSend " i);
}
}
}
