Rabbit交换器

Direct交换器(发布器与订阅 完全匹配)

搭建环境

创建项目

rabbitmq-direct-consumer
rabbitmq-direct-provider

修改全局配置文件

修改 Consumer 的配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=log.direct
#info 队列名称
mq.config.queue.info=log.info
#info 路由键
mq.config.queue.info.routing.key=log.info.routing.key
#error 队列名称
mq.config.queue.error=log.error
#error 路由键
mq.config.queue.error.routing.key=log.error.routing.key

修改 Provider 的配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=log.direct
#info 路由键
mq.config.queue.info.routing.key=log.info.routing.key
#error 路由键
mq.config.queue.error.routing.key=log.error.routing.key
#error 队列名称
mq.config.queue.error=log.error
编写Consumer

InfoReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.info}",autoDelete="true"),
exchange=@Exchange(value="${mq.config.exchange}",type= ExchangeTypes.DIRECT),
key="${mq.config.queue.info.routing.key}"
)
)
public class InfoReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("Info........receiver: "+msg);
}
}

ErrorReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.error}",autoDelete="true"),
exchange=@Exchange(value="${mq.config.exchange}",type= ExchangeTypes.DIRECT),
key="${mq.config.queue.error.routing.key}"
)
)
public class ErrorReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("Error..........receiver: "+msg);
}
}
编写Provider
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
* 消息发送者
* @author Administrator
*
*/
@Component
public class Sender {
@Autowired
private AmqpTemplate rabbitAmqpTemplate;
//exchange 交换器名称
@Value("${mq.config.exchange}")
private String exchange;
//routingkey 路由键
@Value("${mq.config.queue.error.routing.key}")
private String routingkey;
/*
* 发送消息的方法
*/
public void send(String msg){
//向消息队列发送消息
//参数一:交换器名称。
//参数二:路由键
//参数三:消息
this.rabbitAmqpTemplate.convertAndSend(this.exchange,this.routingkey, msg);
}
}


Topic交换器(主题,规则匹配)

搭建环境

创建项目

rabbitmq-topic-consumer
rabbitmq-topic-provider

修改配置文件

Provider

1
2
3
4
5
6
7
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=log.topic

Consumer

1
2
3
4
5
6
7
8
9
10
11
12
13
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=log.topic
#info 队列名称
mq.config.queue.info=log.info
#error 队列名称
mq.config.queue.error=log.error
#log 队列名称
mq.config.queue.logs=log.all
编写Provider

UserSender

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 消息发送者
* @author Administrator
*
*/
@Component
public class UserSender {
@Autowired
private AmqpTemplate rabbitAmqpTemplate;
//exchange 交换器名称
@Value("${mq.config.exchange}")
private String exchange;
/*
* 发送消息的方法
*/
public void send(String msg){
//向消息队列发送消息
//参数一:交换器名称。
//参数二:路由键
//参数三:消息
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.
log.debug", "user.log.debug....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.
log.info", "user.log.info....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.
log.warn","user.log.warn....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.
log.error", "user.log.error....."+msg);
}
}

ProductSender

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 消息发送者
* @author Administrator
*
*/
@Component
public class ProductSender {
@Autowired
private AmqpTemplate rabbitAmqpTemplate;
//exchange 交换器名称
@Value("${mq.config.exchange}")
private String exchange;
/*
* 发送消息的方法
*/
public void send(String msg){
//向消息队列发送消息
//参数一:交换器名称。
//参数二:路由键
//参数三:消息
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"produ
ct.log.debug", "product.log.debug....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"produ
ct.log.info", "product.log.info....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"produ
ct.log.warn","product.log.warn....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"produ
ct.log.error", "product.log.error....."+msg);
}
}

OrderSender

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
/**
* 消息发送者
* @author Administrator
*
*/
@Component
public class OrderSender {
@Autowired
private AmqpTemplate rabbitAmqpTemplate;
//exchange 交换器名称
@Value("${mq.config.exchange}")
private String exchange;
/*
* 发送消息的方法
*/
public void send(String msg){
//向消息队列发送消息
//参数一:交换器名称。
//参数二:路由键
//参数三:消息
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order
.log.debug", "order.log.debug....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order
.log.info", "order.log.info....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order
.log.warn","order.log.warn....."+msg);
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order
.log.error", "order.log.error....."+msg);
}
}
编写Consumer

InfoReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.info}",autoDelete="tr
ue"),
exchange=@Exchange(value="${mq.config.exchange}",type=Excha
ngeTypes.TOPIC),
key="*.log.info"
)
)
public class InfoReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("......Info........receiver:
"+msg);
}
}

ErrorReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.error}",autoDelete="t
rue"),
exchange=@Exchange(value="${mq.config.exchange}",type=Excha
ngeTypes.TOPIC),
key="*.log.error"
)
)
public class ErrorReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("......Error........receiver:
"+msg);
}
}

LogsReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.logs}",autoDelete="tr
ue"),
exchange=@Exchange(value="${mq.config.exchange}",type=Excha
ngeTypes.TOPIC),
key="*.log.*"
)
)
public class LogsReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("......All........receiver:
"+msg);
}
}

测试代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@Autowired
private UserSender usersender;
@Autowired
private ProductSender productsender;
@Autowired
private OrderSender ordersender;
/*
* 测试消息队列
*/
@Test
public void test1(){
this.usersender.send("UserSender.....");
this.productsender.send("ProductSender....");
this.ordersender.send("OrderSender......");
}


Fanout交换器(广播)

搭建环境

创建项目

rabbitmq-fanout-consumer
rabbitmq-fanout-provider

修改配置文件

consumer

1
2
3
4
5
6
7
8
9
10
11
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=order.fanout
#短信服务队列名称
mq.config.queue.sms=order.sms
#push 服务队列名称
mq.config.queue.push=order.push

provider

1
2
3
4
5
6
7
spring.application.name=springcloud-mq
spring.rabbitmq.host=192.168.70.131
spring.rabbitmq.port=5672
spring.rabbitmq.username=oldlu
spring.rabbitmq.password=123456
#设置交换器的名称
mq.config.exchange=order.fanout
编写Consumer

SmsReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
* key:路由键
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.sms}",autoDelete="tru
e"),
exchange=@Exchange(value="${mq.config.exchange}",type=Excha
ngeTypes.FANOUT)
)
)
public class SmsReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("Sms........receiver: "+msg);
}
}

PushReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
/**
* 消息接收者
* @author Administrator
* @RabbitListener bindings:绑定队列
* @QueueBinding value:绑定队列的名称
* exchange:配置交换器
*
* @Queue value:配置队列名称
* autoDelete:是否是一个可删除的临时队列
*
* @Exchange value:为交换器起个名称
* type:指定具体的交换器类型
*/
@Component
@RabbitListener(
bindings=@QueueBinding(
value=@Queue(value="${mq.config.queue.push}",autoDelete="tr
ue"),
exchange=@Exchange(value="${mq.config.exchange}",type=Excha
ngeTypes.FANOUT)
)
)
public class PushReceiver {
/**
* 接收消息的方法。采用消息队列监听机制
* @param msg
*/
@RabbitHandler
public void process(String msg){
System.out.println("Push..........receiver: "+msg);
}
}
编写Provider
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
* 消息发送者
* @author Administrator
*
*/
@Component
public class Sender {
@Autowired
private AmqpTemplate rabbitAmqpTemplate;
//exchange 交换器名称
@Value("${mq.config.exchange}")
private String exchange;
/*
* 发送消息的方法
*/
public void send(String msg){
//向消息队列发送消息
//参数一:交换器名称。
//参数二:路由键
//参数三:消息
this.rabbitAmqpTemplate.convertAndSend(this.exchange,"",
msg);
}
}

RabbitMQ消息持久化

autoDelete 属性
@Queue: 当所有消费客户端连接断开后,是否自动删除
队列 true:删除 false:不删除
@Exchange:当所有绑定队列都不在使用时,是否自动
删除交换器 true:删除 false:不删除