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