rabbitMQ简介：
broker：消息队列服务器实体。
exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
queue：消息队列，每个消息都会被投入到一个或多个队列。
binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。
1、使用docker安装rabbitMQ
官方镜像：https://hub.docker.com/_/rabbitmq?tab=description&page=1&ordering=last_updated
//拉取镜像
docker pull rabbitmq:management-alpine
//运行
docker run \
--name rabbitmq\
--hostname hostmq \
-p 15672:15672 \
-p 5672:5672 \
-d \
rabbitmq:3-management
//默认账号密码都是：guest2、Springboot使用Rabbit
依赖
<!--AMQP依赖，包含RabbitMQ-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--json-->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</dependency>
配置文件（生产者）
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 192.168.150.101 # rabbitMQ的ip地址
port: 5672 # 端口
username: maosi
password: 123456
virtual-host: / #虚拟主机配置文件（消费者）
spring:
rabbitmq:
host: 192.168.150.101 # rabbitMQ的ip地址
port: 5672 # 端口
username: maosi
password: 123456
virtual-host: /
listener:
simple:
prefetch: 1 #一次拿一个请求，不会因为处理不同步导致的处理时间太长的问题生产者
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
//直接发送到单个队列
@Test
public void testSendMessage2SimpleQueue() {
String queueName = "simple.queue";
String message = "hello, spring amqp!";
rabbitTemplate.convertAndSend(queueName, message);
}
//直接发送到单个队列
@Test
public void testSendMessage2WorkQueue() throws InterruptedException {
String queueName = "simple.queue";
String message = "hello, message__";
for (int i = 1; i <= 50; i++) {
rabbitTemplate.convertAndSend(queueName, message + i);
Thread.sleep(20);
}
}
//通过交换机发送到所有队列
@Test
public void testSendFanoutExchange() {
// 交换机名称
String exchangeName = "itcast.fanout";
// 消息
String message = "hello, every one!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "", message);
}
//通过交换机发送到指定队列 red队列
@Test
public void testSendDirectExchange() {
// 交换机名称
String exchangeName = "itcast.direct";
// 消息
String message = "hello, red!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "red", message);
}
//通过交换机发送到主题队列 china.weather队列
@Test
public void testSendTopicExchange() {
// 交换机名称
String exchangeName = "itcast.topic";
// 消息
String message = "今天天气不错，我的心情好极了!";
// 发送消息
rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
}
}消费者：方式一（配置文件法）
@Configuration
public class FanoutConfig {
// itcast.fanout
@Bean
public FanoutExchange fanoutExchange(){
return new FanoutExchange("itcast.fanout");
}
// fanout.queue1
@Bean
public Queue fanoutQueue1(){
return new Queue("fanout.queue1");
}
// 绑定队列1到交换机
@Bean
public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue1)
.to(fanoutExchange);
}
// fanout.queue2
@Bean
public Queue fanoutQueue2(){
return new Queue("fanout.queue2");
}
// 绑定队列2到交换机
@Bean
public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
return BindingBuilder
.bind(fanoutQueue2)
.to(fanoutExchange);
}
@Bean
public Queue objectQueue(){
return new Queue("object.queue");
}
}消费者：方式二（注解法）@RabbitListener（）
@Component
public class SpringRabbitListener {
// @RabbitListener(queues = "simple.queue")
// public void listenSimpleQueue(String msg) {
// System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
// }
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
Thread.sleep(200);
}
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "china.#"
))
public void listenTopicQueue1(String msg){
System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
key = "#.news"
))
public void listenTopicQueue2(String msg){
System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
}
@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String,Object> msg){
System.out.println("接收到object.queue的消息：" + msg);
}
}json转换，因为队列消息是object类型，用jackson转换会显示正常
@Bean
public MessageConverter messageConverter(){
return new Jackson2JsonMessageConverter();
}
Comments | NOTHING