准备
1.新建springboot项目
2.添加Spring for Rabbitmq依赖
3.yml中配置rabbitmq信息:
spring:
rabbitmq:
host: 192.168.64.140
username: admin
password: admin每个模式创建一个包,包中包含主程序(main方法),生产者,消费者,每测试一个程序启动一个main方法即可.
简单模式
主程序
@SpringBootApplication
public class Main {
@Autowired
private Producer p;
public static void main(String[] args) {
SpringApplication.run(Main.class,args);
}
/*
springboot项目完整启动完成
当前对象中需要注入的对象注入完成之后才会执行该方法
*/ @PostConstruct
public void test(){
p.send();
}
/*
封装队列参数的对象
RabbitAutoConfiguraion 自动配置类,会自动发现Queue
在服务器上定义该队列
*/ @Bean
public Queue helloworldQueue(){
//return new Queue("helloworld");//true,false,false 持久,非独占,非自动删除
return new Queue("helloworld",false);
}
}生产者
@Component
public class Producer {
/*
AmqpTemplate的实例是在
RabbitmqAutoConfiguration自动配置类中创建的
*/ @Autowired
private AmqpTemplate t;
public void send(){
t.convertAndSend("helloworld", "Hello World!");
}
}消费者
@Component
public class Consumer {
/*
消费者自动注册,自动连接服务器,自动开启消息监听
*/ @RabbitListener(queues = "helloworld")
public void receive(String msg){
System.out.println("收到:"+msg);
}
}工作模式
工作模式还需要实现两点:
1.合理分发:
手动ack:springboot整合rabbitmq默认手动ack,并且自动返回回执
qos=1:yml中配置:spring.rabbitmq.listener.simple.prefetch:1
2.持久化:
队列持久化/消息持久化:springboot整合rabbitmq中都是默认持久的
若是不想要其持久 -->
- 使用 MessagePostProcessor 前置处理器参数
- 从消息中获取消息的属性对象
- 在属性中把 DeliveryMode 设置为非持久化
//如果需要设置消息为非持久化,可以取得消息的属性对象,修改它的deliveryMode属性
t.convertAndSend("task_queue", (Object) s, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
MessageProperties props = message.getMessageProperties();
props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
return message;
}
});主程序
@SpringBootApplication
public class Main {
@Autowired
private Producer p;
@PostConstruct
public void test(){
new Thread(new Runnable() {
@Override
public void run() {
p.send();
}
}).start();
//new Thread(() -> p.send()).start();
}
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
@Bean
public Queue taskQueue(){
return new Queue("task_queue");//true,false,false
}
}生产者
@Component
public class Producer {
@Autowired
private AmqpTemplate t;
public void send(){
while(true){
System.out.println("nn输入消息:");
String msg = new Scanner(System.in).nextLine();
t.convertAndSend("task_queue", (Object) msg, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
MessageProperties p = message.getMessageProperties();
p.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
return message;
}
});
}
}
}消费者
@Component
public class Consumer {
@RabbitListener(queues = "task_queue")
public void receive1(String msg){
System.out.println("消费者1收到:"+msg);
}
@RabbitListener(queues = "task_queue")
public void receive2(String msg){
System.out.println("消费者2收到:"+msg);
}
}发布订阅模式
主程序
创建 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.
spring boot 的自动配置类会自动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.
@SpringBootApplication
public class Main {
@Autowired
private Producer p;
@PostConstruct
public void test(){
new Thread(new Runnable() {
@Override
public void run() {
p.send();
}
}).start();
}
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
//定义fanout交换机:logs
@Bean
public FanoutExchange logs(){
//return new FanoutExchange("logs");//默认true,false
return new FanoutExchange("logs",false,false);
}
}生产者
生产者向指定的交换机 logs 发送数据.
不需要指定队列名或路由键, 即使指定也无效, 因为 fanout 交换机会向所有绑定的队列发送数据, 而不是有选择的发送
@Component
public class Producer {
@Autowired
private AmqpTemplate t;
public void send(){
while (true){
System.out.println("输入消息:");
String msg=new Scanner(System.in).nextLine();
t.convertAndSend("logs", "",msg);
}
}
}消费者
消费者需要执行以下操作:
- 定义随机队列(随机命名,非持久,排他,自动删除)
- 定义交换机(可以省略, 已在主程序中定义)
- 将队列绑定到交换机
spring boot 通过注解完成以上操作:(注意@RabbitListener注解的嵌套)
@Component
public class Consumer {
/*
1.随即队列
2.指定绑定的交换机
*/ @RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "logs",declare = "false")//declare指定的是一个已定义过的交换机,而不是重新定义
))
public void reveice1(String msg){
System.out.println("消费者1收到:"+msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "logs",declare = "false")
))
public void reveice2(String msg){
System.out.println("消费者2收到:"+msg);
}
}路由模式
与发布和订阅模式代码类似, 只是做以下三点调整:
- 使用
direct交换机 - 队列和交换机绑定时, 设置绑定键
- 发送消息时, 指定路由键
主程序
@SpringBootApplication
public class Main {
@Autowired
private Producer p;
@PostConstruct
public void test(){
new Thread(new Runnable() {
@Override
public void run() {
p.send();
}
}).start();
}
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
//定义direct交换机:logs
@Bean
public DirectExchange logs(){
//return new FanoutExchange("logs");//默认true,false
return new DirectExchange("direct_logs",false,false);
}
}生产者
@Component
public class Producer {
@Autowired
private AmqpTemplate t;
public void send(){
while (true){
System.out.println("输入消息:");
String msg=new Scanner(System.in).nextLine();
System.out.println("输入路由键:");
String key=new Scanner(System.in).nextLine();
t.convertAndSend("direct_logs", key,msg);
}
}
}消费者
@Component
public class Consumer {
/*
1.随即队列
2.指定绑定的交换机
*/ @RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "direct_logs",declare = "false"),//declare指定的是一个已定义过的交换机,而不是重新定义
key = "error"
))
public void reveice1(String msg){
System.out.println("消费者1收到:"+msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "direct_logs",declare = "false"),
key = {"info","warning","error"}
))
public void reveice2(String msg){
System.out.println("消费者2收到:"+msg);
}
}主题模式
主题模式不过是具有特殊规则的路由模式, 代码与路由模式基本相同, 只做如下调整:
- 使用
topic交换机 - 使用特殊的绑定键和路由键规则
主程序
{
@Autowired
private Producer p;
@PostConstruct
public void test(){
new Thread(new Runnable() {
@Override
public void run() {
p.send();
}
}).start();
}
public static void main(String[] args) {
SpringApplication.run(Main.class, args);
}
//定义topic交换机:logs
@Bean
public TopicExchange logs(){
//return new TopicExchange("logs");//默认true,false
return new TopicExchange("topic_logs",false,false);
}
}生产者
@Component
public class Producer {
@Autowired
private AmqpTemplate t;
public void send(){
while (true){
System.out.println("输入消息:");
String msg=new Scanner(System.in).nextLine();
System.out.println("输入路由键:");
String key=new Scanner(System.in).nextLine();
t.convertAndSend("topic_logs", key,msg);
}
}
}消费者
@Component
public class Consumer {
/*
1.随即队列
2.指定绑定的交换机
*/ @RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "topic_logs",declare = "false"),//declare指定的是一个已定义过的交换机,而不是重新定义
key = "*.orange.*"
))
public void reveice1(String msg){
System.out.println("消费者1收到:"+msg);
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue,//服务器自动取名,false,true,true
exchange = @Exchange(name = "topic_logs",declare = "false"),
key = {"*.*.rabbit","lazy.#"}
))
public void reveice2(String msg){
System.out.println("消费者2收到:"+msg);
}
}
java